feat(replication): proxy GET/HEAD/Tagging for unreplicated objects to replication targets

Implements the MinIO active-active read-proxy protocol (P1-5 of the
replication compatibility review): when a GET/HEAD/GetObjectTagging/
PutObjectTagging/DeleteObjectTagging request fails locally with
not-found and the bucket has replication targets, the request is proxied
to the targets in rule order, mirroring bucket-replication.go
proxyGetToReplicationTarget/proxyHeadToRepTarget/proxyTaggingToRepTarget.

Protocol surface:
- Anti-loop: inbound {x-rustfs-,x-minio-}source-proxy-request is parsed
  into ObjectOptions (proxy_request + proxy_header_set, matching MinIO
  ProxyRequest/ProxyHeaderSet); a request carrying the marker with ANY
  value is never re-proxied. Outbound client proxy calls send the marker
  as "true"; replication worker convergence HEADs send it as "false" so
  a peer's proxy layer cannot answer a convergence check by proxying
  back to the source (which would fake Completed without a PUT).
- Target selection: new replication_proxy.rs get_proxy_targets — empty
  when the marker is set, versioning is suspended, or no replication
  config; otherwise filter_target_arns -> TargetClient lookup, skipping
  targets with proxying disabled.
- TargetClient gains head_object_for_proxy/get_object (streaming) and
  the three tagging calls. Proxy calls never send the replication-check
  SSE-C exemption header; customer SSE-C keys are forwarded verbatim so
  the target performs real decryption. Conditional (If-*) headers are
  not forwarded (MinIO parity); Range and part_number are, with
  parts_count/tag_count/storage_class/expiration passed through.
- Metrics: proxy counters now count only real client proxy traffic,
  MinIO-aligned (one total per proxied request, one failed when no
  target served it). The previous misattributed counters — replication
  worker HEAD/PUT (#2672) and local tagging operations (#2682) — are
  removed; ReplProxyMetric now maps the tagging counters instead of
  dropping them.

e2e (fake_s3_target extended with tagging + header journaling): proxied
GET body + outbound header contract (marker present, no
replication-check, SSE-C passthrough), HEAD, anti-loop 404 with zero
outbound requests, GetObjectTagging, and metric mapping unit tests.

Rolling note: proxying only activates for buckets with replication
targets; requests carrying the marker keep pre-upgrade behavior.

Refs rustfs/backlog#1675 (P1-5)

Cargo.lock

@@ -9825,19 +9825,14 @@ name = "rustfs-madmin"
 version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
- "http 1.5.0",
  "humantime",
  "hyper",
  "jiff",
- "reqwest",
  "rmp-serde",
- "rustfs-signer",
- "s3s",
  "serde",
  "serde_json",
  "sysinfo",
  "time",
- "tokio",
 ]
 
 [[package]]

_typos.toml

@@ -40,7 +40,6 @@ mak = "mak"
 gae = "gae"
 GAE = "GAE"
 thr = "thr"
-mis = "mis"
 # s3-tests original test names (cannot be changed)
 nonexisted = "nonexisted"
 consts = "consts"

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, 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, 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.
 
-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.
+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.
 
 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,10 +30,12 @@ use s3s::access::{S3Access, S3AccessContext};
 use s3s::auth::SimpleAuth;
 use s3s::dto::{
     AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput,
-    CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput, ETag,
-    GetBucketVersioningInput, GetBucketVersioningOutput, GetObjectInput, GetObjectOutput, HeadBucketInput, HeadBucketOutput,
+    CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput,
+    DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput, GetBucketVersioningOutput,
+    GetObjectInput, GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput,
     HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ObjectVersionId, PutObjectInput,
-    PutObjectOutput, StreamingBlob, Timestamp, TimestampFormat, UploadPartInput, UploadPartOutput,
+    PutObjectOutput, PutObjectTaggingInput, PutObjectTaggingOutput, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat,
+    UploadPartInput, UploadPartOutput,
 };
 use s3s::service::{S3Service, S3ServiceBuilder};
 use s3s::validation::{AwsNameValidation, NameValidation};
@@ -103,6 +105,9 @@ pub enum Operation {
     GetObject,
     HeadObject,
     DeleteObject,
+    GetObjectTagging,
+    PutObjectTagging,
+    DeleteObjectTagging,
     ListObjectVersions,
     CreateMultipartUpload,
     UploadPart,
@@ -149,6 +154,35 @@ 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 {
@@ -163,6 +197,7 @@ 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>,
 }
 
@@ -199,6 +234,9 @@ 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)]
@@ -569,6 +607,7 @@ 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,
@@ -576,6 +615,7 @@ impl S3Access for FaultAccess {
             parsed,
             content_length,
             replication_timestamps,
+            proxy_headers,
         );
         if let Some(RequestFault {
             action: FaultAction::Status(status),
@@ -615,6 +655,9 @@ 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,
@@ -630,6 +673,7 @@ 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
@@ -655,6 +699,7 @@ fn record_request(
         content_length,
         consumed_bytes: None,
         replication_timestamps,
+        proxy_headers,
         fault: action.clone(),
     });
     action.map(|action| RequestFault { sequence, action })
@@ -721,6 +766,15 @@ 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,
@@ -1135,6 +1189,33 @@ 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>> {
@@ -1248,6 +1329,7 @@ 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(
@@ -1305,6 +1387,72 @@ 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?;
@@ -1381,6 +1529,7 @@ impl S3 for FakeBackend {
                 delete_marker: true,
                 content_type: None,
                 metadata: None,
+                tags: Vec::new(),
             },
         )?;
         Ok(apply_response_fault(
@@ -1583,6 +1732,7 @@ 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
@@ -3074,6 +3224,7 @@ mod tests {
                     },
                     Some(0),
                     ReplicationTimestampHeaders::default(),
+                    ProxyHeaderSnapshot::default(),
                 );
             }
             let records = lock(&control).requests.clone();
@@ -3096,6 +3247,7 @@ mod tests {
                 },
                 None,
                 ReplicationTimestampHeaders::default(),
+                ProxyHeaderSnapshot::default(),
             );
             {
                 let bounded_records = lock(&bounded_control);

crates/e2e_test/src/replication_extension_test.rs

@@ -8417,3 +8417,304 @@ 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,12 +198,13 @@ 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, 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, 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,
         };
     }
 
@@ -373,8 +374,8 @@ pub mod error {
 
 pub mod erasure {
     pub use crate::erasure::coding::{
-        BitrotReader, BitrotSelfTestError, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError,
-        ReedSolomonEncoder, bitrot_self_test, calc_shard_size, calc_shard_size_legacy,
+        BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError, ReedSolomonEncoder,
+        calc_shard_size, calc_shard_size_legacy,
     };
 }
 

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -27,10 +27,15 @@ 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,
 };
@@ -57,8 +62,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_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_PROXY_REQUEST,
+    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,
 };
@@ -1446,6 +1451,43 @@ 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
@@ -1851,6 +1893,13 @@ 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()
@@ -1875,6 +1924,129 @@ impl TargetClient {
         }
     }
 
+    /// HEAD used by the read-proxy path (GET/HEAD of an object not yet
+    /// replicated locally, MinIO `proxyHeadToRepTarget`).
+    ///
+    /// Deliberately different from [`TargetClient::head_object`]: it must NOT
+    /// send `source-replication-check` — that header is the replication
+    /// worker's SSE-C metadata exemption channel. A proxied client request
+    /// instead forwards the client's own SSE-C headers (`extra_headers`) so
+    /// the target performs the real SSE-C validation/decryption. The
+    /// `source-proxy-request` marker is always added so the target does not
+    /// proxy the request onward (anti-loop).
+    pub async fn head_object_for_proxy(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<String>,
+        range: Option<String>,
+        part_number: Option<i32>,
+        extra_headers: HeaderMap,
+    ) -> Result<HeadObjectOutput, SdkError<HeadObjectError>> {
+        let headers = proxy_outbound_headers(extra_headers);
+        self.client
+            .head_object()
+            .bucket(bucket)
+            .key(object)
+            .set_version_id(resolve_read_api_version_id(version_id))
+            .set_range(range)
+            .set_part_number(part_number)
+            .customize()
+            .map_request(move |req| apply_extra_headers(req, &headers))
+            .send()
+            .await
+    }
+
+    /// GET used by the read-proxy path (MinIO `proxyGetToReplicationTarget`).
+    /// Returns the streaming SDK output; callers must forward the body without
+    /// buffering it. Same header contract as [`Self::head_object_for_proxy`]:
+    /// anti-loop marker on, replication-check never sent, client SSE-C /
+    /// conditional headers forwarded verbatim via `extra_headers`.
+    pub async fn get_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<String>,
+        range: Option<String>,
+        part_number: Option<i32>,
+        extra_headers: HeaderMap,
+    ) -> Result<GetObjectOutput, SdkError<GetObjectError>> {
+        let headers = proxy_outbound_headers(extra_headers);
+        self.client
+            .get_object()
+            .bucket(bucket)
+            .key(object)
+            .set_version_id(resolve_read_api_version_id(version_id))
+            .set_range(range)
+            .set_part_number(part_number)
+            .customize()
+            .map_request(move |req| apply_extra_headers(req, &headers))
+            .send()
+            .await
+    }
+
+    /// GetObjectTagging for the tagging read-proxy path
+    /// (MinIO `proxyGetTaggingToRepTarget`). Anti-loop marker always added.
+    pub async fn get_object_tagging(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<String>,
+    ) -> Result<GetObjectTaggingOutput, SdkError<GetObjectTaggingError>> {
+        let headers = proxy_outbound_headers(HeaderMap::new());
+        self.client
+            .get_object_tagging()
+            .bucket(bucket)
+            .key(object)
+            .set_version_id(resolve_read_api_version_id(version_id))
+            .customize()
+            .map_request(move |req| apply_extra_headers(req, &headers))
+            .send()
+            .await
+    }
+
+    /// PutObjectTagging for the tagging proxy path
+    /// (MinIO `proxyTaggingToRepTarget`). Anti-loop marker always added.
+    pub async fn put_object_tagging(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<String>,
+        tagging: SdkTagging,
+    ) -> Result<PutObjectTaggingOutput, SdkError<PutObjectTaggingError>> {
+        let headers = proxy_outbound_headers(HeaderMap::new());
+        self.client
+            .put_object_tagging()
+            .bucket(bucket)
+            .key(object)
+            .set_version_id(resolve_read_api_version_id(version_id))
+            .tagging(tagging)
+            .customize()
+            .map_request(move |req| apply_extra_headers(req, &headers))
+            .send()
+            .await
+    }
+
+    /// DeleteObjectTagging for the tagging proxy path
+    /// (MinIO `proxyTaggingToRepTarget`). Anti-loop marker always added.
+    pub async fn delete_object_tagging(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<String>,
+    ) -> Result<DeleteObjectTaggingOutput, SdkError<DeleteObjectTaggingError>> {
+        let headers = proxy_outbound_headers(HeaderMap::new());
+        self.client
+            .delete_object_tagging()
+            .bucket(bucket)
+            .key(object)
+            .set_version_id(resolve_read_api_version_id(version_id))
+            .customize()
+            .map_request(move |req| apply_extra_headers(req, &headers))
+            .send()
+            .await
+    }
+
     /// On success returns the version id the target assigned (from
     /// `x-amz-version-id`), letting callers audit the version-identity
     /// contract — a target that adopts the source version echoes it back.

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

@@ -14,6 +14,7 @@ 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,6 +29,7 @@ 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;
@@ -74,6 +75,7 @@ pub use replication_pool::{
     get_global_replication_pool, get_global_replication_stats, init_background_replication, persist_force_delete_intent,
     read_durable_mrf_backlog, resync_start_conflict_id,
 };
+pub use replication_proxy::get_proxy_targets;
 pub use replication_queue_boundary::{
     DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation,
     ReplicationPriority, ReplicationQueueAdmission,

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

@@ -667,6 +667,368 @@ 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 {
@@ -1221,71 +1583,12 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         let storage = self.storage.clone();
 
         let handle = tokio::spawn(async move {
-            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(recovery_guard) = acquire_mrf_recovery_guard(&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;
-                }
+            let Some(entries) = load_mrf_recovery_entries(&storage).await else {
+                return;
             };
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
 
@@ -1294,187 +1597,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let mut retry_entries = Vec::new();
 
             for entry in entries.iter() {
-                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
-                            }
-                        }
-                    }
+                let Some(admission) = replay_mrf_entry(entry, &storage, &mut retry_entries).await else {
+                    continue;
                 };
 
                 if admission == ReplicationQueueAdmission::Missed {
@@ -1484,29 +1608,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                 }
             }
 
-            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 = resolve_retained_mrf_entries(&storage, &recovery_guard, &entries, &retry_entries).await;
             let retained_count = retained.len();
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
 

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

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

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

@@ -1161,6 +1161,31 @@ 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/erasure/coding/bitrot.rs

@@ -820,263 +820,10 @@ impl BitrotWriterWrapper {
     }
 }
 
-// --- startup bitrot self-test (rustfs/backlog#1873, MinIO bitrotSelfTest parity) ---
-//
-// A broken hash implementation (bad SIMD feature combination, platform drift, a
-// key-handling regression) fails silently: every shard reads back "corrupt",
-// heal rewrites data that was fine, and cross-platform clusters disagree about
-// which copy is healthy. The self-test below pins the algorithms the moment a
-// process starts, so a drifted build announces itself instead of quietly
-// rewriting objects. See docs/rustfs-heal-scanner-vs-minio-comprehensive-
-// analysis-2026-08-16.md §6 HS-11.
-
-/// Length of the deterministic self-test payload.
-pub const BITROT_SELF_TEST_PAYLOAD_LEN: usize = 4096;
-
-/// Known-answer digest of [`bitrot_self_test_payload`] under `HighwayHash256S`
-/// (the production default). Pinned so any platform or build where the
-/// implementation drifts fails startup instead of miss-hashing shards.
-const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S: [u8; 32] = [
-    0xb9, 0x32, 0xa2, 0xaa, 0x4a, 0xb7, 0x33, 0x6a, 0xa3, 0xca, 0x7e, 0x61, 0x9d, 0x86, 0x52, 0x14, 0x6e, 0x7f, 0xd8, 0x9e, 0xea,
-    0x08, 0xd9, 0x8c, 0x33, 0x85, 0x87, 0x19, 0x30, 0xd6, 0xed, 0x06,
-];
-
-/// Known-answer digest of the same payload under `HighwayHash256SLegacy`.
-const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY: [u8; 32] = [
-    0x98, 0x24, 0x71, 0x4f, 0x16, 0xbb, 0x48, 0x39, 0xed, 0x68, 0xfa, 0x63, 0x5e, 0xd9, 0x07, 0x61, 0xdf, 0x0a, 0xff, 0xcf, 0x7d,
-    0x8c, 0xa8, 0xc7, 0xc0, 0xb6, 0x6f, 0x05, 0xdb, 0xda, 0x5a, 0x22,
-];
-
-/// FIPS 180-2 test vector: SHA-256 of the ASCII string "abc". Unlike the
-/// Highway digests above this one is externally verifiable, so it guards the
-/// whole `HashAlgorithm` plumbing even for readers who distrust pinned
-/// self-computed constants.
-const BITROT_SELF_TEST_KAT_SHA256_ABC: [u8; 32] = [
-    0xba, 0x78, 0x16, 0xbf, 0x8f, 0x01, 0xcf, 0xea, 0x41, 0x41, 0x40, 0xde, 0x5d, 0xae, 0x22, 0x23, 0xb0, 0x03, 0x61, 0xa3, 0x96,
-    0x17, 0x7a, 0x9c, 0xb4, 0x10, 0xff, 0x61, 0xf2, 0x00, 0x15, 0xad,
-];
-
-/// Deterministic self-test payload: xorshift64* from a fixed seed, so every
-/// platform and every run hashes the same 4096 bytes.
-fn bitrot_self_test_payload() -> [u8; BITROT_SELF_TEST_PAYLOAD_LEN] {
-    let mut state = 0x9E37_79B9_7F4A_7C15u64;
-    let mut payload = [0u8; BITROT_SELF_TEST_PAYLOAD_LEN];
-    for byte in payload.iter_mut() {
-        state ^= state >> 12;
-        state ^= state << 25;
-        state ^= state >> 27;
-        *byte = state.wrapping_mul(0x2545_F491_4F6C_DD1D) as u8;
-    }
-    payload
-}
-
-/// Why a bitrot self-test failed.
-#[derive(Debug)]
-pub enum BitrotSelfTestError {
-    /// A known-answer digest mismatched the pinned constant.
-    KnownAnswerMismatch {
-        algorithm: &'static str,
-        got: String,
-        want: String,
-    },
-    /// A freshly encoded shard failed `bitrot_verify`.
-    RoundtripVerify { algorithm: &'static str, detail: String },
-    /// A verified roundtrip read back different bytes than were written.
-    RoundtripReadback { algorithm: &'static str },
-    /// A deliberately tampered shard was not rejected by `bitrot_verify`.
-    TamperNotRejected {
-        algorithm: &'static str,
-        tampered: &'static str,
-    },
-}
-
-impl std::fmt::Display for BitrotSelfTestError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::KnownAnswerMismatch { algorithm, got, want } => {
-                write!(f, "known-answer mismatch for {algorithm}: got {got}, want {want}")
-            }
-            Self::RoundtripVerify { algorithm, detail } => write!(f, "{algorithm} roundtrip shard failed verification: {detail}"),
-            Self::RoundtripReadback { algorithm } => write!(f, "{algorithm} roundtrip read back different bytes"),
-            Self::TamperNotRejected { algorithm, tampered } => {
-                write!(f, "{algorithm} tampered shard ({tampered}) was not rejected")
-            }
-        }
-    }
-}
-
-impl std::error::Error for BitrotSelfTestError {}
-
-fn self_test_hex(bytes: &[u8]) -> String {
-    rustfs_utils::hex(bytes)
-}
-
-// (kept as a named one-liner so every KAT failure site reads the same; the
-// underlying formatter is the shared `rustfs_utils::hex`)
-
-/// Compare a digest against its pinned constant. Split out so a test can drive
-/// it with a wrong constant and prove the mismatch path fires.
-fn bitrot_kat_check(
-    algorithm: &'static str,
-    algo: &HashAlgorithm,
-    payload: &[u8],
-    expected: &[u8; 32],
-) -> Result<(), BitrotSelfTestError> {
-    let digest = algo.hash_encode(payload);
-    let digest = digest.as_ref();
-    if digest.len() != expected.len() || digest != expected.as_slice() {
-        return Err(BitrotSelfTestError::KnownAnswerMismatch {
-            algorithm,
-            got: self_test_hex(digest),
-            want: self_test_hex(expected),
-        });
-    }
-    Ok(())
-}
-
-/// Encode `payload` with `shard_size` blocks, verify it end to end, and read
-/// every block back through `BitrotReader` comparing bytes.
-async fn bitrot_roundtrip_check(
-    algorithm: &'static str,
-    algo: HashAlgorithm,
-    payload: &[u8],
-    shard_size: usize,
-) -> Result<(), BitrotSelfTestError> {
-    let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
-    for chunk in payload.chunks(shard_size) {
-        writer
-            .write(chunk)
-            .await
-            .map_err(|err| BitrotSelfTestError::RoundtripVerify {
-                algorithm,
-                detail: format!("encode failed: {err}"),
-            })?;
-    }
-    let encoded = writer.into_inner().into_inner();
-
-    let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
-    if encoded.len() != on_disk {
-        return Err(BitrotSelfTestError::RoundtripVerify {
-            algorithm,
-            detail: format!("encoded {} bytes, size formula says {on_disk}", encoded.len()),
-        });
-    }
-    bitrot_verify(std::io::Cursor::new(encoded.clone()), on_disk, payload.len(), algo.clone(), shard_size)
-        .await
-        .map_err(|err| BitrotSelfTestError::RoundtripVerify {
-            algorithm,
-            detail: err.to_string(),
-        })?;
-
-    let mut reader = BitrotReader::new(std::io::Cursor::new(encoded), shard_size, algo, false);
-    let mut offset = 0usize;
-    while offset < payload.len() {
-        let want = shard_size.min(payload.len() - offset);
-        let mut buf = vec![0u8; want];
-        let read = reader
-            .read(&mut buf)
-            .await
-            .map_err(|err| BitrotSelfTestError::RoundtripVerify {
-                algorithm,
-                detail: format!("read back failed at offset {offset}: {err}"),
-            })?;
-        if read != want || buf[..read] != payload[offset..offset + read] {
-            return Err(BitrotSelfTestError::RoundtripReadback { algorithm });
-        }
-        offset += read;
-    }
-    Ok(())
-}
-
-/// Flip one byte and require `bitrot_verify` to reject the result.
-async fn bitrot_tamper_check(
-    algorithm: &'static str,
-    algo: HashAlgorithm,
-    payload: &[u8],
-    shard_size: usize,
-    tampered: &'static str,
-    flip_at: usize,
-) -> Result<(), BitrotSelfTestError> {
-    let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
-    for chunk in payload.chunks(shard_size) {
-        writer.write(chunk).await.expect("self-test encode should not fail");
-    }
-    let mut corrupt = writer.into_inner().into_inner();
-    let flip_index = flip_at % corrupt.len();
-    corrupt[flip_index] ^= 0x80;
-
-    let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
-    match bitrot_verify(std::io::Cursor::new(corrupt), on_disk, payload.len(), algo, shard_size).await {
-        // The flipped byte must be rejected as a hash mismatch specifically, not
-        // by any incidental read error: an in-memory cursor cannot fail reads,
-        // so accepting any other failure here would mask a verify path that
-        // errors out before it ever compares hashes.
-        Err(err) if err.to_string().contains("hash mismatch") => Ok(()),
-        Ok(()) => Err(BitrotSelfTestError::TamperNotRejected { algorithm, tampered }),
-        Err(err) => Err(BitrotSelfTestError::RoundtripVerify {
-            algorithm,
-            detail: format!("tampered shard rejected with an unexpected error: {err}"),
-        }),
-    }
-}
-
-/// Verify every bitrot algorithm this crate can write or verify in production:
-/// both streaming Highway variants roundtrip end to end (encode → size formula
-/// → `bitrot_verify` → read back) and reject a flipped byte in both the data
-/// and the leading hash, while all three hashed algorithms reproduce their
-/// pinned known-answer digests.
-///
-/// Runs in well under a millisecond on 4 KiB of data; callers may run it inline
-/// at startup. Pure CPU, no allocation beyond a few KiB of scratch.
-pub async fn bitrot_self_test() -> Result<(), BitrotSelfTestError> {
-    let payload = bitrot_self_test_payload();
-
-    // Externally verifiable vector first: it guards the HashAlgorithm plumbing
-    // itself, before any self-pinned constants are consulted.
-    let abc = HashAlgorithm::SHA256.hash_encode(b"abc");
-    if abc.as_ref() != BITROT_SELF_TEST_KAT_SHA256_ABC.as_slice() {
-        return Err(BitrotSelfTestError::KnownAnswerMismatch {
-            algorithm: "SHA256",
-            got: self_test_hex(abc.as_ref()),
-            want: self_test_hex(&BITROT_SELF_TEST_KAT_SHA256_ABC),
-        });
-    }
-
-    bitrot_kat_check(
-        "HighwayHash256S",
-        &HashAlgorithm::HighwayHash256S,
-        &payload,
-        &BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S,
-    )?;
-    bitrot_kat_check(
-        "HighwayHash256SLegacy",
-        &HashAlgorithm::HighwayHash256SLegacy,
-        &payload,
-        &BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY,
-    )?;
-
-    for (algorithm, algo) in [
-        ("HighwayHash256S", HashAlgorithm::HighwayHash256S),
-        ("HighwayHash256SLegacy", HashAlgorithm::HighwayHash256SLegacy),
-    ] {
-        // Full blocks plus a partial tail, exactly like a real part stripe.
-        let tail_len = 2 * 1024 + 333;
-        bitrot_roundtrip_check(algorithm, algo.clone(), &payload, 1024).await?;
-        bitrot_roundtrip_check(algorithm, algo.clone(), &payload[..tail_len], 1024).await?;
-        // One flipped byte in the final data block, one in the first leading
-        // hash: both must fail verification.
-        bitrot_tamper_check(algorithm, algo.clone(), &payload, 1024, "final data byte", payload.len() - 1).await?;
-        bitrot_tamper_check(algorithm, algo, &payload, 1024, "leading hash byte", 0).await?;
-    }
-
-    Ok(())
-}
-
 #[cfg(test)]
 mod tests {
     use super::{
-        BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_kat_check, bitrot_self_test,
-        bitrot_self_test_payload, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
+        BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
     };
     use super::{MAX_RETAINED_CHUNKS_PER_BLOCK, ShardChunkRead, ShardSource};
     use bytes::Bytes;
@@ -1343,32 +1090,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn bitrot_self_test_payload_is_deterministic() {
-        // Two independent builds of the payload must agree byte for byte, or
-        // the pinned known-answer digests below would be meaningless.
-        assert_eq!(bitrot_self_test_payload(), bitrot_self_test_payload());
-    }
-
-    #[test]
-    fn bitrot_self_test_rejects_a_wrong_known_answer_digest() {
-        let payload = bitrot_self_test_payload();
-        let wrong = [0u8; 32];
-        let err = bitrot_kat_check("HighwayHash256S", &HashAlgorithm::HighwayHash256S, &payload, &wrong)
-            .expect_err("a zeroed digest must never match");
-        match err {
-            super::BitrotSelfTestError::KnownAnswerMismatch { algorithm, .. } => assert_eq!(algorithm, "HighwayHash256S"),
-            other => panic!("expected KnownAnswerMismatch, got {other:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn bitrot_self_test_passes() {
-        bitrot_self_test()
-            .await
-            .expect("the pinned digests and roundtrip checks must all pass on this platform");
-    }
-
     #[tokio::test]
     async fn vectored_test_writers_cover_fallback_flush_and_shutdown_paths() {
         let mut counting = VectoredCountingWriter::default();
@@ -1468,7 +1189,7 @@ mod tests {
         let last = corrupt.len() - 1;
         corrupt[last] ^= 0x80;
         let err = bitrot_verify(
-            std::io::Cursor::new(corrupt),
+            Cursor::new(corrupt),
             super::bitrot_shard_file_size(data.len(), shard_size, algo.clone()),
             data.len(),
             algo,
@@ -1561,7 +1282,7 @@ mod tests {
 
     #[tokio::test]
     async fn bitrot_reader_rejects_output_buffers_larger_than_shard_size() {
-        let mut reader = BitrotReader::new(std::io::Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
+        let mut reader = BitrotReader::new(Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
         let mut out = [0u8; 5];
         let err = reader
             .read(&mut out)
@@ -1686,7 +1407,7 @@ mod tests {
             (HashAlgorithm::HighwayHash256, true),
         ] {
             let label = format!("{algo:?}");
-            let writer = std::io::Cursor::new(Vec::<u8>::new());
+            let writer = Cursor::new(Vec::<u8>::new());
             let mut w = BitrotWriter::new(writer, shard_size, algo.clone());
             w.write(&[7u8; 16]).await.unwrap();
             let written = w.into_inner().into_inner();
@@ -1771,7 +1492,7 @@ mod tests {
     }
 
     async fn encode_one_block(payload: &[u8], shard_size: usize, algo: HashAlgorithm) -> Vec<u8> {
-        let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo);
+        let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo);
         w.write(payload).await.unwrap();
         w.into_inner().into_inner()
     }
@@ -1879,7 +1600,7 @@ mod tests {
         for algo in [HashAlgorithm::HighwayHash256S, HashAlgorithm::HighwayHash256SLegacy] {
             for &size in &[1usize, 16, 17, 32, 40, 48] {
                 let payload: Vec<u8> = (0..size).map(|i| i as u8).collect();
-                let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
+                let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
                 for chunk in payload.chunks(shard_size) {
                     w.write(chunk).await.unwrap();
                 }
@@ -1953,14 +1674,14 @@ mod tests {
             w.write(&data).await.expect("write shard");
 
             let mut via_read = vec![0u8; SHARD];
-            let n1 = BitrotReader::new(std::io::Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
+            let n1 = BitrotReader::new(Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
                 .read(&mut via_read)
                 .await
                 .expect("read");
 
             // A buffer with only capacity — no initialized bytes at all.
             let mut via_append: Vec<u8> = Vec::with_capacity(SHARD);
-            let n2 = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
+            let n2 = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
                 .read_appending(&mut via_append, SHARD)
                 .await
                 .expect("read_appending");
@@ -1985,7 +1706,7 @@ mod tests {
             encoded.truncate(encoded.len() - 1);
 
             let mut out: Vec<u8> = Vec::with_capacity(SHARD);
-            let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
+            let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
                 .read_appending(&mut out, SHARD)
                 .await
                 .expect_err("a truncated shard must not succeed");
@@ -2011,7 +1732,7 @@ mod tests {
         encoded[last] ^= 0xff;
 
         let mut out: Vec<u8> = Vec::with_capacity(SHARD);
-        let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
+        let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
             .read_appending(&mut out, SHARD)
             .await
             .expect_err("a corrupt shard must not verify");
@@ -2123,7 +1844,7 @@ mod tests {
             "Cursor<Bytes> must be able to hand out a block, otherwise the fast path is dead code"
         );
         assert_eq!(mem.position(), 8, "taking a block must advance like a read of the same length");
-        let mut streamed = std::io::Cursor::new(encoded.clone());
+        let mut streamed = Cursor::new(encoded.clone());
         assert!(
             ShardSource::try_take_block(&mut streamed, 8).is_none(),
             "a non-Bytes source must stay on the streaming path"
@@ -2151,7 +1872,7 @@ mod tests {
         );
 
         let mut via_stream: Vec<u8> = Vec::with_capacity(SHARD);
-        BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
+        BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
             .read_appending(&mut via_stream, SHARD)
             .await
             .expect("streaming read");

crates/ecstore/src/object_api/types.rs

@@ -277,6 +277,20 @@ 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/madmin/Cargo.toml

@@ -37,11 +37,7 @@ hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 [dependencies]
 hotpath.workspace = true
 humantime.workspace = true
-http.workspace = true
 hyper = { workspace = true, features = ["http2", "http1", "server"] }
-reqwest = { workspace = true, features = ["json"] }
-rustfs-signer.workspace = true
-s3s.workspace = true
 jiff = { workspace = true, features = ["serde"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["raw_value"] }
@@ -53,4 +49,3 @@ doctest = false
 
 [dev-dependencies]
 rmp-serde.workspace = true
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net"] }

crates/madmin/src/client.rs

@@ -1,851 +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.
-
-//! Admin API HTTP client for heal and scanner management (rustfs/backlog#1869).
-//!
-//! [`AdminClient`] speaks the `/rustfs/admin/v3` surface with S3 SigV4
-//! request signing (the same scheme the server's admin router authenticates),
-//! so `mc`-style tooling and automation can drive heal start/query/cancel and
-//! read background-heal / scanner status without hand-rolling HTTP.
-//!
-//! Wire structs in this module mirror the server-side shapes
-//! (`rustfs/src/admin/handlers/heal.rs`, `handlers/scanner.rs`,
-//! `rustfs-common/src/heal_channel.rs`), following the madmin-go model where
-//! the SDK owns its own copies and round-trip tests pin the encoding. Deeply
-//! nested status payloads that the server composes from runtime types are
-//! carried through as `serde_json::Value` and flattened maps rather than
-//! duplicated field-for-field, so the client cannot silently drift on fields
-//! it never interprets.
-
-use crate::heal_commands::HealResultItem;
-use http::Method;
-use serde::{Deserialize, Serialize, de};
-use std::time::Duration;
-
-/// Default admin API path prefix on a RustFS endpoint.
-pub const DEFAULT_ADMIN_API_PREFIX: &str = "/rustfs/admin";
-/// Default SigV4 region when the server has no explicit region configured.
-pub const DEFAULT_REGION: &str = "us-east-1";
-
-/// Scan mode for a heal request, mirroring the server's numeric-or-name wire
-/// encoding (`0` unknown/default, `1` normal, `2` deep).
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
-pub enum HealScanMode {
-    /// Server default; behaves as [`HealScanMode::Normal`].
-    #[default]
-    Unknown,
-    /// Metadata-level checks only.
-    Normal,
-    /// Full bitrot verification while healing.
-    Deep,
-}
-
-impl HealScanMode {
-    fn wire_number(self) -> u8 {
-        match self {
-            Self::Unknown => 0,
-            Self::Normal => 1,
-            Self::Deep => 2,
-        }
-    }
-
-    fn from_wire_number(value: u8) -> Option<Self> {
-        match value {
-            0 => Some(Self::Unknown),
-            1 => Some(Self::Normal),
-            2 => Some(Self::Deep),
-            _ => None,
-        }
-    }
-
-    fn from_wire_name(value: &str) -> Option<Self> {
-        match value {
-            "unknown" => Some(Self::Unknown),
-            "normal" => Some(Self::Normal),
-            "deep" => Some(Self::Deep),
-            _ => None,
-        }
-    }
-}
-
-impl Serialize for HealScanMode {
-    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        serializer.serialize_u8(self.wire_number())
-    }
-}
-
-impl<'de> Deserialize<'de> for HealScanMode {
-    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
-        struct HealScanModeVisitor;
-
-        impl de::Visitor<'_> for HealScanModeVisitor {
-            type Value = HealScanMode;
-
-            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-                formatter.write_str("a heal scan mode number or name")
-            }
-
-            fn visit_u64<E: de::Error>(self, value: u64) -> Result<Self::Value, E> {
-                u8::try_from(value)
-                    .ok()
-                    .and_then(HealScanMode::from_wire_number)
-                    .ok_or_else(|| E::custom(format!("unknown heal scan mode number: {value}")))
-            }
-
-            fn visit_str<E: de::Error>(self, value: &str) -> Result<Self::Value, E> {
-                HealScanMode::from_wire_name(value).ok_or_else(|| E::custom(format!("unknown heal scan mode name: {value}")))
-            }
-        }
-
-        deserializer.deserialize_any(HealScanModeVisitor)
-    }
-}
-
-/// Heal options for an admin heal request (mirror of the server body type).
-/// Fields default on decode: a client should tolerate a server response whose
-/// settings object omits fields it never set.
-#[derive(Debug, Clone, Default, Serialize, Deserialize)]
-pub struct HealOpts {
-    #[serde(default)]
-    pub recursive: bool,
-    #[serde(rename = "dryRun", default)]
-    pub dry_run: bool,
-    #[serde(default)]
-    pub remove: bool,
-    #[serde(default)]
-    pub recreate: bool,
-    #[serde(rename = "scanMode", default)]
-    pub scan_mode: HealScanMode,
-    #[serde(rename = "updateParity", default)]
-    pub update_parity: bool,
-    #[serde(rename = "nolock", default)]
-    pub no_lock: bool,
-    #[serde(rename = "pool", default)]
-    pub pool: Option<usize>,
-    #[serde(rename = "set", default)]
-    pub set: Option<usize>,
-}
-
-/// Successful heal start / path-scoped cancel response.
-#[derive(Debug, Clone, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct HealStartSuccess {
-    pub client_token: String,
-    pub client_address: String,
-    #[serde(default)]
-    pub start_time: String,
-}
-
-/// Heal task status response (query, cancel-with-token, start-then-poll).
-#[derive(Debug, Clone, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct HealTaskStatus {
-    /// `running` | `finished` | `stopped` | `notFound`.
-    pub summary: String,
-    /// Failure detail for stopped tasks; empty otherwise.
-    #[serde(rename = "detail", default)]
-    pub failure_detail: String,
-    #[serde(default)]
-    pub start_time: String,
-    #[serde(default)]
-    pub settings: HealOpts,
-    #[serde(default)]
-    pub items: Vec<HealResultItem>,
-    #[serde(default)]
-    pub truncated: bool,
-    /// Live progress snapshot; the exact shape is owned by the heal runtime.
-    #[serde(default)]
-    pub progress: Option<serde_json::Value>,
-}
-
-/// `POST /v3/background-heal/status` response. Known top-level fields are
-/// typed; the flattened heal info and operations matrix pass through verbatim.
-#[derive(Debug, Clone, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct BackgroundHealStatus {
-    /// `disabled` | `uninitialized` | `idle` | `active` | `degraded`.
-    pub state: String,
-    #[serde(default)]
-    pub heal_queue_length: u64,
-    #[serde(default)]
-    pub heal_active_tasks: u64,
-    #[serde(default)]
-    pub cluster_status_complete: bool,
-    #[serde(default)]
-    pub progress: Option<serde_json::Value>,
-    /// Remaining wire fields (flattened `BackgroundHealInfo` plus the
-    /// priority-by-source operations matrix), carried verbatim.
-    #[serde(flatten)]
-    pub extra: serde_json::Map<String, serde_json::Value>,
-}
-
-/// `GET /v3/scanner/status` response, typed at the fields operators branch
-/// on; everything else passes through verbatim.
-#[derive(Debug, Clone, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct ScannerStatus {
-    pub enabled: bool,
-    /// `fresh` | `stale` | `unknown`; absent when the scanner never completed
-    /// a cycle.
-    #[serde(default)]
-    pub freshness: Option<ScannerFreshness>,
-    #[serde(flatten)]
-    pub extra: serde_json::Map<String, serde_json::Value>,
-}
-
-/// Freshness block of the scanner status response.
-#[derive(Debug, Clone, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct ScannerFreshness {
-    /// `fresh` | `stale` | `unknown`.
-    pub state: String,
-}
-
-impl ScannerStatus {
-    /// Convenience accessor for the freshness state string.
-    pub fn freshness(&self) -> &str {
-        self.freshness
-            .as_ref()
-            .map(|freshness| freshness.state.as_str())
-            .unwrap_or("unknown")
-    }
-}
-
-/// Everything that can go wrong in an admin client call.
-#[derive(Debug)]
-pub enum AdminClientError {
-    /// The endpoint URL could not be parsed.
-    InvalidEndpoint(String),
-    /// Request build/send failed (DNS, connect, timeout, body read).
-    Transport(reqwest::Error),
-    /// The server answered a non-2xx status.
-    HttpStatus { status: u16, body: String },
-    /// The response body did not decode into the expected shape.
-    Decode { message: String },
-}
-
-impl std::fmt::Display for AdminClientError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::InvalidEndpoint(message) => write!(f, "invalid admin endpoint: {message}"),
-            Self::Transport(err) => write!(f, "admin request transport failure: {err}"),
-            Self::HttpStatus { status, body } => write!(f, "admin request failed with HTTP {status}: {body}"),
-            Self::Decode { message } => write!(f, "admin response decode failure: {message}"),
-        }
-    }
-}
-
-impl std::error::Error for AdminClientError {}
-
-impl From<reqwest::Error> for AdminClientError {
-    fn from(err: reqwest::Error) -> Self {
-        Self::Transport(err)
-    }
-}
-
-/// A signed client for a RustFS admin API.
-#[derive(Debug, Clone)]
-pub struct AdminClient {
-    endpoint: reqwest::Url,
-    access_key: String,
-    secret_key: String,
-    session_token: String,
-    region: String,
-    api_prefix: String,
-    http: reqwest::Client,
-}
-
-impl AdminClient {
-    /// Build a client for `endpoint` (e.g. `http://127.0.0.1:9000`) using root
-    /// or admin credentials. Requests are SigV4-signed with the same scheme
-    /// the server's admin router authenticates.
-    pub fn new(endpoint: &str, access_key: &str, secret_key: &str) -> Result<Self, AdminClientError> {
-        let url = reqwest::Url::parse(endpoint).map_err(|err| AdminClientError::InvalidEndpoint(err.to_string()))?;
-        if url.host_str().is_none() {
-            return Err(AdminClientError::InvalidEndpoint("endpoint has no host".to_string()));
-        }
-        let http = reqwest::Client::builder()
-            .connect_timeout(Duration::from_secs(10))
-            .timeout(Duration::from_secs(30))
-            .build()
-            .map_err(AdminClientError::Transport)?;
-        Ok(Self {
-            endpoint: url,
-            access_key: access_key.to_string(),
-            secret_key: secret_key.to_string(),
-            session_token: String::new(),
-            region: DEFAULT_REGION.to_string(),
-            api_prefix: DEFAULT_ADMIN_API_PREFIX.to_string(),
-            http,
-        })
-    }
-
-    /// Attach an STS session token (signed as `x-amz-security-token`).
-    pub fn with_session_token(mut self, session_token: impl Into<String>) -> Self {
-        self.session_token = session_token.into();
-        self
-    }
-
-    /// Override the SigV4 region (defaults to `us-east-1`, matching a
-    /// region-less RustFS deployment).
-    pub fn with_region(mut self, region: impl Into<String>) -> Self {
-        self.region = region.into();
-        self
-    }
-
-    /// Override the admin API path prefix (defaults to `/rustfs/admin`).
-    pub fn with_api_prefix(mut self, prefix: impl Into<String>) -> Self {
-        self.api_prefix = prefix.into();
-        self
-    }
-
-    /// Start a heal. `bucket` empty and `prefix` empty heals the whole
-    /// deployment (requires `recursive` or a `pool`/`set` pair in `opts`,
-    /// enforced server-side); a bucket alone heals the bucket (the server
-    /// forces `recursive` for bucket heals).
-    pub async fn heal_start(
-        &self,
-        bucket: Option<&str>,
-        prefix: Option<&str>,
-        opts: &HealOpts,
-        force_start: bool,
-    ) -> Result<HealStartSuccess, AdminClientError> {
-        let body = serde_json::to_vec(opts).map_err(|err| AdminClientError::Decode {
-            message: err.to_string(),
-        })?;
-        let mut query = Vec::new();
-        if force_start {
-            query.push(("forceStart", "true".to_string()));
-        }
-        self.post_json(&heal_path(bucket, prefix), &query, body).await
-    }
-
-    /// Query the status of the heal identified by `client_token` (the token
-    /// returned by [`Self::heal_start`]) at the path it was started on.
-    pub async fn heal_status(
-        &self,
-        bucket: Option<&str>,
-        prefix: Option<&str>,
-        client_token: &str,
-    ) -> Result<HealTaskStatus, AdminClientError> {
-        self.post_json(&heal_path(bucket, prefix), &[("clientToken", client_token.to_string())], Vec::new())
-            .await
-    }
-
-    /// Stop a heal: with a `client_token` only that task is cancelled and its
-    /// final status returned; without one, every heal task at the path is
-    /// cancelled (the server answers with a start-success-shaped receipt).
-    pub async fn heal_stop(
-        &self,
-        bucket: Option<&str>,
-        prefix: Option<&str>,
-        client_token: Option<&str>,
-    ) -> Result<HealStopOutcome, AdminClientError> {
-        let mut query = vec![("forceStop", "true".to_string())];
-        if let Some(token) = client_token {
-            query.push(("clientToken", token.to_string()));
-        }
-        match client_token {
-            Some(_) => {
-                let status: HealTaskStatus = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?;
-                Ok(HealStopOutcome::Stopped(status))
-            }
-            None => {
-                let success: HealStartSuccess = self.post_json(&heal_path(bucket, prefix), &query, Vec::new()).await?;
-                Ok(HealStopOutcome::PathStopped(success))
-            }
-        }
-    }
-
-    /// Cluster-aggregated background heal status.
-    pub async fn background_heal_status(&self) -> Result<BackgroundHealStatus, AdminClientError> {
-        self.get_json("/v3/background-heal/status").await
-    }
-
-    /// Data scanner status (enabled state, freshness, runtime config).
-    pub async fn scanner_status(&self) -> Result<ScannerStatus, AdminClientError> {
-        self.get_json("/v3/scanner/status").await
-    }
-
-    /// ILM expiry worker status. The payload is owned by the expiry
-    /// subsystem and still evolving; returned verbatim.
-    pub async fn ilm_expiry_status(&self) -> Result<serde_json::Value, AdminClientError> {
-        self.get_json("/v3/ilm/expiry/status").await
-    }
-
-    /// Durable replacement-recovery status (admin v4). The payload is owned
-    /// by the heal runtime; returned verbatim.
-    pub async fn replacement_recovery_status(&self) -> Result<serde_json::Value, AdminClientError> {
-        self.get_json("/v4/heal/replacement-recovery").await
-    }
-
-    /// Signed GET returning a decoded JSON body; escape hatch for endpoints
-    /// this client does not wrap yet.
-    pub async fn get_json<T: for<'de> Deserialize<'de>>(&self, path: &str) -> Result<T, AdminClientError> {
-        let url = self.url_for(path, &[])?;
-        let request = self.sign_and_build(Method::GET, url, Vec::new(), None).await?;
-        self.execute(request).await
-    }
-
-    /// Signed POST returning a decoded JSON body.
-    async fn post_json<T: for<'de> Deserialize<'de>>(
-        &self,
-        path: &str,
-        query: &[(&str, String)],
-        body: Vec<u8>,
-    ) -> Result<T, AdminClientError> {
-        let content_type = if body.is_empty() { None } else { Some("application/json") };
-        let url = self.url_for(path, query)?;
-        let request = self.sign_and_build(Method::POST, url, body, content_type).await?;
-        self.execute(request).await
-    }
-
-    fn url_for(&self, path: &str, query: &[(&str, String)]) -> Result<reqwest::Url, AdminClientError> {
-        let mut url = self
-            .endpoint
-            .join(&format!("{}{}", self.api_prefix.trim_end_matches('/'), path))
-            .map_err(|err| AdminClientError::InvalidEndpoint(err.to_string()))?;
-        if !query.is_empty() {
-            let mut pairs = url.query_pairs_mut();
-            for (key, value) in query {
-                pairs.append_pair(key, value);
-            }
-        }
-        Ok(url)
-    }
-
-    /// Build a SigV4-signed request via the same signer the server trusts,
-    /// then hand the signed headers to the HTTP client. The signature covers
-    /// method, path, query, and an unsigned-payload marker — the same shape
-    /// RustFS itself sends for peer admin calls.
-    async fn sign_and_build(
-        &self,
-        method: Method,
-        url: reqwest::Url,
-        body: Vec<u8>,
-        content_type: Option<&str>,
-    ) -> Result<reqwest::Request, AdminClientError> {
-        let authority = match (url.host_str(), url.port_or_known_default()) {
-            (Some(host), Some(port)) => format!("{host}:{port}"),
-            _ => return Err(AdminClientError::InvalidEndpoint("endpoint has no authority".to_string())),
-        };
-        let mut builder = http::Request::builder()
-            .method(method.clone())
-            .uri(url.as_str())
-            .header(http::header::HOST, &authority)
-            .header("x-amz-content-sha256", rustfs_signer::constants::UNSIGNED_PAYLOAD);
-        if let Some(content_type) = content_type {
-            builder = builder.header(http::header::CONTENT_TYPE, content_type);
-        }
-        let unsigned = builder
-            .body(s3s::Body::empty())
-            .map_err(|err| AdminClientError::InvalidEndpoint(format!("build request failed: {err}")))?;
-        let signed = rustfs_signer::sign_v4(
-            unsigned,
-            body.len() as i64,
-            &self.access_key,
-            &self.secret_key,
-            &self.session_token,
-            &self.region,
-        );
-
-        let mut request = self
-            .http
-            .request(method, url)
-            .body(body)
-            .build()
-            .map_err(AdminClientError::Transport)?;
-        let headers = request.headers_mut();
-        for (name, value) in signed.headers().iter() {
-            // HOST is owned by the HTTP client; the signed value above was
-            // built from the same URL authority, so they always agree.
-            if name == http::header::HOST {
-                continue;
-            }
-            headers.insert(name, value.clone());
-        }
-        Ok(request)
-    }
-
-    async fn execute<T: for<'de> Deserialize<'de>>(&self, request: reqwest::Request) -> Result<T, AdminClientError> {
-        let response = self.http.execute(request).await?;
-        let status = response.status();
-        let bytes = response.bytes().await?;
-        if !status.is_success() {
-            return Err(AdminClientError::HttpStatus {
-                status: status.as_u16(),
-                body: String::from_utf8_lossy(&bytes).into_owned(),
-            });
-        }
-        serde_json::from_slice(&bytes).map_err(|err| AdminClientError::Decode {
-            message: err.to_string(),
-        })
-    }
-}
-
-/// Response of [`AdminClient::heal_stop`]: cancelling a single tokened task
-/// answers with that task's status, cancelling a whole path answers with a
-/// start-success-shaped receipt.
-#[derive(Debug, Clone)]
-pub enum HealStopOutcome {
-    Stopped(HealTaskStatus),
-    PathStopped(HealStartSuccess),
-}
-
-fn heal_path(bucket: Option<&str>, prefix: Option<&str>) -> String {
-    match (bucket, prefix) {
-        (Some(bucket), Some(prefix)) if !bucket.is_empty() && !prefix.is_empty() => {
-            format!("/v3/heal/{}/{}", percent_encode_path_segment(bucket), percent_encode_path_segment(prefix))
-        }
-        (Some(bucket), Some(_)) | (Some(bucket), None) if !bucket.is_empty() => {
-            format!("/v3/heal/{}", percent_encode_path_segment(bucket))
-        }
-        _ => "/v3/heal/".to_string(),
-    }
-}
-
-/// Encode a single path segment (slashes are content, not separators, inside
-/// bucket/prefix path params).
-fn percent_encode_path_segment(segment: &str) -> String {
-    let mut out = String::with_capacity(segment.len());
-    for byte in segment.bytes() {
-        match byte {
-            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(byte as char),
-            _ => out.push_str(&format!("%{byte:02X}")),
-        }
-    }
-    out
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{
-        AdminClient, AdminClientError, BackgroundHealStatus, HealOpts, HealScanMode, HealStartSuccess, HealTaskStatus,
-        ScannerStatus, heal_path, percent_encode_path_segment,
-    };
-    use serde_json::json;
-    use std::sync::{Arc, Mutex};
-
-    #[test]
-    fn heal_paths_cover_root_bucket_and_prefix() {
-        assert_eq!(heal_path(None, None), "/v3/heal/");
-        assert_eq!(heal_path(Some(""), Some("")), "/v3/heal/");
-        assert_eq!(heal_path(Some("bucket"), None), "/v3/heal/bucket");
-        assert_eq!(heal_path(Some("bucket"), Some("pre/fix")), "/v3/heal/bucket/pre%2Ffix");
-    }
-
-    #[test]
-    fn path_segments_percent_encode_reserved_characters() {
-        assert_eq!(percent_encode_path_segment("a b"), "a%20b");
-        assert_eq!(percent_encode_path_segment("a/b"), "a%2Fb");
-        assert_eq!(percent_encode_path_segment("ü"), "%C3%BC");
-    }
-
-    #[test]
-    fn heal_opts_round_trip_through_the_server_wire_shape() {
-        let opts = HealOpts {
-            recursive: true,
-            dry_run: false,
-            remove: true,
-            recreate: false,
-            scan_mode: HealScanMode::Deep,
-            update_parity: true,
-            no_lock: false,
-            pool: Some(1),
-            set: Some(2),
-        };
-        let wire = serde_json::to_value(&opts).unwrap();
-        assert_eq!(wire["scanMode"], json!(2), "the server body decodes scanMode as a number");
-        let back: HealOpts = serde_json::from_value(wire).unwrap();
-        assert_eq!(back.scan_mode, HealScanMode::Deep);
-        assert_eq!(back.pool, Some(1));
-    }
-
-    #[test]
-    fn heal_scan_mode_accepts_both_wire_encodings() {
-        assert_eq!(serde_json::from_value::<HealScanMode>(json!(1)).unwrap(), HealScanMode::Normal);
-        assert_eq!(serde_json::from_value::<HealScanMode>(json!("deep")).unwrap(), HealScanMode::Deep);
-        assert!(serde_json::from_value::<HealScanMode>(json!(9)).is_err());
-        assert!(serde_json::from_value::<HealScanMode>(json!("sideways")).is_err());
-    }
-
-    #[test]
-    fn heal_task_status_decodes_the_server_response_shape() {
-        let raw = json!({
-            "summary": "finished",
-            "detail": "",
-            "startTime": "2026-08-17T00:00:00Z",
-            "settings": {"recursive": false, "scanMode": 1},
-            "items": [{
-                "resultId": 1, "type": "object", "bucket": "b", "object": "o", "versionId": "", "detail": "",
-                "parityBlocks": 2, "dataBlocks": 2, "diskCount": 4, "setCount": 1,
-                "before": {"drives": []}, "after": {"drives": []}, "objectSize": 128
-            }],
-            "truncated": false
-        });
-        let status: HealTaskStatus = serde_json::from_value(raw).unwrap();
-        assert_eq!(status.summary, "finished");
-        assert_eq!(status.items.len(), 1);
-        assert_eq!(status.settings.scan_mode, HealScanMode::Normal);
-        assert!(status.progress.is_none());
-    }
-
-    #[test]
-    fn background_heal_status_types_known_fields_and_passes_the_rest_through() {
-        let raw = json!({
-            "state": "active",
-            "bitrotStartTime": "t",
-            "healQueueLength": 3,
-            "healActiveTasks": 1,
-            "healOperations": {"queueLength": 3},
-            "clusterStatusComplete": true
-        });
-        let status: BackgroundHealStatus = serde_json::from_value(raw).unwrap();
-        assert_eq!(status.state, "active");
-        assert_eq!(status.heal_queue_length, 3);
-        assert!(status.cluster_status_complete);
-        assert!(status.extra.contains_key("healOperations"), "unknown nested payloads must pass through");
-    }
-
-    #[test]
-    fn scanner_status_defaults_freshness_to_unknown() {
-        let raw = json!({"enabled": true, "freshness": {"state": "stale"}, "metrics": {}});
-        let status: ScannerStatus = serde_json::from_value(raw).unwrap();
-        assert_eq!(status.freshness(), "stale");
-        let bare: ScannerStatus = serde_json::from_value(json!({"enabled": false})).unwrap();
-        assert_eq!(bare.freshness(), "unknown");
-    }
-
-    #[test]
-    fn invalid_endpoint_is_rejected_without_io() {
-        let err = AdminClient::new("not a url", "ak", "sk").unwrap_err();
-        assert!(matches!(err, AdminClientError::InvalidEndpoint(_)));
-    }
-
-    #[tokio::test]
-    async fn signed_requests_carry_sigv4_authorization_and_correct_target() {
-        let server = TestServer::spawn(r#"{"clientToken":"token-1","clientAddress":"127.0.0.1:9","startTime":"t"}"#, 200).await;
-        let client = AdminClient::new(&format!("http://{}", server.addr), "minioadmin", "minioadmin")
-            .expect("client builds against the test server");
-
-        let start: HealStartSuccess = client
-            .heal_start(
-                Some("bucket"),
-                None,
-                &HealOpts {
-                    recursive: true,
-                    ..Default::default()
-                },
-                false,
-            )
-            .await
-            .expect("signed heal start decodes");
-
-        assert_eq!(start.client_token, "token-1");
-        let request = server.recorded();
-        assert_eq!(request.method, "POST");
-        assert_eq!(request.path, "/rustfs/admin/v3/heal/bucket");
-        assert!(!request.query.contains("forceStart"), "absent flags must not be sent");
-        let auth = request.header("authorization").expect("request must be signed");
-        assert!(auth.starts_with("AWS4-HMAC-SHA256"), "SigV4 scheme, got: {auth}");
-        assert!(auth.contains("Credential=minioadmin/"), "credentials must be in the Authorization header");
-        assert_eq!(
-            request.header("x-amz-content-sha256").as_deref(),
-            Some("UNSIGNED-PAYLOAD"),
-            "the client signs the same payload marker RustFS peer calls use"
-        );
-        assert_eq!(request.header("content-type").as_deref(), Some("application/json"));
-        assert!(request.body.contains("\"recursive\":true"));
-    }
-
-    #[tokio::test]
-    async fn query_sends_client_token_on_the_same_path() {
-        let body = r#"{"summary":"running","detail":"","settings":{"recursive":false},"items":[],"truncated":false}"#;
-        let server = TestServer::spawn(body, 200).await;
-        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
-
-        let status = client
-            .heal_status(Some("bucket"), None, "token-1")
-            .await
-            .expect("status decodes");
-        assert_eq!(status.summary, "running");
-        let request = server.recorded();
-        assert_eq!(request.path, "/rustfs/admin/v3/heal/bucket");
-        assert!(request.query.contains("clientToken=token-1"));
-        assert!(!request.query.contains("forceStop"));
-    }
-
-    #[tokio::test]
-    async fn stop_without_token_takes_the_path_cancel_branch() {
-        let server = TestServer::spawn(r#"{"clientToken":"path","clientAddress":"c","startTime":"t"}"#, 200).await;
-        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
-
-        let outcome = client.heal_stop(Some("bucket"), None, None).await.expect("path stop decodes");
-        assert!(matches!(outcome, super::HealStopOutcome::PathStopped(_)));
-        let request = server.recorded();
-        assert!(request.query.contains("forceStop=true"));
-        assert!(!request.query.contains("clientToken"));
-    }
-
-    #[tokio::test]
-    async fn http_error_status_maps_to_a_typed_error_with_body() {
-        let server = TestServer::spawn(r#"{"code":"AccessDenied","message":"denied"}"#, 403).await;
-        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
-        let err = client.scanner_status().await.unwrap_err();
-        match err {
-            AdminClientError::HttpStatus { status, body } => {
-                assert_eq!(status, 403);
-                assert!(body.contains("AccessDenied"));
-            }
-            other => panic!("expected HttpStatus, got {other:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn malformed_success_body_maps_to_a_decode_error() {
-        let server = TestServer::spawn("not json", 200).await;
-        let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap();
-        assert!(matches!(client.scanner_status().await.unwrap_err(), AdminClientError::Decode { .. }));
-    }
-
-    /// One recorded request, parsed off the wire with the minimum needed for
-    /// assertions: method, path, query, headers, body.
-    #[derive(Debug, Clone)]
-    struct RecordedRequest {
-        method: String,
-        path: String,
-        query: String,
-        headers: Vec<(String, String)>,
-        body: String,
-    }
-
-    impl RecordedRequest {
-        fn header(&self, name: &str) -> Option<String> {
-            self.headers
-                .iter()
-                .find(|(key, _)| key.eq_ignore_ascii_case(name))
-                .map(|(_, value)| value.clone())
-        }
-    }
-
-    /// Minimal HTTP/1.1 server: one canned response per connection, every
-    /// request recorded behind an `Arc<Mutex>`. Deliberately dependency-free —
-    /// the assertions only need the raw request bytes.
-    struct TestServer {
-        addr: std::net::SocketAddr,
-        requests: Arc<Mutex<Vec<RecordedRequest>>>,
-    }
-
-    impl TestServer {
-        async fn spawn(response_body: &'static str, status: u16) -> Self {
-            use tokio::io::{AsyncReadExt, AsyncWriteExt};
-
-            let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
-                .await
-                .expect("bind ephemeral port");
-            let addr = listener.local_addr().expect("local addr");
-            let requests: Arc<Mutex<Vec<RecordedRequest>>> = Arc::new(Mutex::new(Vec::new()));
-
-            let recorded = requests.clone();
-            tokio::spawn(async move {
-                let reason = if status == 200 { "OK" } else { "Forbidden" };
-                let response = format!(
-                    "HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response_body}",
-                    response_body.len()
-                );
-                // Each request is a fresh connection (connection: close); a
-                // bounded loop serves every call a test makes while letting
-                // the task exit instead of lingering for the whole process.
-                for _ in 0..16 {
-                    let Ok((mut stream, _)) = listener.accept().await else {
-                        break;
-                    };
-                    let mut buffer = Vec::with_capacity(2048);
-                    let mut chunk = [0u8; 2048];
-                    // Read headers plus content-length body, or stop on close.
-                    loop {
-                        if let Some(end) = find_header_end(&buffer) {
-                            let content_length = extract_content_length(&buffer[..end]);
-                            if buffer.len() >= end + content_length {
-                                break;
-                            }
-                        }
-                        let n = match stream.read(&mut chunk).await {
-                            Ok(0) | Err(_) => break,
-                            Ok(n) => n,
-                        };
-                        buffer.extend_from_slice(&chunk[..n]);
-                        if buffer.len() > 64 * 1024 {
-                            break;
-                        }
-                    }
-                    if let Some(request) = parse_request(&buffer) {
-                        recorded.lock().expect("recorded lock").push(request);
-                    }
-                    let _ = stream.write_all(response.as_bytes()).await;
-                    let _ = stream.shutdown().await;
-                }
-            });
-
-            Self { addr, requests }
-        }
-
-        fn recorded(&self) -> RecordedRequest {
-            self.requests
-                .lock()
-                .expect("recorded lock")
-                .last()
-                .cloned()
-                .expect("the client call must have produced one recorded request")
-        }
-    }
-
-    fn find_header_end(buffer: &[u8]) -> Option<usize> {
-        buffer.windows(4).position(|window| window == b"\r\n\r\n").map(|pos| pos + 4)
-    }
-
-    fn extract_content_length(headers: &[u8]) -> usize {
-        let text = String::from_utf8_lossy(headers).to_ascii_lowercase();
-        text.lines()
-            .find_map(|line| line.strip_prefix("content-length:"))
-            .and_then(|value| value.trim().parse().ok())
-            .unwrap_or(0)
-    }
-
-    fn parse_request(raw: &[u8]) -> Option<RecordedRequest> {
-        let end = find_header_end(raw)?;
-        let head = String::from_utf8_lossy(&raw[..end]);
-        let body = String::from_utf8_lossy(&raw[end..]).into_owned();
-        let mut lines = head.lines();
-        let request_line = lines.next()?;
-        let mut parts = request_line.split_whitespace();
-        let method = parts.next()?.to_string();
-        let target = parts.next()?.to_string();
-        let (path, query) = match target.split_once('?') {
-            Some((path, query)) => (path.to_string(), query.to_string()),
-            None => (target, String::new()),
-        };
-        let headers = lines
-            .filter_map(|line| line.split_once(':'))
-            .map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
-            .collect();
-        Some(RecordedRequest {
-            method,
-            path,
-            query,
-            headers,
-            body,
-        })
-    }
-}

crates/madmin/src/lib.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod client;
 pub mod group;
 pub mod heal_commands;
 pub mod health;
@@ -26,7 +25,6 @@ pub mod trace;
 pub mod user;
 pub mod utils;
 
-pub use client::*;
 pub use group::*;
 pub use info_commands::*;
 pub use policy::*;

rustfs/src/admin/handlers/site_replication.rs

@@ -66,18 +66,20 @@ use rustfs_config::{
 };
 use rustfs_iam::error::is_err_no_such_service_account;
 use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
+use rustfs_iam::store::object::ObjectStore;
 use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type, user_type_from_sr_wire};
 use rustfs_iam::sys::{
-    NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
+    IamSys, NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
 };
 use rustfs_madmin::{
     AddOrUpdateUserReq, BucketBandwidth, GroupAddRemove, GroupStatus, IDPSettings, InProgressMetric, InQueueMetric,
     LDAPConfigSettings, LDAPSettings, OpenIDProviderSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus,
     ReplicateEditStatus, ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC,
     SR_IAM_ITEM_STS_ACC_LEGACY, SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem,
-    SRIAMPolicy, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation,
-    SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRSiteSummary,
-    SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
+    SRIAMPolicy, SRIAMUser, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq,
+    SRPendingOperation, SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSTSCredential,
+    SRSessionPolicy, SRSiteSummary, SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate,
+    SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
 };
 use rustfs_policy::policy::{
     Policy,
@@ -4584,7 +4586,10 @@ async fn build_metrics_summary(local_peer: &PeerInfo) -> SRMetricsSummary {
             head_failed_total: non_negative_u64(node.proxy_head_failed),
             put_tag_total: non_negative_u64(node.proxy_put_tag_total),
             put_tag_failed_total: non_negative_u64(node.proxy_put_tag_failed),
-            ..Default::default()
+            get_tag_total: non_negative_u64(node.proxy_get_tag_total),
+            get_tag_failed_total: non_negative_u64(node.proxy_get_tag_failed),
+            remove_tag_total: non_negative_u64(node.proxy_delete_tag_total),
+            remove_tag_failed_total: non_negative_u64(node.proxy_delete_tag_failed),
         },
         metrics,
         uptime: node.uptime,
@@ -9278,247 +9283,16 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
     let incoming_updated_at = item.updated_at;
 
     match item.r#type.as_str() {
-        "policy" => {
-            if let Some(policy) = item.policy {
-                let policy: Policy =
-                    serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
-                iam_sys.set_policy(&item.name, policy).await.map_err(ApiError::from)?;
-            } else {
-                iam_sys.delete_policy(&item.name, true).await.map_err(ApiError::from)?;
-            }
-            Ok(())
-        }
-        "policy-mapping" => {
-            let Some(mapping) = item.policy_mapping else {
-                return Err(s3_error!(InvalidRequest, "policyMapping is required"));
-            };
-            let user_type =
-                user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
-            iam_sys
-                .policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
-                .await
-                .map_err(ApiError::from)?;
-            Ok(())
-        }
-        "group-info" => {
-            let Some(group_info) = item.group_info else {
-                return Err(s3_error!(InvalidRequest, "groupInfo is required"));
-            };
-            let update = group_info.update_req;
-            if !group_info_requires_upsert(&update) {
-                iam_sys
-                    .remove_users_from_group(&update.group, update.members)
-                    .await
-                    .map_err(ApiError::from)?;
-                return Ok(());
-            }
-
-            iam_sys
-                .add_users_to_group(&update.group, update.members)
-                .await
-                .map_err(ApiError::from)?;
-            iam_sys
-                .set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
-                .await
-                .map_err(ApiError::from)?;
-            Ok(())
-        }
+        "policy" => apply_iam_policy_item(&iam_sys, &item.name, item.policy).await,
+        "policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, item.policy_mapping).await,
+        "group-info" => apply_iam_group_info_item(&iam_sys, item.group_info).await,
         // MinIO madmin-go sends `SRIAMItemSTSAcc = "sts-account"`. The legacy alias
         // `sts-credential` (emitted by older RustFS releases) stays accepted permanently
         // so mixed-version RustFS sites keep replicating STS credentials during rolling
         // upgrades; it is a compatibility layer, not temporary code.
-        SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => {
-            let Some(sts_credential) = item.sts_credential else {
-                return Err(s3_error!(InvalidRequest, "stsCredential is required"));
-            };
-            let Some(secret) = current_token_signing_key() else {
-                return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
-            };
-            let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
-                .map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
-            let expiration = claims
-                .get("exp")
-                .and_then(claims_unix_timestamp)
-                .map(OffsetDateTime::from_unix_timestamp)
-                .transpose()
-                .map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
-            let groups = string_list_claim(&claims, "groups");
-            let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
-            let cred = rustfs_credentials::Credentials {
-                access_key: sts_credential.access_key.clone(),
-                secret_key: sts_credential.secret_key.clone(),
-                session_token: sts_credential.session_token.clone(),
-                expiration,
-                status: "on".to_string(),
-                parent_user: sts_credential.parent_user.clone(),
-                groups,
-                claims: Some(claims),
-                ..Default::default()
-            };
-            iam_sys
-                .set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
-                .await
-                .map_err(ApiError::from)?;
-            Ok(())
-        }
-        "iam-user" => {
-            let Some(user) = item.iam_user else {
-                return Err(s3_error!(InvalidRequest, "iamUser is required"));
-            };
-            if let Some(local) = iam_sys.get_user(&user.access_key).await
-                && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
-            {
-                return Ok(());
-            }
-            if user.is_delete_req {
-                iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
-            } else {
-                let Some(user_req) = user.user_req else {
-                    return Err(s3_error!(InvalidRequest, "userReq is required"));
-                };
-                let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
-                if is_status_only_update {
-                    iam_sys
-                        .set_user_status(&user.access_key, user_req.status)
-                        .await
-                        .map_err(ApiError::from)?;
-                } else {
-                    iam_sys
-                        .create_user(&user.access_key, &user_req)
-                        .await
-                        .map_err(ApiError::from)?;
-                }
-            }
-            Ok(())
-        }
-        "service-account" => {
-            let Some(change) = item.svc_acc_change else {
-                return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
-            };
-            let envelope = change.oidc_service_account_envelope;
-            if let Some(create) = change.create {
-                let local_updated_at = iam_sys
-                    .get_user(&create.access_key)
-                    .await
-                    .map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
-                let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
-                    if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
-                        return Ok(());
-                    }
-                    ReplicatedServiceAccountPolicy {
-                        policy: Some(site_replicator_service_account_policy()?),
-                        is_envelope: false,
-                    }
-                } else {
-                    let Some(replicated_policy) = decode_service_account_replication_policy(
-                        &create,
-                        envelope.as_ref(),
-                        incoming_updated_at,
-                        local_updated_at,
-                    )?
-                    else {
-                        return Ok(());
-                    };
-                    replicated_policy
-                };
-                match iam_sys.get_service_account(&create.access_key).await {
-                    Ok((existing, _)) => {
-                        if existing.parent_user != create.parent {
-                            return Err(s3_error!(
-                                InvalidRequest,
-                                "service account {} already exists with a different parent user",
-                                create.access_key
-                            ));
-                        }
-                        iam_sys
-                            .update_service_account(
-                                &create.access_key,
-                                UpdateServiceAccountOpts {
-                                    name: replicated_policy.metadata_for_existing_account(create.name),
-                                    description: replicated_policy.metadata_for_existing_account(create.description),
-                                    session_policy: replicated_policy.for_existing_account(),
-                                    secret_key: Some(create.secret_key),
-                                    expiration: create.expiration,
-                                    status: (!create.status.is_empty()).then_some(create.status),
-                                    parent_user: None,
-                                    allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
-                                },
-                            )
-                            .await
-                            .map_err(ApiError::from)?;
-                    }
-                    Err(err) if is_err_no_such_service_account(&err) => {
-                        iam_sys
-                            .new_service_account(
-                                &create.parent,
-                                Some(create.groups),
-                                NewServiceAccountOpts {
-                                    session_policy: replicated_policy.policy,
-                                    access_key: create.access_key,
-                                    secret_key: create.secret_key,
-                                    name: (!create.name.is_empty()).then_some(create.name),
-                                    description: (!create.description.is_empty()).then_some(create.description),
-                                    expiration: create.expiration,
-                                    allow_site_replicator_account: true,
-                                    claims: Some(create.claims),
-                                },
-                            )
-                            .await
-                            .map_err(ApiError::from)?;
-                    }
-                    Err(err) => return Err(ApiError::from(err).into()),
-                }
-                return Ok(());
-            }
-
-            if let Some(update) = change.update {
-                if let Some(local) = iam_sys.get_user(&update.access_key).await
-                    && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
-                {
-                    return Ok(());
-                }
-                let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
-                let session_policy = if allow_site_replicator_account {
-                    Some(site_replicator_service_account_policy()?)
-                } else {
-                    update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
-                };
-                iam_sys
-                    .update_service_account(
-                        &update.access_key,
-                        UpdateServiceAccountOpts {
-                            session_policy,
-                            secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
-                            name: (!update.name.is_empty()).then_some(update.name),
-                            description: (!update.description.is_empty()).then_some(update.description),
-                            expiration: update.expiration,
-                            status: (!update.status.is_empty()).then_some(update.status),
-                            // Peers replicate credentials, never the local parent binding:
-                            // each site resolves its own parent from its own IAM.
-                            parent_user: None,
-                            allow_site_replicator_account,
-                        },
-                    )
-                    .await
-                    .map_err(ApiError::from)?;
-                return Ok(());
-            }
-
-            if let Some(delete) = change.delete {
-                if let Some(local) = iam_sys.get_user(&delete.access_key).await
-                    && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
-                {
-                    return Ok(());
-                }
-                iam_sys
-                    .delete_service_account(&delete.access_key, true)
-                    .await
-                    .map_err(ApiError::from)?;
-                return Ok(());
-            }
-
-            Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
-        }
+        SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => apply_iam_sts_account_item(&iam_sys, item.sts_credential).await,
+        "iam-user" => apply_iam_user_item(&iam_sys, item.iam_user, incoming_updated_at).await,
+        "service-account" => apply_iam_service_account_item(&iam_sys, item.svc_acc_change, incoming_updated_at).await,
         _ => Err(s3_error!(
             NotImplemented,
             "site replication IAM item type `{}` is not supported",
@@ -9527,6 +9301,252 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
     }
 }
 
+async fn apply_iam_policy_item(iam_sys: &IamSys<ObjectStore>, name: &str, policy: Option<Value>) -> S3Result<()> {
+    if let Some(policy) = policy {
+        let policy: Policy =
+            serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
+        iam_sys.set_policy(name, policy).await.map_err(ApiError::from)?;
+    } else {
+        iam_sys.delete_policy(name, true).await.map_err(ApiError::from)?;
+    }
+    Ok(())
+}
+
+async fn apply_iam_policy_mapping_item(iam_sys: &IamSys<ObjectStore>, policy_mapping: Option<SRPolicyMapping>) -> S3Result<()> {
+    let Some(mapping) = policy_mapping else {
+        return Err(s3_error!(InvalidRequest, "policyMapping is required"));
+    };
+    let user_type = user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
+    iam_sys
+        .policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
+        .await
+        .map_err(ApiError::from)?;
+    Ok(())
+}
+
+async fn apply_iam_group_info_item(iam_sys: &IamSys<ObjectStore>, group_info: Option<SRGroupInfo>) -> S3Result<()> {
+    let Some(group_info) = group_info else {
+        return Err(s3_error!(InvalidRequest, "groupInfo is required"));
+    };
+    let update = group_info.update_req;
+    if !group_info_requires_upsert(&update) {
+        iam_sys
+            .remove_users_from_group(&update.group, update.members)
+            .await
+            .map_err(ApiError::from)?;
+        return Ok(());
+    }
+
+    iam_sys
+        .add_users_to_group(&update.group, update.members)
+        .await
+        .map_err(ApiError::from)?;
+    iam_sys
+        .set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
+        .await
+        .map_err(ApiError::from)?;
+    Ok(())
+}
+
+async fn apply_iam_sts_account_item(iam_sys: &IamSys<ObjectStore>, sts_credential: Option<SRSTSCredential>) -> S3Result<()> {
+    let Some(sts_credential) = sts_credential else {
+        return Err(s3_error!(InvalidRequest, "stsCredential is required"));
+    };
+    let Some(secret) = current_token_signing_key() else {
+        return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
+    };
+    let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
+        .map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
+    let expiration = claims
+        .get("exp")
+        .and_then(claims_unix_timestamp)
+        .map(OffsetDateTime::from_unix_timestamp)
+        .transpose()
+        .map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
+    let groups = string_list_claim(&claims, "groups");
+    let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
+    let cred = rustfs_credentials::Credentials {
+        access_key: sts_credential.access_key.clone(),
+        secret_key: sts_credential.secret_key.clone(),
+        session_token: sts_credential.session_token.clone(),
+        expiration,
+        status: "on".to_string(),
+        parent_user: sts_credential.parent_user.clone(),
+        groups,
+        claims: Some(claims),
+        ..Default::default()
+    };
+    iam_sys
+        .set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
+        .await
+        .map_err(ApiError::from)?;
+    Ok(())
+}
+
+async fn apply_iam_user_item(
+    iam_sys: &IamSys<ObjectStore>,
+    iam_user: Option<SRIAMUser>,
+    incoming_updated_at: Option<OffsetDateTime>,
+) -> S3Result<()> {
+    let Some(user) = iam_user else {
+        return Err(s3_error!(InvalidRequest, "iamUser is required"));
+    };
+    if let Some(local) = iam_sys.get_user(&user.access_key).await
+        && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
+    {
+        return Ok(());
+    }
+    if user.is_delete_req {
+        iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
+    } else {
+        let Some(user_req) = user.user_req else {
+            return Err(s3_error!(InvalidRequest, "userReq is required"));
+        };
+        let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
+        if is_status_only_update {
+            iam_sys
+                .set_user_status(&user.access_key, user_req.status)
+                .await
+                .map_err(ApiError::from)?;
+        } else {
+            iam_sys
+                .create_user(&user.access_key, &user_req)
+                .await
+                .map_err(ApiError::from)?;
+        }
+    }
+    Ok(())
+}
+
+async fn apply_iam_service_account_item(
+    iam_sys: &IamSys<ObjectStore>,
+    svc_acc_change: Option<SRSvcAccChange>,
+    incoming_updated_at: Option<OffsetDateTime>,
+) -> S3Result<()> {
+    let Some(change) = svc_acc_change else {
+        return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
+    };
+    let envelope = change.oidc_service_account_envelope;
+    if let Some(create) = change.create {
+        let local_updated_at = iam_sys
+            .get_user(&create.access_key)
+            .await
+            .map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
+        let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
+            if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
+                return Ok(());
+            }
+            ReplicatedServiceAccountPolicy {
+                policy: Some(site_replicator_service_account_policy()?),
+                is_envelope: false,
+            }
+        } else {
+            let Some(replicated_policy) =
+                decode_service_account_replication_policy(&create, envelope.as_ref(), incoming_updated_at, local_updated_at)?
+            else {
+                return Ok(());
+            };
+            replicated_policy
+        };
+        match iam_sys.get_service_account(&create.access_key).await {
+            Ok((existing, _)) => {
+                if existing.parent_user != create.parent {
+                    return Err(s3_error!(
+                        InvalidRequest,
+                        "service account {} already exists with a different parent user",
+                        create.access_key
+                    ));
+                }
+                iam_sys
+                    .update_service_account(
+                        &create.access_key,
+                        UpdateServiceAccountOpts {
+                            name: replicated_policy.metadata_for_existing_account(create.name),
+                            description: replicated_policy.metadata_for_existing_account(create.description),
+                            session_policy: replicated_policy.for_existing_account(),
+                            secret_key: Some(create.secret_key),
+                            expiration: create.expiration,
+                            status: (!create.status.is_empty()).then_some(create.status),
+                            parent_user: None,
+                            allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
+                        },
+                    )
+                    .await
+                    .map_err(ApiError::from)?;
+            }
+            Err(err) if is_err_no_such_service_account(&err) => {
+                iam_sys
+                    .new_service_account(
+                        &create.parent,
+                        Some(create.groups),
+                        NewServiceAccountOpts {
+                            session_policy: replicated_policy.policy,
+                            access_key: create.access_key,
+                            secret_key: create.secret_key,
+                            name: (!create.name.is_empty()).then_some(create.name),
+                            description: (!create.description.is_empty()).then_some(create.description),
+                            expiration: create.expiration,
+                            allow_site_replicator_account: true,
+                            claims: Some(create.claims),
+                        },
+                    )
+                    .await
+                    .map_err(ApiError::from)?;
+            }
+            Err(err) => return Err(ApiError::from(err).into()),
+        }
+        return Ok(());
+    }
+
+    if let Some(update) = change.update {
+        if let Some(local) = iam_sys.get_user(&update.access_key).await
+            && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
+        {
+            return Ok(());
+        }
+        let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
+        let session_policy = if allow_site_replicator_account {
+            Some(site_replicator_service_account_policy()?)
+        } else {
+            update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
+        };
+        iam_sys
+            .update_service_account(
+                &update.access_key,
+                UpdateServiceAccountOpts {
+                    session_policy,
+                    secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
+                    name: (!update.name.is_empty()).then_some(update.name),
+                    description: (!update.description.is_empty()).then_some(update.description),
+                    expiration: update.expiration,
+                    status: (!update.status.is_empty()).then_some(update.status),
+                    // Peers replicate credentials, never the local parent binding:
+                    // each site resolves its own parent from its own IAM.
+                    parent_user: None,
+                    allow_site_replicator_account,
+                },
+            )
+            .await
+            .map_err(ApiError::from)?;
+        return Ok(());
+    }
+
+    if let Some(delete) = change.delete {
+        if let Some(local) = iam_sys.get_user(&delete.access_key).await
+            && is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
+        {
+            return Ok(());
+        }
+        iam_sys
+            .delete_service_account(&delete.access_key, true)
+            .await
+            .map_err(ApiError::from)?;
+        return Ok(());
+    }
+
+    Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
+}
+
 fn claims_unix_timestamp(value: &Value) -> Option<i64> {
     match value {
         Value::Number(number) => number.as_i64(),

rustfs/src/admin/handlers/system.rs

@@ -417,13 +417,6 @@ struct SystemAdminDiscovery {
 struct ServerInfoResponse {
     info: InfoMessage,
     admin_discovery: SystemAdminDiscovery,
-    /// Startup bitrot algorithm self-test outcome (rustfs/backlog#1873):
-    /// `passed` (algorithms verified at boot), `failed` (a drifted hash
-    /// implementation — the process is serving with degraded integrity
-    /// checking unless `RUSTFS_BITROT_SELFTEST_STRICT` aborted it), or
-    /// `unknown` (not yet run or disabled).
-    #[serde(rename = "bitrotSelftest")]
-    bitrot_selftest: &'static str,
 }
 
 #[derive(Serialize)]
@@ -440,14 +433,6 @@ fn system_admin_discovery(usecase: &DefaultAdminUsecase) -> SystemAdminDiscovery
     }
 }
 
-fn bitrot_selftest_status_str() -> &'static str {
-    match crate::bitrot_selftest::bitrot_selftest_passed() {
-        Some(true) => "passed",
-        Some(false) => "failed",
-        None => "unknown",
-    }
-}
-
 #[async_trait::async_trait]
 impl Operation for ServerInfoHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -479,7 +464,6 @@ impl Operation for ServerInfoHandler {
         let response = ServerInfoResponse {
             info,
             admin_discovery: system_admin_discovery(&usecase),
-            bitrot_selftest: bitrot_selftest_status_str(),
         };
 
         let data = serde_json::to_vec(&response).map_err(|e| {
@@ -1551,18 +1535,6 @@ mod tests {
         );
     }
 
-    /// The startup bitrot self-test outcome must surface in server info as one
-    /// of three closed-set strings, never an internal enum or a null
-    /// (rustfs/backlog#1873). This test pins the string mapping; whether the
-    /// process-global cell holds Some(true)/Some(false)/None is owned by
-    /// `crate::bitrot_selftest`'s own tests.
-    #[test]
-    fn bitrot_selftest_status_str_is_a_closed_set_of_operators_strings() {
-        let rendered = super::bitrot_selftest_status_str();
-        assert!(matches!(rendered, "passed" | "failed" | "unknown"));
-        assert_eq!(super::bitrot_selftest_status_str(), rendered);
-    }
-
     #[test]
     fn server_info_response_exposes_admin_discovery_paths() {
         let usecase = DefaultAdminUsecase::without_context();
@@ -1584,7 +1556,6 @@ mod tests {
                 pools: None,
             },
             admin_discovery: system_admin_discovery(&usecase),
-            bitrot_selftest: super::bitrot_selftest_status_str(),
         };
 
         let value = serde_json::to_value(response).expect("server info response should serialize");

rustfs/src/app/object_usecase.rs

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

rustfs/src/app/storage_api.rs

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

rustfs/src/bitrot_selftest.rs

@@ -1,181 +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.
-
-//! Startup bitrot algorithm self-test (rustfs/backlog#1873).
-//!
-//! A drifted hash implementation fails silently in production: every shard
-//! reads back "corrupt", heal rewrites healthy data, and cross-platform
-//! clusters disagree about which copy is good. [`run_startup_bitrot_self_test`]
-//! pins the algorithms once at process start — the check itself runs in well
-//! under a millisecond on 4 KiB, so it executes inline before background
-//! services come up and the result is published before the server accepts
-//! traffic.
-//!
-//! Outcome surface:
-//! - one structured `bitrot_selftest` log event (`passed`/`failed`/`skipped`),
-//! - the `rustfs_bitrot_selftest_status` gauge (1=passed, 0=failed, 2=skipped),
-//! - [`bitrot_selftest_passed`] for admin/health surfaces,
-//! - `RUSTFS_BITROT_SELFTEST_STRICT=on` turns a failure into a startup error
-//!   (MinIO `bitrotSelfTest` Fatal parity); the default only degrades the
-//!   status so a bad build cannot brick an existing fleet on upgrade.
-
-use crate::storage_api::startup::background::{BitrotSelfTestError, bitrot_self_test};
-use metrics::gauge;
-use std::future::Future;
-use std::io;
-use std::sync::atomic::{AtomicU8, Ordering};
-use std::time::Instant;
-use tracing::{debug, error, info};
-
-const LOG_COMPONENT_MAIN: &str = "main";
-const LOG_SUBSYSTEM_STARTUP: &str = "startup";
-const EVENT_BITROT_SELFTEST: &str = "bitrot_selftest";
-const METRIC_BITROT_SELFTEST_STATUS: &str = "rustfs_bitrot_selftest_status";
-
-/// Gauge values for [`METRIC_BITROT_SELFTEST_STATUS`].
-const STATUS_PASSED: f64 = 1.0;
-const STATUS_FAILED: f64 = 0.0;
-const STATUS_SKIPPED: f64 = 2.0;
-
-/// Internal cell values for [`BITROT_SELF_TEST_STATUS`].
-const STATUS_CELL_UNSET: u8 = 0;
-const STATUS_CELL_PASSED: u8 = 1;
-const STATUS_CELL_FAILED: u8 = 2;
-
-static BITROT_SELF_TEST_STATUS: AtomicU8 = AtomicU8::new(STATUS_CELL_UNSET);
-
-/// Last recorded self-test outcome: `None` before the first run, then
-/// `Some(true)` on a passing check and `Some(false)` on a failed one (a
-/// skipped check never publishes, so it cannot read as a pass). The cell is
-/// last-writer-wins rather than set-once: production runs the self-test once,
-/// and last-writer-wins keeps tests that exercise both outcomes
-/// order-independent.
-pub fn bitrot_selftest_passed() -> Option<bool> {
-    match BITROT_SELF_TEST_STATUS.load(Ordering::Acquire) {
-        STATUS_CELL_UNSET => None,
-        STATUS_CELL_PASSED => Some(true),
-        STATUS_CELL_FAILED => Some(false),
-        _ => None,
-    }
-}
-
-/// Run the bitrot self-test and publish the outcome. In strict mode a failure
-/// is returned as an error so the caller aborts startup.
-pub(crate) async fn run_startup_bitrot_self_test(enabled: bool, strict: bool) -> io::Result<()> {
-    run_startup_bitrot_self_test_with(enabled, strict, bitrot_self_test).await
-}
-
-async fn run_startup_bitrot_self_test_with<F, Fut>(enabled: bool, strict: bool, run_check: F) -> io::Result<()>
-where
-    F: FnOnce() -> Fut,
-    Fut: Future<Output = Result<(), BitrotSelfTestError>>,
-{
-    if !enabled {
-        gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_SKIPPED);
-        debug!(
-            target: "rustfs::main::run",
-            event = EVENT_BITROT_SELFTEST,
-            component = LOG_COMPONENT_MAIN,
-            subsystem = LOG_SUBSYSTEM_STARTUP,
-            state = "skipped",
-            reason = "disabled",
-            "Bitrot self-test skipped"
-        );
-        return Ok(());
-    }
-
-    let started = Instant::now();
-    match run_check().await {
-        Ok(()) => {
-            BITROT_SELF_TEST_STATUS.store(STATUS_CELL_PASSED, Ordering::Release);
-            gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_PASSED);
-            info!(
-                target: "rustfs::main::run",
-                event = EVENT_BITROT_SELFTEST,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                state = "passed",
-                duration_us = started.elapsed().as_micros() as u64,
-                "Bitrot self-test passed"
-            );
-        }
-        Err(err) => {
-            BITROT_SELF_TEST_STATUS.store(STATUS_CELL_FAILED, Ordering::Release);
-            gauge!(METRIC_BITROT_SELFTEST_STATUS).set(STATUS_FAILED);
-            error!(
-                target: "rustfs::main::run",
-                event = EVENT_BITROT_SELFTEST,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                state = "failed",
-                duration_us = started.elapsed().as_micros() as u64,
-                error = %err,
-                "Bitrot self-test failed"
-            );
-            if strict {
-                return Err(io::Error::other(format!("bitrot self-test failed: {err}")));
-            }
-        }
-    }
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{BITROT_SELF_TEST_STATUS, STATUS_CELL_UNSET, bitrot_selftest_passed, run_startup_bitrot_self_test_with};
-    use crate::storage_api::startup::background::BitrotSelfTestError;
-    use std::future::ready;
-    use std::sync::atomic::Ordering;
-
-    fn failing_check() -> impl Future<Output = Result<(), BitrotSelfTestError>> {
-        ready(Err(BitrotSelfTestError::RoundtripReadback {
-            algorithm: "HighwayHash256S",
-        }))
-    }
-
-    /// All scenarios run sequentially inside one test: the status cell is
-    /// process-global, so parallel per-scenario tests would race the reset and
-    /// read each other's outcomes (the exact order-dependent flake class this
-    /// module exists to avoid).
-    #[tokio::test]
-    async fn startup_self_test_publishes_outcome_and_strict_gates_abort() {
-        BITROT_SELF_TEST_STATUS.store(STATUS_CELL_UNSET, Ordering::Release);
-
-        // Skipped: publishes nothing, never fails, never aborts.
-        run_startup_bitrot_self_test_with(false, true, || async { Ok(()) })
-            .await
-            .expect("a disabled self-test must not fail even in strict mode");
-        assert_eq!(bitrot_selftest_passed(), None, "a skipped run must leave the status unset");
-
-        // Passing: publishes Some(true), never fails.
-        run_startup_bitrot_self_test_with(true, false, || async { Ok(()) })
-            .await
-            .expect("a passing check must never fail startup");
-        assert_eq!(bitrot_selftest_passed(), Some(true), "a passing run must publish Some(true)");
-
-        // Failing, non-strict: publishes Some(false) but startup continues.
-        run_startup_bitrot_self_test_with(true, false, failing_check)
-            .await
-            .expect("a failed check must not abort startup in non-strict mode");
-        assert_eq!(bitrot_selftest_passed(), Some(false), "a failing run must publish Some(false)");
-
-        // Failing, strict: startup error carries the failure and the published
-        // outcome stays a failure.
-        let err = run_startup_bitrot_self_test_with(true, true, failing_check)
-            .await
-            .expect_err("strict mode must turn a failed check into a startup error");
-        assert!(err.to_string().contains("bitrot self-test failed"));
-        assert_eq!(bitrot_selftest_passed(), Some(false));
-    }
-}

rustfs/src/lib.rs

@@ -76,7 +76,6 @@ pub mod allocator_reclaim;
 pub mod app;
 pub mod auth;
 pub mod auth_keystone;
-pub(crate) mod bitrot_selftest;
 pub mod capacity;
 pub mod cluster_snapshot;
 pub mod config;

rustfs/src/module_switches.rs

@@ -33,8 +33,6 @@ pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
 pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
 pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
 pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
-pub(crate) const ENV_BITROT_SELFTEST_ENABLE: &str = "RUSTFS_BITROT_SELFTEST_ENABLE";
-pub(crate) const ENV_BITROT_SELFTEST_STRICT: &str = "RUSTFS_BITROT_SELFTEST_STRICT";
 
 static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
 static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
@@ -49,18 +47,6 @@ pub(crate) fn heal_enabled_from_env() -> bool {
     get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
 }
 
-/// Whether the startup bitrot algorithm self-test runs, defaulting to on
-/// (rustfs/backlog#1873).
-pub(crate) fn bitrot_selftest_enabled_from_env() -> bool {
-    rustfs_utils::get_env_bool(ENV_BITROT_SELFTEST_ENABLE, true)
-}
-
-/// Whether a failed bitrot self-test aborts startup instead of only logging
-/// and exposing a failed status, defaulting to off.
-pub(crate) fn bitrot_selftest_strict_from_env() -> bool {
-    rustfs_utils::get_env_bool(ENV_BITROT_SELFTEST_STRICT, false)
-}
-
 /// Last published audit-module state.
 pub fn is_audit_module_enabled() -> bool {
     AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)

rustfs/src/startup_background.rs

@@ -12,10 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bitrot_selftest::run_startup_bitrot_self_test;
-use crate::module_switches::{
-    bitrot_selftest_enabled_from_env, bitrot_selftest_strict_from_env, heal_enabled_from_env, scanner_enabled_from_env,
-};
+use crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
 use crate::storage_api::startup::background::{ECStore, set_workload_admission_snapshot_provider};
 use crate::workload_admission::RustFsWorkloadAdmissionSnapshotProvider;
 use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
@@ -30,12 +27,6 @@ const LOG_SUBSYSTEM_STARTUP: &str = "startup";
 const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
 
 pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
-    // Pin the bitrot algorithms before anything can write or verify a shard:
-    // the check costs well under a millisecond, and in strict mode a drifted
-    // build must abort here rather than after it has touched data
-    // (rustfs/backlog#1873).
-    run_startup_bitrot_self_test(bitrot_selftest_enabled_from_env(), bitrot_selftest_strict_from_env()).await?;
-
     let _ = create_ahm_services_cancel_token();
 
     let enable_scanner = scanner_enabled_from_env();

rustfs/src/storage/ecfs.rs

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

rustfs/src/storage/mod.rs

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

rustfs/src/storage/options.rs

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

rustfs/src/storage/storage_api.rs

@@ -569,10 +569,6 @@ pub(crate) mod ecstore_erasure {
     pub(crate) use rustfs_ecstore::api::erasure::{BitrotReader, Erasure};
 }
 
-/// Startup bitrot algorithm self-test (rustfs/backlog#1873), re-exported for
-/// the root facade's background-startup section.
-pub(crate) use rustfs_ecstore::api::erasure::{BitrotSelfTestError, bitrot_self_test};
-
 pub(crate) mod ecstore_storage {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::storage::init_local_disks;
@@ -809,6 +805,10 @@ impl StorageReplicationStatsHandle {
             proxy_head_failed: metrics.proxied.head_failed,
             proxy_put_tag_total: metrics.proxied.put_tag_total,
             proxy_put_tag_failed: metrics.proxied.put_tag_failed,
+            proxy_get_tag_total: metrics.proxied.get_tag_total,
+            proxy_get_tag_failed: metrics.proxied.get_tag_failed,
+            proxy_delete_tag_total: metrics.proxied.delete_tag_total,
+            proxy_delete_tag_failed: metrics.proxied.delete_tag_failed,
             replica_size: metrics.replica_size,
             replica_count: metrics.replica_count,
         }
@@ -845,6 +845,10 @@ pub(crate) struct ReplicationSiteMetricsSnapshot {
     pub(crate) proxy_head_failed: i64,
     pub(crate) proxy_put_tag_total: i64,
     pub(crate) proxy_put_tag_failed: i64,
+    pub(crate) proxy_get_tag_total: i64,
+    pub(crate) proxy_get_tag_failed: i64,
+    pub(crate) proxy_delete_tag_total: i64,
+    pub(crate) proxy_delete_tag_failed: i64,
     pub(crate) replica_size: i64,
     pub(crate) replica_count: i64,
 }
@@ -1491,12 +1495,6 @@ pub(crate) async fn get_bucket_object_lock_config(
     ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
 }
 
-pub(crate) async fn get_bucket_replication_config(
-    bucket: &str,
-) -> Result<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {
-    ecstore_bucket::metadata_sys::get_replication_config(bucket).await
-}
-
 pub(crate) async fn persist_force_delete_intent(
     api: Arc<ECStore>,
     entry: ecstore_bucket::replication::MrfReplicateEntry,
@@ -1842,18 +1840,6 @@ pub(crate) async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore>
     ecstore_storage::find_local_disk_by_ref(disk_ref).await
 }
 
-pub(crate) trait StorageReplicationConfigExt {
-    fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
-}
-
-impl StorageReplicationConfigExt for s3s::dto::ReplicationConfiguration {
-    fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
-        <s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_active_rules(
-            self, prefix, recursive,
-        )
-    }
-}
-
 pub(crate) trait StorageVersioningConfigExt {
     fn enabled(&self) -> bool;
 }

rustfs/src/storage_api.rs

@@ -214,9 +214,7 @@ pub(crate) mod startup {
     }
 
     pub(crate) mod background {
-        pub(crate) use crate::storage::storage_api::{
-            BitrotSelfTestError, ECStore, bitrot_self_test, set_workload_admission_snapshot_provider,
-        };
+        pub(crate) use crate::storage::storage_api::{ECStore, set_workload_admission_snapshot_provider};
     }
 
     pub(crate) mod bucket_metadata {