test(e2e): rename stall timing variable flagged by typos

Add ListObjectsV2 paging, Range GET/HEAD, unversioned buckets, standard
and user metadata replay, ResponseStatus/TruncateBodyAt/Stall fault
actions, Range/User-Agent/prefix/continuation-token journal fields,
count_requests, direct seeding, and a configurable object cap to the
programmable fake S3 target, and add the on_demand_migration e2e
harness (OdmTestEnv, admin wrappers, source seeding, local-state
assertions, second RustFS source) with its self-test.

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
 sha256-darwin=9dccb0cd537cf79ae70c1c20e8281d36d03f2f09f81142a5341e26e3dc18709d
-sha256-linux=86e69337ad1440252a2ee20a12063c989ed12442d3b1ddf9e9233acf0f2ec089
+sha256-linux=a8a816d7bb0e7cb5632b1863b33794bcb9fc7e765f150aa5e1bf16518e28dfb4

Cargo.lock

@@ -10449,7 +10449,6 @@ dependencies = [
  "base64-simd",
  "bytes",
  "crc-fast",
- "criterion",
  "faster-hex",
  "futures",
  "hex-simd",
@@ -10461,7 +10460,6 @@ dependencies = [
  "md-5 0.11.0",
  "minlz",
  "pin-project-lite",
- "proptest",
  "rand 0.10.2",
  "reqwest",
  "rustfs-config",

crates/e2e_test/src/on_demand_migration/harness_self_test.rs

@@ -452,9 +452,9 @@ async fn fake_source_fault_actions_truncate_stall_and_status() -> TestResult {
     let stalled = client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
     assert!(started.elapsed() >= Duration::from_millis(350), "stall must delay the first byte");
     assert_eq!(stalled.content_length(), Some(4096));
-    let unstalled_started = Instant::now();
+    let post_stall_started = Instant::now();
     client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
-    assert!(unstalled_started.elapsed() < Duration::from_millis(350), "stall is consumed once");
+    assert!(post_stall_started.elapsed() < Duration::from_millis(350), "stall is consumed once");
 
     // The object is intact once the script is drained.
     let intact = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;

crates/ecstore/src/api/mod.rs

@@ -166,16 +166,6 @@ pub mod bucket {
         pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config};
     }
 
-    pub mod on_demand_migration {
-        pub mod source_client {
-            pub use crate::bucket::on_demand_migration::source_client::{
-                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,
-                SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
-                resolve_path_style,
-            };
-        }
-    }
-
     pub mod object_lock {
         pub use crate::bucket::object_lock::{ObjectLockApi, ObjectLockStatusExt};
 
@@ -209,13 +199,6 @@ pub mod bucket {
         }
     }
 
-    pub mod remote_s3_client {
-        pub use crate::bucket::remote_s3_client::{
-            PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, build_remote_s3_client,
-            validate_remote_endpoint,
-        };
-    }
-
     pub mod replication {
         pub use crate::bucket::replication::replication_pool::{
             DurableMrfBacklogSummary, DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBacklogObservabilitySummary,

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -15,13 +15,17 @@
 use crate::bucket::metadata::BucketMetadata;
 use crate::bucket::metadata_sys::get_bucket_targets_config;
 use crate::bucket::metadata_sys::get_replication_config;
-use crate::bucket::remote_s3_client::{PathStyle, RemoteCredentials, RemoteS3EndpointSpec, build_remote_s3_client};
 use crate::bucket::replication::{ReplicationStatusType, ReplicationTargetConfigBridge};
 use crate::bucket::target::ARN;
 use crate::bucket::target::BucketTargetType;
 use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials};
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::runtime::sources as runtime_sources;
+use aws_credential_types::Credentials as SdkCredentials;
+use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future};
+use aws_sdk_s3::config::Region as SdkRegion;
+use aws_sdk_s3::config::RequestChecksumCalculation;
+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;
@@ -33,17 +37,28 @@ 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::BucketVersioningStatus;
 use aws_sdk_s3::types::Tagging as SdkTagging;
 use aws_sdk_s3::types::{
     ChecksumMode, CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockRetentionMode,
     ServerSideEncryption,
 };
-use aws_sdk_s3::{Client as S3Client, operation::head_object::HeadObjectOutput};
-use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
+use aws_sdk_s3::{Client as S3Client, Config as S3Config, operation::head_object::HeadObjectOutput};
+use aws_sdk_s3::{config::SharedCredentialsProvider, types::BucketVersioningStatus};
+use aws_smithy_http_client::{Builder as SmithyHttpClientBuilder, tls as smithy_tls};
+use aws_smithy_runtime_api::box_error::BoxError;
+use aws_smithy_runtime_api::client::http::{
+    HttpConnector as SmithyHttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn,
+};
+use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
+use aws_smithy_runtime_api::client::result::ConnectorError;
+use aws_smithy_types::body::SdkBody;
 use futures::{StreamExt, stream};
-use http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
+use http::{HeaderMap, HeaderName, HeaderValue, StatusCode, Uri};
+use hyper_util::client::legacy::Client as HyperClient;
+use hyper_util::rt::{TokioExecutor, TokioTimer};
 use reqwest::Client as HttpClient;
+use rustfs_config::{DEFAULT_TRUST_LEAF_CERT_AS_CA, ENV_TRUST_LEAF_CERT_AS_CA, RUSTFS_CA_CERT, RUSTFS_TLS_CERT};
+use rustfs_utils::egress::{OutboundUrlError, validate_outbound_url};
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE,
     AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, AMZ_OBJECT_TAGGING_LOWER, AMZ_STORAGE_CLASS, AMZ_WEBSITE_REDIRECT_LOCATION, is_amz_header,
@@ -55,10 +70,12 @@ use rustfs_utils::http::{
     SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
     insert_header,
 };
+use rustls_pki_types::pem::PemObject;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt;
+use std::path::Path;
 use std::str::FromStr as _;
 use std::sync::Arc;
 use std::sync::OnceLock;
@@ -67,6 +84,7 @@ use std::time::{Duration, Instant, SystemTime};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tokio::sync::Mutex;
 use tokio::sync::RwLock;
+use tower::Service;
 use tracing::error;
 use tracing::warn;
 use url::Url;
@@ -74,50 +92,72 @@ use uuid::Uuid;
 
 const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
+const EXPIRED_REMOTE_TARGET_CREDENTIALS: &str = "remote target credentials have expired";
 
-fn remote_credentials(credentials: &Credentials, account_id: &str) -> RemoteCredentials {
-    RemoteCredentials {
-        access_key: credentials.access_key.clone(),
-        secret_key: credentials.secret_key.clone(),
-        session_token: credentials.effective_session_token().map(str::to_string),
-        expiration: credentials.effective_expiration().map(SystemTime::from),
-        account_id: account_id.to_string(),
-    }
+#[derive(Clone)]
+struct RemoteTargetCredentialsProvider {
+    credentials: SdkCredentials,
 }
 
-fn target_path_style(path: &str) -> PathStyle {
-    match path.trim().to_ascii_lowercase().as_str() {
-        // Explicit DNS/virtual-hosted-style requested by user.
-        "dns" | "off" | "false" => PathStyle::VirtualHost,
-        // Explicit path-style or legacy boolean-like values.
-        "path" | "on" | "true" => PathStyle::Path,
-        // `auto` and empty are defaulted to path-style for custom S3-compatible endpoints.
-        "auto" | "" => PathStyle::Auto,
-        // Unknown values: prefer compatibility with S3-compatible services.
-        _ => PathStyle::Path,
-    }
-}
-
-impl From<&BucketTarget> for RemoteS3EndpointSpec {
-    fn from(target: &BucketTarget) -> Self {
-        RemoteS3EndpointSpec {
-            endpoint: target.endpoint.clone(),
-            secure: target.secure,
-            region: target.region.clone(),
-            path_style: target_path_style(&target.path),
-            credentials: target
-                .credentials
-                .as_ref()
-                .map(|credentials| remote_credentials(credentials, &target.reset_id)),
-            skip_tls_verify: target.skip_tls_verify,
-            ca_cert_pem: (!target.ca_cert_pem.trim().is_empty()).then(|| target.ca_cert_pem.clone()),
-            connect_timeout: None,
-            read_timeout: None,
-            user_agent_suffix: "",
+impl RemoteTargetCredentialsProvider {
+    fn resolve_at(&self, now: SystemTime) -> aws_credential_types::provider::Result {
+        if self.credentials.expiry().is_some_and(|expiration| expiration <= now) {
+            return Err(CredentialsError::provider_error(std::io::Error::other(EXPIRED_REMOTE_TARGET_CREDENTIALS)));
         }
+        Ok(self.credentials.clone())
     }
 }
 
+impl fmt::Debug for RemoteTargetCredentialsProvider {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("RemoteTargetCredentialsProvider")
+            .field("temporary", &self.credentials.session_token().is_some())
+            .field("expiration", &self.credentials.expiry())
+            .finish()
+    }
+}
+
+impl ProvideCredentials for RemoteTargetCredentialsProvider {
+    fn provide_credentials<'a>(&'a self) -> future::ProvideCredentials<'a>
+    where
+        Self: 'a,
+    {
+        future::ProvideCredentials::ready(self.resolve_at(SystemTime::now()))
+    }
+
+    fn fallback_on_interrupt(&self) -> Option<SdkCredentials> {
+        self.resolve_at(SystemTime::now()).ok()
+    }
+}
+
+fn remote_target_sdk_credentials(
+    credentials: &Credentials,
+    account_id: &str,
+    now: SystemTime,
+) -> Result<SdkCredentials, &'static str> {
+    let session_token = credentials.effective_session_token();
+    let expiration = credentials.effective_expiration().map(SystemTime::from);
+    if expiration.is_some() && session_token.is_none() {
+        return Err("remote target credential expiration requires a session token");
+    }
+    if expiration.is_some_and(|expiration| expiration <= now) {
+        return Err(EXPIRED_REMOTE_TARGET_CREDENTIALS);
+    }
+
+    let mut builder = SdkCredentials::builder()
+        .access_key_id(credentials.access_key.clone())
+        .secret_access_key(credentials.secret_key.clone())
+        .account_id(account_id.to_string())
+        .provider_name("bucket_target_sys");
+    if let Some(session_token) = session_token {
+        builder = builder.session_token(session_token.to_string());
+    }
+    if let Some(expiration) = expiration {
+        builder = builder.expiry(expiration);
+    }
+    Ok(builder.build())
+}
+
 pub type HeadObjectSdkError = Box<SdkError<HeadObjectError>>;
 pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
 pub type GetObjectTaggingSdkError = Box<SdkError<GetObjectTaggingError>>;
@@ -1018,17 +1058,57 @@ impl BucketTargetSys {
             });
         };
 
-        let spec = RemoteS3EndpointSpec::from(target);
-        let client = build_remote_s3_client(&spec)
-            .await
-            .map_err(|err| BucketTargetError::RemoteTargetConnectionErr {
+        let creds = remote_target_sdk_credentials(credentials, &target.reset_id, SystemTime::now()).map_err(|error| {
+            BucketTargetError::RemoteTargetConnectionErr {
                 bucket: target.target_bucket.clone(),
                 access_key: credentials.access_key.clone(),
-                error: err.to_string(),
-            })?;
+                error: error.to_string(),
+            }
+        })?;
+
+        let endpoint = if target.secure {
+            format!("https://{}", target.endpoint)
+        } else {
+            format!("http://{}", target.endpoint)
+        };
+        let parsed_endpoint = Url::parse(&endpoint).map_err(|err| BucketTargetError::RemoteTargetConnectionErr {
+            bucket: target.target_bucket.clone(),
+            access_key: credentials.access_key.clone(),
+            error: format!("invalid target endpoint: {err}"),
+        })?;
+        validate_replication_target_endpoint(&parsed_endpoint).map_err(|err| BucketTargetError::RemoteTargetConnectionErr {
+            bucket: target.target_bucket.clone(),
+            access_key: credentials.access_key.clone(),
+            error: format!("target endpoint is not allowed: {err}"),
+        })?;
+
+        let mut config_builder = S3Config::builder()
+            .endpoint_url(endpoint.clone())
+            .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds }))
+            .region(SdkRegion::new(target.region.clone()))
+            .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+            .request_checksum_calculation(replication_request_checksum_calculation());
+
+        if should_force_path_style(target) {
+            config_builder = config_builder.force_path_style(true);
+        }
+
+        if let Some(http_client) =
+            build_aws_s3_http_client_for_target(target)
+                .await
+                .map_err(|err| BucketTargetError::RemoteTargetConnectionErr {
+                    bucket: target.target_bucket.clone(),
+                    access_key: credentials.access_key.clone(),
+                    error: err.to_string(),
+                })?
+        {
+            config_builder = config_builder.http_client(http_client);
+        }
+
+        let config = config_builder.build();
 
         Ok(TargetClient {
-            endpoint: spec.endpoint_url(),
+            endpoint,
             credentials: target.credentials.clone(),
             bucket: target.target_bucket.clone(),
             storage_class: target.storage_class.clone(),
@@ -1038,7 +1118,7 @@ impl BucketTargetSys {
             secure: target.secure,
             health_check_duration: target.health_check_duration,
             replicate_sync: target.replication_sync,
-            client: Arc::new(client),
+            client: Arc::new(S3Client::from_conf(config)),
         })
     }
 
@@ -1201,6 +1281,327 @@ impl BucketTargetSys {
     }
 }
 
+#[derive(Debug)]
+struct AcceptAnyServerCertVerifier;
+
+impl rustls::client::danger::ServerCertVerifier for AcceptAnyServerCertVerifier {
+    fn verify_server_cert(
+        &self,
+        _end_entity: &rustls_pki_types::CertificateDer<'_>,
+        _intermediates: &[rustls_pki_types::CertificateDer<'_>],
+        _server_name: &rustls_pki_types::ServerName<'_>,
+        _ocsp_response: &[u8],
+        _now: rustls_pki_types::UnixTime,
+    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
+        Ok(rustls::client::danger::ServerCertVerified::assertion())
+    }
+
+    fn verify_tls12_signature(
+        &self,
+        _message: &[u8],
+        _cert: &rustls_pki_types::CertificateDer<'_>,
+        _dss: &rustls::DigitallySignedStruct,
+    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
+        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
+    }
+
+    fn verify_tls13_signature(
+        &self,
+        _message: &[u8],
+        _cert: &rustls_pki_types::CertificateDer<'_>,
+        _dss: &rustls::DigitallySignedStruct,
+    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
+        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
+    }
+
+    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
+        rustls::crypto::aws_lc_rs::default_provider()
+            .signature_verification_algorithms
+            .supported_schemes()
+    }
+}
+
+#[derive(Clone)]
+struct TargetHyperHttpConnector<C> {
+    client: HyperClient<C, SdkBody>,
+}
+
+impl<C> fmt::Debug for TargetHyperHttpConnector<C> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("TargetHyperHttpConnector")
+            .field("client", &"** hyper client **")
+            .finish()
+    }
+}
+
+impl<C> SmithyHttpConnector for TargetHyperHttpConnector<C>
+where
+    C: Clone + Send + Sync + 'static,
+    C: Service<Uri>,
+    C::Response:
+        hyper::rt::Read + hyper::rt::Write + hyper_util::client::legacy::connect::Connection + Send + Sync + Unpin + 'static,
+    C::Future: Unpin + Send + 'static,
+    C::Error: Into<BoxError>,
+{
+    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
+        let request = match request.try_into_http1x() {
+            Ok(request) => request,
+            Err(err) => return HttpConnectorFuture::ready(Err(ConnectorError::user(err.into()))),
+        };
+
+        let mut client = self.client.clone();
+        let fut = client.call(request);
+        HttpConnectorFuture::new(async move {
+            let response = fut
+                .await
+                .map_err(|err| ConnectorError::io(err.into()))?
+                .map(SdkBody::from_body_1_x);
+            HttpResponse::try_from(response).map_err(|err| ConnectorError::other(err.into(), None))
+        })
+    }
+}
+
+fn ensure_rustls_crypto_provider() {
+    if rustls::crypto::CryptoProvider::get_default().is_none() {
+        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+    }
+}
+
+fn has_custom_ca_pem(target: &BucketTarget) -> bool {
+    !target.ca_cert_pem.trim().is_empty()
+}
+
+/// Env opt-in that re-enables loopback replication targets. Loopback (`127.0.0.1`,
+/// `::1`, `localhost`) is a classic SSRF vector and stays rejected by default, but
+/// single-host multi-instance dev setups and the e2e harness legitimately replicate
+/// over loopback. Never set this in production.
+const ALLOW_LOOPBACK_REPLICATION_TARGET_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET";
+
+fn loopback_replication_targets_allowed() -> bool {
+    std::env::var(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV)
+        .map(|v| v.eq_ignore_ascii_case("true") || v == "1")
+        .unwrap_or(false)
+}
+
+const REPLICATION_STREAMING_CHECKSUMS_ENV: &str = "RUSTFS_REPLICATION_STREAMING_CHECKSUMS";
+
+/// Streaming trailer checksums make the SDK frame request bodies as
+/// `aws-chunked`; a target that does not decode that framing stores the frames
+/// verbatim, silently corrupting every replica while the transfer itself
+/// succeeds (#6853). Plain signed payloads are the compatible default; the env
+/// knob restores trailer checksums for fleets whose targets are all known to
+/// decode them.
+fn replication_request_checksum_calculation() -> RequestChecksumCalculation {
+    if std::env::var(REPLICATION_STREAMING_CHECKSUMS_ENV)
+        .map(|v| v.eq_ignore_ascii_case("true") || v == "1")
+        .unwrap_or(false)
+    {
+        RequestChecksumCalculation::WhenSupported
+    } else {
+        RequestChecksumCalculation::WhenRequired
+    }
+}
+
+fn validate_replication_target_endpoint(url: &Url) -> Result<(), OutboundUrlError> {
+    validate_replication_target_endpoint_inner(url, loopback_replication_targets_allowed())
+}
+
+fn validate_replication_target_endpoint_inner(url: &Url, allow_loopback: bool) -> Result<(), OutboundUrlError> {
+    match validate_outbound_url(url) {
+        Ok(()) => Ok(()),
+        // Replication targets are trusted infrastructure the operator configures, and
+        // legitimately live on private networks, so private addresses are always allowed.
+        Err(OutboundUrlError::ForbiddenHost {
+            reason: "private address",
+            ..
+        }) => Ok(()),
+        // Loopback is far higher SSRF risk, so it is allowed only under the explicit,
+        // off-by-default opt-in above (single-host multi-instance / the e2e harness).
+        Err(OutboundUrlError::ForbiddenHost {
+            reason: "loopback address" | "loopback host",
+            ..
+        }) if allow_loopback => Ok(()),
+        Err(err) => Err(err),
+    }
+}
+
+fn build_insecure_aws_s3_http_client() -> SharedHttpClient {
+    ensure_rustls_crypto_provider();
+
+    let tls_config = rustls::ClientConfig::builder()
+        .dangerous()
+        .with_custom_certificate_verifier(Arc::new(AcceptAnyServerCertVerifier))
+        .with_no_client_auth();
+
+    let https = hyper_rustls::HttpsConnectorBuilder::new()
+        .with_tls_config(tls_config)
+        .https_or_http()
+        .enable_http1()
+        .enable_http2()
+        .build();
+    let mut client_builder = HyperClient::builder(TokioExecutor::new());
+    client_builder.pool_timer(TokioTimer::new());
+    let client = client_builder.build(https);
+    let connector = SharedHttpConnector::new(TargetHyperHttpConnector { client });
+
+    http_client_fn(move |_settings, _components| connector.clone())
+}
+
+fn validate_ca_pem_bundle(ca_cert_pem: &[u8]) -> Result<(), String> {
+    let certs = rustls_pki_types::CertificateDer::pem_slice_iter(ca_cert_pem)
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|err| format!("invalid PEM encoding: {err}"))?;
+
+    if certs.is_empty() {
+        return Err("no certificates found".to_string());
+    }
+
+    // Smithy's rustls adapter defers parsing custom certificates and assumes
+    // they are valid when the HTTPS connector is built. Validate every DER
+    // certificate first so malformed configuration is reported rather than
+    // reaching an `expect` in the dependency.
+    let mut validation_store = rustls::RootCertStore::empty();
+    for cert in certs {
+        validation_store
+            .add(cert)
+            .map_err(|err| format!("invalid X.509 certificate: {err}"))?;
+    }
+
+    Ok(())
+}
+
+fn validate_target_ca_pem(ca_cert_pem: &str) -> Result<(), BucketTargetError> {
+    validate_ca_pem_bundle(ca_cert_pem.as_bytes())
+        .map_err(|err| BucketTargetError::Io(std::io::Error::other(format!("invalid target CA PEM: {err}"))))
+}
+
+fn compose_replication_trust_store(certificate_bundles: impl IntoIterator<Item = Vec<u8>>) -> (smithy_tls::TrustStore, usize) {
+    // `TrustStore::default()` keeps the platform-native roots enabled. Target
+    // and RUSTFS_TLS_PATH certificates extend that baseline instead of
+    // replacing it with a target-specific trust island.
+    let mut trust_store = smithy_tls::TrustStore::default();
+    let mut custom_bundle_count = 0;
+    for pem in certificate_bundles {
+        trust_store.add_pem_certificate(pem);
+        custom_bundle_count += 1;
+    }
+
+    (trust_store, custom_bundle_count)
+}
+
+fn build_aws_s3_http_client_with_trust_store(trust_store: smithy_tls::TrustStore) -> Result<SharedHttpClient, BucketTargetError> {
+    let tls_context = smithy_tls::TlsContext::builder()
+        .with_trust_store(trust_store)
+        .build()
+        .map_err(|err| BucketTargetError::Io(std::io::Error::other(format!("invalid target CA PEM: {err}"))))?;
+
+    Ok(SmithyHttpClientBuilder::new()
+        .tls_provider(smithy_tls::Provider::rustls(smithy_tls::rustls_provider::CryptoMode::AwsLc))
+        .tls_context(tls_context)
+        .build_https())
+}
+
+async fn load_tls_path_ca_bundles(tls_dir: &Path, trust_leaf_cert_as_ca: bool) -> Vec<Vec<u8>> {
+    let mut certificate_bundles = Vec::new();
+
+    let ca_path = tls_dir.join(RUSTFS_CA_CERT);
+    match tokio::fs::read(&ca_path).await {
+        Ok(pem) => match validate_ca_pem_bundle(&pem) {
+            Ok(()) => certificate_bundles.push(pem),
+            Err(err) => warn!("ignoring invalid custom CA bundle {:?} for replication client: {}", ca_path, err),
+        },
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+        Err(e) => warn!("failed to read custom CA bundle {:?} for replication client: {}", ca_path, e),
+    }
+
+    if trust_leaf_cert_as_ca {
+        let leaf_cert_path = tls_dir.join(RUSTFS_TLS_CERT);
+        match tokio::fs::read(&leaf_cert_path).await {
+            Ok(pem) => match validate_ca_pem_bundle(&pem) {
+                Ok(()) => certificate_bundles.push(pem),
+                Err(err) => warn!(
+                    "ignoring invalid leaf certificate {:?} for replication client trust store: {}",
+                    leaf_cert_path, err
+                ),
+            },
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+            Err(e) => warn!("failed to read leaf cert {:?} for replication client trust store: {}", leaf_cert_path, e),
+        }
+    }
+
+    certificate_bundles
+}
+
+async fn load_configured_tls_ca_bundles() -> Vec<Vec<u8>> {
+    let tls_path = rustfs_utils::get_env_str(rustfs_config::ENV_RUSTFS_TLS_PATH, rustfs_config::DEFAULT_RUSTFS_TLS_PATH);
+    if tls_path.is_empty() {
+        return Vec::new();
+    }
+
+    load_tls_path_ca_bundles(
+        Path::new(&tls_path),
+        rustfs_utils::get_env_bool(ENV_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_LEAF_CERT_AS_CA),
+    )
+    .await
+}
+
+async fn build_aws_s3_http_client_from_target_ca_pem(ca_cert_pem: &str) -> Result<SharedHttpClient, BucketTargetError> {
+    validate_target_ca_pem(ca_cert_pem)?;
+
+    let mut certificate_bundles = load_configured_tls_ca_bundles().await;
+    certificate_bundles.push(ca_cert_pem.as_bytes().to_vec());
+    let (trust_store, _) = compose_replication_trust_store(certificate_bundles);
+
+    build_aws_s3_http_client_with_trust_store(trust_store)
+}
+
+async fn build_aws_s3_http_client_for_target(target: &BucketTarget) -> Result<Option<SharedHttpClient>, BucketTargetError> {
+    if !target.secure {
+        return Ok(None);
+    }
+
+    if target.skip_tls_verify {
+        return Ok(Some(build_insecure_aws_s3_http_client()));
+    }
+
+    if has_custom_ca_pem(target) {
+        return build_aws_s3_http_client_from_target_ca_pem(&target.ca_cert_pem)
+            .await
+            .map(Some);
+    }
+
+    Ok(build_aws_s3_http_client_from_tls_path().await)
+}
+
+async fn build_aws_s3_http_client_from_tls_path() -> Option<aws_sdk_s3::config::SharedHttpClient> {
+    let certificate_bundles = load_configured_tls_ca_bundles().await;
+    if certificate_bundles.is_empty() {
+        return None;
+    }
+
+    let (trust_store, _) = compose_replication_trust_store(certificate_bundles);
+    match build_aws_s3_http_client_with_trust_store(trust_store) {
+        Ok(client) => Some(client),
+        Err(e) => {
+            warn!("failed to build AWS SDK TLS context for replication client: {}", e);
+            None
+        }
+    }
+}
+
+fn should_force_path_style(target: &BucketTarget) -> bool {
+    match target.path.trim().to_ascii_lowercase().as_str() {
+        // Explicit DNS/virtual-hosted-style requested by user.
+        "dns" | "off" | "false" => false,
+        // Explicit path-style or legacy boolean-like values.
+        "path" | "on" | "true" => true,
+        // `auto` and empty are defaulted to path-style for custom S3-compatible endpoints.
+        "auto" | "" => true,
+        // Unknown values: prefer compatibility with S3-compatible services.
+        _ => true,
+    }
+}
+
 // generate ARN that is unique to this target type
 fn generate_arn(t: &BucketTarget, depl_id: &str) -> String {
     let uuid = if depl_id.is_empty() {
@@ -2306,24 +2707,7 @@ impl Error for BucketTargetError {}
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::bucket::remote_s3_client::{
-        EXPIRED_REMOTE_TARGET_CREDENTIALS, RemoteTargetCredentialsProvider, build_aws_s3_http_client_for_spec,
-        build_aws_s3_http_client_from_target_ca_pem, build_aws_s3_http_client_with_trust_store,
-        build_insecure_aws_s3_http_client, compose_replication_trust_store, ensure_rustls_crypto_provider,
-        load_tls_path_ca_bundles, remote_sdk_credentials, replication_request_checksum_calculation,
-        validate_remote_endpoint_inner, validate_target_ca_pem,
-    };
-    use aws_credential_types::Credentials as SdkCredentials;
-    use aws_sdk_s3::Config as S3Config;
-    use aws_sdk_s3::config::{Region as SdkRegion, RequestChecksumCalculation, SharedCredentialsProvider, SharedHttpClient};
-    use aws_smithy_runtime_api::client::http::{
-        HttpConnector as SmithyHttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn,
-    };
-    use aws_smithy_runtime_api::client::orchestrator::HttpResponse;
-    use aws_smithy_types::body::SdkBody;
     use rcgen::generate_simple_self_signed;
-    use rustfs_config::{RUSTFS_CA_CERT, RUSTFS_TLS_CERT};
-    use rustfs_utils::egress::OutboundUrlError;
 
     // The startup panic fix for hosts without a CA bundle (issue #6734) rests
     // on two properties: the health-check client constructor never panics, and
@@ -2550,8 +2934,8 @@ mod tests {
             expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")),
         };
 
-        let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, "account"), now)
-            .expect("unexpired temporary credentials should build");
+        let sdk_credentials =
+            remote_target_sdk_credentials(&credentials, "account", now).expect("unexpired temporary credentials should build");
 
         assert_eq!(sdk_credentials.session_token(), Some("temporary-session-token"));
         assert_eq!(sdk_credentials.expiry(), Some(expiration));
@@ -2567,7 +2951,7 @@ mod tests {
             expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")),
         };
 
-        let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::now())
+        let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now())
             .expect("Go zero expiration should remain compatible with static credentials");
 
         assert!(sdk_credentials.session_token().is_none());
@@ -2585,14 +2969,14 @@ mod tests {
         };
 
         assert_eq!(
-            remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::UNIX_EPOCH + Duration::from_secs(1_000))
+            remote_target_sdk_credentials(&credentials, "", SystemTime::UNIX_EPOCH + Duration::from_secs(1_000))
                 .expect_err("expiration without a session token must fail"),
             "remote target credential expiration requires a session token"
         );
 
         credentials.session_token = Some("temporary-session-token".to_string());
         assert_eq!(
-            remote_sdk_credentials(&remote_credentials(&credentials, ""), expiration)
+            remote_target_sdk_credentials(&credentials, "", expiration)
                 .expect_err("credentials expire at the exact expiration boundary"),
             EXPIRED_REMOTE_TARGET_CREDENTIALS
         );
@@ -2652,7 +3036,7 @@ mod tests {
             session_token: Some("temporary-session-token".to_string()),
             expiration: Some("2099-01-01T00:00:00Z".parse().expect("future expiration should parse")),
         };
-        let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::now())
+        let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now())
             .expect("unexpired temporary credentials should build");
         let client = S3Client::from_conf(
             S3Config::builder()
@@ -2827,46 +3211,6 @@ mod tests {
         assert!(!replication_target_versioning_enabled(None));
     }
 
-    #[test]
-    fn remote_endpoint_spec_from_target_keeps_legacy_path_style_and_trust_semantics() {
-        for (path, expected) in [
-            ("dns", PathStyle::VirtualHost),
-            ("OFF", PathStyle::VirtualHost),
-            ("false", PathStyle::VirtualHost),
-            ("path", PathStyle::Path),
-            ("on", PathStyle::Path),
-            ("true", PathStyle::Path),
-            (" auto ", PathStyle::Auto),
-            ("", PathStyle::Auto),
-            ("something-else", PathStyle::Path),
-        ] {
-            assert_eq!(target_path_style(path), expected, "path={path:?}");
-        }
-
-        let spec = RemoteS3EndpointSpec::from(&BucketTarget {
-            endpoint: "192.168.1.10:9000".to_string(),
-            secure: true,
-            region: "us-east-1".to_string(),
-            ca_cert_pem: "   ".to_string(),
-            reset_id: "reset-1".to_string(),
-            credentials: Some(Credentials {
-                access_key: "access".to_string(),
-                secret_key: "secret".to_string(),
-                session_token: Some("  ".to_string()),
-                expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")),
-            }),
-            ..Default::default()
-        });
-        assert_eq!(spec.endpoint_url(), "https://192.168.1.10:9000");
-        assert!(spec.ca_cert_pem.is_none(), "whitespace-only CA PEM means unset");
-        assert!(spec.connect_timeout.is_none() && spec.read_timeout.is_none());
-        assert_eq!(spec.user_agent_suffix, "");
-        let credentials = spec.credentials.expect("credentials carry over");
-        assert_eq!(credentials.account_id, "reset-1");
-        assert!(credentials.session_token.is_none(), "blank session token is absent");
-        assert!(credentials.expiration.is_none(), "Go zero expiration is absent");
-    }
-
     fn parse_url(raw: &str) -> Url {
         Url::parse(raw).expect("test URL should parse")
     }
@@ -2876,16 +3220,16 @@ mod tests {
         // Public hosts and private-network targets are allowed regardless of the
         // loopback opt-in — replication commonly runs across trusted private infra.
         for allow_loopback in [false, true] {
-            assert!(validate_remote_endpoint_inner(&parse_url("https://s3.example.com"), allow_loopback).is_ok());
-            assert!(validate_remote_endpoint_inner(&parse_url("http://10.0.0.5:9000"), allow_loopback).is_ok());
-            assert!(validate_remote_endpoint_inner(&parse_url("http://192.168.1.20"), allow_loopback).is_ok());
+            assert!(validate_replication_target_endpoint_inner(&parse_url("https://s3.example.com"), allow_loopback).is_ok());
+            assert!(validate_replication_target_endpoint_inner(&parse_url("http://10.0.0.5:9000"), allow_loopback).is_ok());
+            assert!(validate_replication_target_endpoint_inner(&parse_url("http://192.168.1.20"), allow_loopback).is_ok());
         }
     }
 
     #[test]
     fn replication_endpoint_rejects_loopback_without_opt_in() {
         // Default (production) behaviour: loopback IP and localhost host both rejected.
-        let err = validate_remote_endpoint_inner(&parse_url("http://127.0.0.1:9000"), false)
+        let err = validate_replication_target_endpoint_inner(&parse_url("http://127.0.0.1:9000"), false)
             .expect_err("loopback IP must be rejected by default");
         assert!(matches!(
             err,
@@ -2894,7 +3238,7 @@ mod tests {
                 ..
             }
         ));
-        let err = validate_remote_endpoint_inner(&parse_url("http://localhost:9000"), false)
+        let err = validate_replication_target_endpoint_inner(&parse_url("http://localhost:9000"), false)
             .expect_err("localhost must be rejected by default");
         assert!(matches!(
             err,
@@ -2909,15 +3253,15 @@ mod tests {
     fn replication_endpoint_allows_loopback_with_opt_in() {
         // e2e harness / single-host multi-instance: opt-in re-enables loopback in
         // both IP (127.0.0.1, ::1) and hostname (localhost) forms.
-        assert!(validate_remote_endpoint_inner(&parse_url("http://127.0.0.1:9000"), true).is_ok());
-        assert!(validate_remote_endpoint_inner(&parse_url("http://[::1]:9000"), true).is_ok());
-        assert!(validate_remote_endpoint_inner(&parse_url("http://localhost:9000"), true).is_ok());
+        assert!(validate_replication_target_endpoint_inner(&parse_url("http://127.0.0.1:9000"), true).is_ok());
+        assert!(validate_replication_target_endpoint_inner(&parse_url("http://[::1]:9000"), true).is_ok());
+        assert!(validate_replication_target_endpoint_inner(&parse_url("http://localhost:9000"), true).is_ok());
     }
 
     #[test]
     fn replication_endpoint_opt_in_does_not_open_other_ssrf_targets() {
         // The loopback opt-in must not widen into link-local / metadata endpoints.
-        let err = validate_remote_endpoint_inner(&parse_url("http://169.254.169.254/latest/meta-data"), true)
+        let err = validate_replication_target_endpoint_inner(&parse_url("http://169.254.169.254/latest/meta-data"), true)
             .expect_err("metadata endpoint must stay rejected even with loopback opt-in");
         assert!(matches!(
             err,
@@ -2926,7 +3270,7 @@ mod tests {
                 ..
             }
         ));
-        let err = validate_remote_endpoint_inner(&parse_url("http://[fe80::1]:9000"), true)
+        let err = validate_replication_target_endpoint_inner(&parse_url("http://[fe80::1]:9000"), true)
             .expect_err("link-local must stay rejected even with loopback opt-in");
         assert!(matches!(
             err,
@@ -3935,12 +4279,12 @@ mod tests {
 
     #[tokio::test]
     async fn skip_tls_verify_takes_priority_over_invalid_custom_ca_pem() {
-        let client = build_aws_s3_http_client_for_spec(&RemoteS3EndpointSpec::from(&BucketTarget {
+        let client = build_aws_s3_http_client_for_target(&BucketTarget {
             secure: true,
             skip_tls_verify: true,
             ca_cert_pem: "not a pem".to_string(),
             ..Default::default()
-        }))
+        })
         .await
         .expect("skip verification should bypass custom CA parsing");
 

crates/ecstore/src/bucket/mod.rs

@@ -26,10 +26,8 @@ mod metadata_test;
 pub mod migration;
 mod msgp_decode;
 pub mod object_lock;
-pub mod on_demand_migration;
 pub mod policy_sys;
 pub mod quota;
-pub mod remote_s3_client;
 pub mod replication;
 pub mod tagging;
 pub mod target;

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

@@ -1,18 +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.
-
-//! On-demand migration (ODM): serve and back-fill objects from an external
-//! S3-compatible source bucket.
-
-pub mod source_client;

crates/ecstore/src/bucket/remote_s3_client.rs

@@ -1,789 +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.
-
-//! Shared builder for outbound `aws_sdk_s3::Client`s.
-//!
-//! Replication targets (`bucket_target_sys`) and the on-demand migration
-//! source client build their remote clients from one neutral
-//! [`RemoteS3EndpointSpec`]: endpoint assembly, credential handling, path-style
-//! selection, custom CA / skip-TLS transports and the outbound SSRF gate all
-//! live here so both callers share exactly one policy. The gate keeps the
-//! relaxed replication semantics documented in
-//! `docs/operations/outbound-connection-policy.md`: private addresses are
-//! always allowed, loopback only behind `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET`.
-
-use aws_credential_types::Credentials as SdkCredentials;
-use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future};
-use aws_sdk_s3::config::Region as SdkRegion;
-use aws_sdk_s3::config::RequestChecksumCalculation;
-use aws_sdk_s3::config::SharedCredentialsProvider;
-use aws_sdk_s3::config::SharedHttpClient;
-use aws_sdk_s3::{Client as S3Client, Config as S3Config};
-use aws_smithy_http_client::{Builder as SmithyHttpClientBuilder, tls as smithy_tls};
-use aws_smithy_runtime_api::box_error::BoxError;
-use aws_smithy_runtime_api::client::http::{
-    HttpConnector as SmithyHttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn,
-};
-use aws_smithy_runtime_api::client::interceptors::Intercept;
-use aws_smithy_runtime_api::client::interceptors::context::BeforeTransmitInterceptorContextMut;
-use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
-use aws_smithy_runtime_api::client::result::ConnectorError;
-use aws_smithy_runtime_api::client::runtime_components::RuntimeComponents;
-use aws_smithy_types::body::SdkBody;
-use aws_smithy_types::config_bag::ConfigBag;
-use aws_smithy_types::timeout::TimeoutConfig;
-use http::Uri;
-use hyper_util::client::legacy::Client as HyperClient;
-use hyper_util::rt::{TokioExecutor, TokioTimer};
-use rustfs_config::{DEFAULT_TRUST_LEAF_CERT_AS_CA, ENV_TRUST_LEAF_CERT_AS_CA, RUSTFS_CA_CERT, RUSTFS_TLS_CERT};
-use rustfs_utils::egress::{OutboundUrlError, validate_outbound_url};
-use rustls_pki_types::pem::PemObject;
-use std::fmt;
-use std::path::Path;
-use std::sync::Arc;
-use std::time::{Duration, SystemTime};
-use tower::Service;
-use tracing::warn;
-use url::Url;
-
-const REDACTED_CREDENTIAL: &str = "<redacted>";
-pub(crate) const EXPIRED_REMOTE_TARGET_CREDENTIALS: &str = "remote target credentials have expired";
-
-/// Request addressing style for a remote S3-compatible endpoint.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum PathStyle {
-    /// Caller did not choose; the builder defaults to path-style because that
-    /// is what custom S3-compatible endpoints accept most reliably.
-    Auto,
-    /// `https://endpoint/bucket/key`.
-    Path,
-    /// `https://bucket.endpoint/key`.
-    VirtualHost,
-}
-
-impl PathStyle {
-    /// Resolves the style to the SDK `force_path_style` flag. `Auto` keeps
-    /// the historical replication default (path-style).
-    pub fn force_path_style(self) -> bool {
-        !matches!(self, PathStyle::VirtualHost)
-    }
-}
-
-/// Static or temporary credentials for a remote endpoint. `expiration` without
-/// a `session_token` is rejected at build time: only STS-style temporary
-/// credentials expire, so that combination is a corrupted configuration
-/// rather than a static key.
-#[derive(Clone)]
-pub struct RemoteCredentials {
-    pub access_key: String,
-    pub secret_key: String,
-    pub session_token: Option<String>,
-    pub expiration: Option<SystemTime>,
-    /// SDK credential `account_id`; replication targets pass their reset id.
-    pub account_id: String,
-}
-
-impl fmt::Debug for RemoteCredentials {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("RemoteCredentials")
-            .field("access_key", &self.access_key)
-            .field("secret_key", &REDACTED_CREDENTIAL)
-            .field("session_token", &self.session_token.as_ref().map(|_| REDACTED_CREDENTIAL))
-            .field("expiration", &self.expiration)
-            .field("account_id", &self.account_id)
-            .finish()
-    }
-}
-
-/// Neutral description of a remote S3 endpoint from which an
-/// `aws_sdk_s3::Client` is built.
-#[derive(Clone, Debug)]
-pub struct RemoteS3EndpointSpec {
-    /// `host[:port]` without a scheme; `secure` selects `https` or `http`.
-    pub endpoint: String,
-    pub secure: bool,
-    pub region: String,
-    pub path_style: PathStyle,
-    pub credentials: Option<RemoteCredentials>,
-    /// Accept any server certificate. Takes priority over `ca_cert_pem`.
-    pub skip_tls_verify: bool,
-    /// Extra PEM bundle trusted alongside the platform roots and the
-    /// `RUSTFS_TLS_PATH` bundle. `None` and whitespace-only mean "not set".
-    pub ca_cert_pem: Option<String>,
-    pub connect_timeout: Option<Duration>,
-    pub read_timeout: Option<Duration>,
-    /// Appended to the SDK `User-Agent` (space separated) so the remote side
-    /// can identify the caller; empty means no suffix.
-    pub user_agent_suffix: &'static str,
-}
-
-impl RemoteS3EndpointSpec {
-    /// Full endpoint URL (`scheme://host[:port]`) as handed to the SDK.
-    pub fn endpoint_url(&self) -> String {
-        if self.secure {
-            format!("https://{}", self.endpoint)
-        } else {
-            format!("http://{}", self.endpoint)
-        }
-    }
-
-    fn custom_ca_pem(&self) -> Option<&str> {
-        self.ca_cert_pem.as_deref().filter(|pem| !pem.trim().is_empty())
-    }
-}
-
-#[derive(Debug, thiserror::Error)]
-pub enum RemoteS3ClientError {
-    #[error("remote endpoint requires credentials")]
-    MissingCredentials,
-    #[error("{0}")]
-    Credentials(&'static str),
-    #[error("invalid target endpoint: {0}")]
-    InvalidEndpoint(String),
-    #[error("target endpoint is not allowed: {0}")]
-    EndpointNotAllowed(#[source] OutboundUrlError),
-    #[error("invalid target CA PEM: {0}")]
-    InvalidCaPem(String),
-}
-
-#[derive(Clone)]
-pub(crate) struct RemoteTargetCredentialsProvider {
-    pub(crate) credentials: SdkCredentials,
-}
-
-impl RemoteTargetCredentialsProvider {
-    pub(crate) fn resolve_at(&self, now: SystemTime) -> aws_credential_types::provider::Result {
-        if self.credentials.expiry().is_some_and(|expiration| expiration <= now) {
-            return Err(CredentialsError::provider_error(std::io::Error::other(EXPIRED_REMOTE_TARGET_CREDENTIALS)));
-        }
-        Ok(self.credentials.clone())
-    }
-}
-
-impl fmt::Debug for RemoteTargetCredentialsProvider {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("RemoteTargetCredentialsProvider")
-            .field("temporary", &self.credentials.session_token().is_some())
-            .field("expiration", &self.credentials.expiry())
-            .finish()
-    }
-}
-
-impl ProvideCredentials for RemoteTargetCredentialsProvider {
-    fn provide_credentials<'a>(&'a self) -> future::ProvideCredentials<'a>
-    where
-        Self: 'a,
-    {
-        future::ProvideCredentials::ready(self.resolve_at(SystemTime::now()))
-    }
-
-    fn fallback_on_interrupt(&self) -> Option<SdkCredentials> {
-        self.resolve_at(SystemTime::now()).ok()
-    }
-}
-
-pub(crate) fn remote_sdk_credentials(credentials: &RemoteCredentials, now: SystemTime) -> Result<SdkCredentials, &'static str> {
-    if credentials.expiration.is_some() && credentials.session_token.is_none() {
-        return Err("remote target credential expiration requires a session token");
-    }
-    if credentials.expiration.is_some_and(|expiration| expiration <= now) {
-        return Err(EXPIRED_REMOTE_TARGET_CREDENTIALS);
-    }
-
-    let mut builder = SdkCredentials::builder()
-        .access_key_id(credentials.access_key.clone())
-        .secret_access_key(credentials.secret_key.clone())
-        .account_id(credentials.account_id.clone())
-        .provider_name("bucket_target_sys");
-    if let Some(session_token) = &credentials.session_token {
-        builder = builder.session_token(session_token.clone());
-    }
-    if let Some(expiration) = credentials.expiration {
-        builder = builder.expiry(expiration);
-    }
-    Ok(builder.build())
-}
-
-/// Appends a caller-identifying token to the SDK `User-Agent`. Runs after
-/// signing: SigV4 excludes `user-agent` from the canonical request, so the
-/// signature stays valid.
-#[derive(Debug)]
-struct UserAgentSuffixInterceptor {
-    suffix: &'static str,
-}
-
-impl Intercept for UserAgentSuffixInterceptor {
-    fn name(&self) -> &'static str {
-        "RustfsUserAgentSuffix"
-    }
-
-    fn modify_before_transmit(
-        &self,
-        context: &mut BeforeTransmitInterceptorContextMut<'_>,
-        _runtime_components: &RuntimeComponents,
-        _cfg: &mut ConfigBag,
-    ) -> Result<(), BoxError> {
-        let headers = context.request_mut().headers_mut();
-        let user_agent = match headers.get(http::header::USER_AGENT.as_str()) {
-            Some(existing) => format!("{existing} {}", self.suffix),
-            None => self.suffix.to_string(),
-        };
-        headers.try_insert(http::header::USER_AGENT.as_str(), user_agent)?;
-        Ok(())
-    }
-}
-
-/// Builds the SDK config for `spec` without finalizing it, so callers can add
-/// interceptors or (in tests) swap the HTTP client before `build()`.
-pub(crate) async fn build_remote_s3_config(
-    spec: &RemoteS3EndpointSpec,
-) -> Result<aws_sdk_s3::config::Builder, RemoteS3ClientError> {
-    let Some(credentials) = &spec.credentials else {
-        return Err(RemoteS3ClientError::MissingCredentials);
-    };
-    let creds = remote_sdk_credentials(credentials, SystemTime::now()).map_err(RemoteS3ClientError::Credentials)?;
-
-    let endpoint = spec.endpoint_url();
-    let parsed_endpoint = Url::parse(&endpoint).map_err(|err| RemoteS3ClientError::InvalidEndpoint(err.to_string()))?;
-    validate_remote_endpoint(&parsed_endpoint).map_err(RemoteS3ClientError::EndpointNotAllowed)?;
-
-    let mut config_builder = S3Config::builder()
-        .endpoint_url(endpoint)
-        .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds }))
-        .region(SdkRegion::new(spec.region.clone()))
-        .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
-        .request_checksum_calculation(replication_request_checksum_calculation());
-
-    if spec.path_style.force_path_style() {
-        config_builder = config_builder.force_path_style(true);
-    }
-
-    if let Some(http_client) = build_aws_s3_http_client_for_spec(spec).await? {
-        config_builder = config_builder.http_client(http_client);
-    }
-
-    if spec.connect_timeout.is_some() || spec.read_timeout.is_some() {
-        let mut timeouts = TimeoutConfig::builder();
-        if let Some(connect_timeout) = spec.connect_timeout {
-            timeouts = timeouts.connect_timeout(connect_timeout);
-        }
-        if let Some(read_timeout) = spec.read_timeout {
-            timeouts = timeouts.read_timeout(read_timeout);
-        }
-        config_builder = config_builder.timeout_config(timeouts.build());
-    }
-
-    if !spec.user_agent_suffix.is_empty() {
-        config_builder = config_builder.interceptor(UserAgentSuffixInterceptor {
-            suffix: spec.user_agent_suffix,
-        });
-    }
-
-    Ok(config_builder)
-}
-
-/// Builds an `aws_sdk_s3::Client` for `spec`, applying the outbound endpoint
-/// gate, credential validation and the TLS transport selection.
-pub async fn build_remote_s3_client(spec: &RemoteS3EndpointSpec) -> Result<S3Client, RemoteS3ClientError> {
-    Ok(S3Client::from_conf(build_remote_s3_config(spec).await?.build()))
-}
-
-#[derive(Debug)]
-struct AcceptAnyServerCertVerifier;
-
-impl rustls::client::danger::ServerCertVerifier for AcceptAnyServerCertVerifier {
-    fn verify_server_cert(
-        &self,
-        _end_entity: &rustls_pki_types::CertificateDer<'_>,
-        _intermediates: &[rustls_pki_types::CertificateDer<'_>],
-        _server_name: &rustls_pki_types::ServerName<'_>,
-        _ocsp_response: &[u8],
-        _now: rustls_pki_types::UnixTime,
-    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
-        Ok(rustls::client::danger::ServerCertVerified::assertion())
-    }
-
-    fn verify_tls12_signature(
-        &self,
-        _message: &[u8],
-        _cert: &rustls_pki_types::CertificateDer<'_>,
-        _dss: &rustls::DigitallySignedStruct,
-    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
-        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
-    }
-
-    fn verify_tls13_signature(
-        &self,
-        _message: &[u8],
-        _cert: &rustls_pki_types::CertificateDer<'_>,
-        _dss: &rustls::DigitallySignedStruct,
-    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
-        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
-    }
-
-    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
-        rustls::crypto::aws_lc_rs::default_provider()
-            .signature_verification_algorithms
-            .supported_schemes()
-    }
-}
-
-#[derive(Clone)]
-struct TargetHyperHttpConnector<C> {
-    client: HyperClient<C, SdkBody>,
-}
-
-impl<C> fmt::Debug for TargetHyperHttpConnector<C> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("TargetHyperHttpConnector")
-            .field("client", &"** hyper client **")
-            .finish()
-    }
-}
-
-impl<C> SmithyHttpConnector for TargetHyperHttpConnector<C>
-where
-    C: Clone + Send + Sync + 'static,
-    C: Service<Uri>,
-    C::Response:
-        hyper::rt::Read + hyper::rt::Write + hyper_util::client::legacy::connect::Connection + Send + Sync + Unpin + 'static,
-    C::Future: Unpin + Send + 'static,
-    C::Error: Into<BoxError>,
-{
-    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
-        let request = match request.try_into_http1x() {
-            Ok(request) => request,
-            Err(err) => return HttpConnectorFuture::ready(Err(ConnectorError::user(err.into()))),
-        };
-
-        let mut client = self.client.clone();
-        let fut = client.call(request);
-        HttpConnectorFuture::new(async move {
-            let response = fut
-                .await
-                .map_err(|err| ConnectorError::io(err.into()))?
-                .map(SdkBody::from_body_1_x);
-            HttpResponse::try_from(response).map_err(|err| ConnectorError::other(err.into(), None))
-        })
-    }
-}
-
-pub(crate) fn ensure_rustls_crypto_provider() {
-    if rustls::crypto::CryptoProvider::get_default().is_none() {
-        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-    }
-}
-
-/// Env opt-in that re-enables loopback replication targets. Loopback (`127.0.0.1`,
-/// `::1`, `localhost`) is a classic SSRF vector and stays rejected by default, but
-/// single-host multi-instance dev setups and the e2e harness legitimately replicate
-/// over loopback. Never set this in production.
-const ALLOW_LOOPBACK_REPLICATION_TARGET_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET";
-
-fn loopback_replication_targets_allowed() -> bool {
-    std::env::var(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV)
-        .map(|v| v.eq_ignore_ascii_case("true") || v == "1")
-        .unwrap_or(false)
-}
-
-const REPLICATION_STREAMING_CHECKSUMS_ENV: &str = "RUSTFS_REPLICATION_STREAMING_CHECKSUMS";
-
-/// Streaming trailer checksums make the SDK frame request bodies as
-/// `aws-chunked`; a target that does not decode that framing stores the frames
-/// verbatim, silently corrupting every replica while the transfer itself
-/// succeeds (#6853). Plain signed payloads are the compatible default; the env
-/// knob restores trailer checksums for fleets whose targets are all known to
-/// decode them.
-pub(crate) fn replication_request_checksum_calculation() -> RequestChecksumCalculation {
-    if std::env::var(REPLICATION_STREAMING_CHECKSUMS_ENV)
-        .map(|v| v.eq_ignore_ascii_case("true") || v == "1")
-        .unwrap_or(false)
-    {
-        RequestChecksumCalculation::WhenSupported
-    } else {
-        RequestChecksumCalculation::WhenRequired
-    }
-}
-
-/// Outbound gate for operator-configured remote endpoints (replication
-/// targets, on-demand migration sources). See
-/// `docs/operations/outbound-connection-policy.md`.
-pub fn validate_remote_endpoint(url: &Url) -> Result<(), OutboundUrlError> {
-    validate_remote_endpoint_inner(url, loopback_replication_targets_allowed())
-}
-
-pub(crate) fn validate_remote_endpoint_inner(url: &Url, allow_loopback: bool) -> Result<(), OutboundUrlError> {
-    match validate_outbound_url(url) {
-        Ok(()) => Ok(()),
-        // Replication targets are trusted infrastructure the operator configures, and
-        // legitimately live on private networks, so private addresses are always allowed.
-        Err(OutboundUrlError::ForbiddenHost {
-            reason: "private address",
-            ..
-        }) => Ok(()),
-        // Loopback is far higher SSRF risk, so it is allowed only under the explicit,
-        // off-by-default opt-in above (single-host multi-instance / the e2e harness).
-        Err(OutboundUrlError::ForbiddenHost {
-            reason: "loopback address" | "loopback host",
-            ..
-        }) if allow_loopback => Ok(()),
-        Err(err) => Err(err),
-    }
-}
-
-pub(crate) fn build_insecure_aws_s3_http_client() -> SharedHttpClient {
-    ensure_rustls_crypto_provider();
-
-    let tls_config = rustls::ClientConfig::builder()
-        .dangerous()
-        .with_custom_certificate_verifier(Arc::new(AcceptAnyServerCertVerifier))
-        .with_no_client_auth();
-
-    let https = hyper_rustls::HttpsConnectorBuilder::new()
-        .with_tls_config(tls_config)
-        .https_or_http()
-        .enable_http1()
-        .enable_http2()
-        .build();
-    let mut client_builder = HyperClient::builder(TokioExecutor::new());
-    client_builder.pool_timer(TokioTimer::new());
-    let client = client_builder.build(https);
-    let connector = SharedHttpConnector::new(TargetHyperHttpConnector { client });
-
-    http_client_fn(move |_settings, _components| connector.clone())
-}
-
-fn validate_ca_pem_bundle(ca_cert_pem: &[u8]) -> Result<(), String> {
-    let certs = rustls_pki_types::CertificateDer::pem_slice_iter(ca_cert_pem)
-        .collect::<Result<Vec<_>, _>>()
-        .map_err(|err| format!("invalid PEM encoding: {err}"))?;
-
-    if certs.is_empty() {
-        return Err("no certificates found".to_string());
-    }
-
-    // Smithy's rustls adapter defers parsing custom certificates and assumes
-    // they are valid when the HTTPS connector is built. Validate every DER
-    // certificate first so malformed configuration is reported rather than
-    // reaching an `expect` in the dependency.
-    let mut validation_store = rustls::RootCertStore::empty();
-    for cert in certs {
-        validation_store
-            .add(cert)
-            .map_err(|err| format!("invalid X.509 certificate: {err}"))?;
-    }
-
-    Ok(())
-}
-
-pub(crate) fn validate_target_ca_pem(ca_cert_pem: &str) -> Result<(), RemoteS3ClientError> {
-    validate_ca_pem_bundle(ca_cert_pem.as_bytes()).map_err(RemoteS3ClientError::InvalidCaPem)
-}
-
-pub(crate) fn compose_replication_trust_store(
-    certificate_bundles: impl IntoIterator<Item = Vec<u8>>,
-) -> (smithy_tls::TrustStore, usize) {
-    // `TrustStore::default()` keeps the platform-native roots enabled. Target
-    // and RUSTFS_TLS_PATH certificates extend that baseline instead of
-    // replacing it with a target-specific trust island.
-    let mut trust_store = smithy_tls::TrustStore::default();
-    let mut custom_bundle_count = 0;
-    for pem in certificate_bundles {
-        trust_store.add_pem_certificate(pem);
-        custom_bundle_count += 1;
-    }
-
-    (trust_store, custom_bundle_count)
-}
-
-pub(crate) fn build_aws_s3_http_client_with_trust_store(
-    trust_store: smithy_tls::TrustStore,
-) -> Result<SharedHttpClient, RemoteS3ClientError> {
-    let tls_context = smithy_tls::TlsContext::builder()
-        .with_trust_store(trust_store)
-        .build()
-        .map_err(|err| RemoteS3ClientError::InvalidCaPem(err.to_string()))?;
-
-    Ok(SmithyHttpClientBuilder::new()
-        .tls_provider(smithy_tls::Provider::rustls(smithy_tls::rustls_provider::CryptoMode::AwsLc))
-        .tls_context(tls_context)
-        .build_https())
-}
-
-pub(crate) async fn load_tls_path_ca_bundles(tls_dir: &Path, trust_leaf_cert_as_ca: bool) -> Vec<Vec<u8>> {
-    let mut certificate_bundles = Vec::new();
-
-    let ca_path = tls_dir.join(RUSTFS_CA_CERT);
-    match tokio::fs::read(&ca_path).await {
-        Ok(pem) => match validate_ca_pem_bundle(&pem) {
-            Ok(()) => certificate_bundles.push(pem),
-            Err(err) => warn!("ignoring invalid custom CA bundle {:?} for replication client: {}", ca_path, err),
-        },
-        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
-        Err(e) => warn!("failed to read custom CA bundle {:?} for replication client: {}", ca_path, e),
-    }
-
-    if trust_leaf_cert_as_ca {
-        let leaf_cert_path = tls_dir.join(RUSTFS_TLS_CERT);
-        match tokio::fs::read(&leaf_cert_path).await {
-            Ok(pem) => match validate_ca_pem_bundle(&pem) {
-                Ok(()) => certificate_bundles.push(pem),
-                Err(err) => warn!(
-                    "ignoring invalid leaf certificate {:?} for replication client trust store: {}",
-                    leaf_cert_path, err
-                ),
-            },
-            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
-            Err(e) => warn!("failed to read leaf cert {:?} for replication client trust store: {}", leaf_cert_path, e),
-        }
-    }
-
-    certificate_bundles
-}
-
-async fn load_configured_tls_ca_bundles() -> Vec<Vec<u8>> {
-    let tls_path = rustfs_utils::get_env_str(rustfs_config::ENV_RUSTFS_TLS_PATH, rustfs_config::DEFAULT_RUSTFS_TLS_PATH);
-    if tls_path.is_empty() {
-        return Vec::new();
-    }
-
-    load_tls_path_ca_bundles(
-        Path::new(&tls_path),
-        rustfs_utils::get_env_bool(ENV_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_LEAF_CERT_AS_CA),
-    )
-    .await
-}
-
-pub(crate) async fn build_aws_s3_http_client_from_target_ca_pem(
-    ca_cert_pem: &str,
-) -> Result<SharedHttpClient, RemoteS3ClientError> {
-    validate_target_ca_pem(ca_cert_pem)?;
-
-    let mut certificate_bundles = load_configured_tls_ca_bundles().await;
-    certificate_bundles.push(ca_cert_pem.as_bytes().to_vec());
-    let (trust_store, _) = compose_replication_trust_store(certificate_bundles);
-
-    build_aws_s3_http_client_with_trust_store(trust_store)
-}
-
-/// Selects the HTTP client for `spec`: `None` keeps the SDK default (plain
-/// HTTP, or HTTPS with platform roots when no custom trust is configured).
-pub(crate) async fn build_aws_s3_http_client_for_spec(
-    spec: &RemoteS3EndpointSpec,
-) -> Result<Option<SharedHttpClient>, RemoteS3ClientError> {
-    if !spec.secure {
-        return Ok(None);
-    }
-
-    if spec.skip_tls_verify {
-        return Ok(Some(build_insecure_aws_s3_http_client()));
-    }
-
-    if let Some(ca_cert_pem) = spec.custom_ca_pem() {
-        return build_aws_s3_http_client_from_target_ca_pem(ca_cert_pem).await.map(Some);
-    }
-
-    Ok(build_aws_s3_http_client_from_tls_path().await)
-}
-
-async fn build_aws_s3_http_client_from_tls_path() -> Option<SharedHttpClient> {
-    let certificate_bundles = load_configured_tls_ca_bundles().await;
-    if certificate_bundles.is_empty() {
-        return None;
-    }
-
-    let (trust_store, _) = compose_replication_trust_store(certificate_bundles);
-    match build_aws_s3_http_client_with_trust_store(trust_store) {
-        Ok(client) => Some(client),
-        Err(e) => {
-            warn!("failed to build AWS SDK TLS context for replication client: {}", e);
-            None
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use aws_smithy_runtime_api::http::StatusCode as SmithyStatusCode;
-    use std::sync::Mutex;
-
-    fn spec(endpoint: &str, secure: bool) -> RemoteS3EndpointSpec {
-        RemoteS3EndpointSpec {
-            endpoint: endpoint.to_string(),
-            secure,
-            region: "us-east-1".to_string(),
-            path_style: PathStyle::Auto,
-            credentials: Some(RemoteCredentials {
-                access_key: "access".to_string(),
-                secret_key: "secret".to_string(),
-                session_token: None,
-                expiration: None,
-                account_id: String::new(),
-            }),
-            skip_tls_verify: false,
-            ca_cert_pem: None,
-            connect_timeout: None,
-            read_timeout: None,
-            user_agent_suffix: "",
-        }
-    }
-
-    type RecordedHeaders = Arc<Mutex<Vec<Vec<(String, String)>>>>;
-
-    #[derive(Clone, Debug)]
-    struct RecordingHeaderConnector {
-        request_headers: RecordedHeaders,
-    }
-
-    impl SmithyHttpConnector for RecordingHeaderConnector {
-        fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
-            self.request_headers
-                .lock()
-                .expect("recorded header lock should not be poisoned")
-                .push(
-                    request
-                        .headers()
-                        .iter()
-                        .map(|(k, v)| (k.to_string(), v.to_string()))
-                        .collect(),
-                );
-            HttpConnectorFuture::ready(Ok(HttpResponse::new(
-                SmithyStatusCode::try_from(200_u16).expect("200 should be a valid response status"),
-                SdkBody::empty(),
-            )))
-        }
-    }
-
-    #[tokio::test]
-    async fn build_remote_s3_client_rejects_loopback_and_metadata_endpoints() {
-        // Default (no loopback opt-in): loopback in IPv4, IPv6 and hostname
-        // forms plus the metadata endpoint all return the typed gate error.
-        for endpoint in ["127.0.0.1:9000", "[::1]:9000", "localhost:9000", "169.254.169.254"] {
-            let err = build_remote_s3_client(&spec(endpoint, false))
-                .await
-                .err()
-                .unwrap_or_else(|| panic!("{endpoint} must be rejected by the outbound gate"));
-            assert!(
-                matches!(err, RemoteS3ClientError::EndpointNotAllowed(OutboundUrlError::ForbiddenHost { .. })),
-                "{endpoint}: unexpected error {err:?}"
-            );
-            assert!(err.to_string().contains("not allowed"), "{endpoint}: {err}");
-        }
-    }
-
-    #[tokio::test]
-    async fn build_remote_s3_client_allows_private_and_public_endpoints() {
-        for endpoint in ["10.0.0.1:9000", "192.168.1.20", "s3.example.com"] {
-            build_remote_s3_client(&spec(endpoint, false))
-                .await
-                .unwrap_or_else(|err| panic!("{endpoint} should be allowed: {err}"));
-        }
-    }
-
-    #[tokio::test]
-    async fn build_remote_s3_client_requires_credentials() {
-        let mut spec = spec("s3.example.com", true);
-        spec.credentials = None;
-        let err = build_remote_s3_client(&spec)
-            .await
-            .expect_err("missing credentials must be a typed error");
-        assert!(matches!(err, RemoteS3ClientError::MissingCredentials));
-    }
-
-    #[tokio::test]
-    async fn build_remote_s3_client_rejects_expiration_without_session_token() {
-        let mut spec = spec("s3.example.com", true);
-        spec.credentials
-            .as_mut()
-            .expect("spec fixture carries credentials")
-            .expiration = Some(SystemTime::now() + Duration::from_secs(3_600));
-        let err = build_remote_s3_client(&spec)
-            .await
-            .expect_err("expiration without session token must be rejected");
-        assert_eq!(err.to_string(), "remote target credential expiration requires a session token");
-    }
-
-    #[tokio::test]
-    async fn build_remote_s3_client_rejects_invalid_custom_ca_pem() {
-        let mut spec = spec("192.168.1.10:9000", true);
-        spec.ca_cert_pem = Some("not a pem".to_string());
-        let err = build_remote_s3_client(&spec)
-            .await
-            .expect_err("invalid custom CA PEM must be rejected");
-        assert!(matches!(err, RemoteS3ClientError::InvalidCaPem(_)));
-        assert!(err.to_string().contains("invalid target CA PEM"));
-    }
-
-    #[test]
-    fn path_style_auto_and_path_force_path_style() {
-        assert!(PathStyle::Auto.force_path_style());
-        assert!(PathStyle::Path.force_path_style());
-        assert!(!PathStyle::VirtualHost.force_path_style());
-    }
-
-    #[test]
-    fn remote_credentials_debug_redacts_secrets() {
-        let credentials = RemoteCredentials {
-            access_key: "access".to_string(),
-            secret_key: "very-secret".to_string(),
-            session_token: Some("session-token".to_string()),
-            expiration: None,
-            account_id: String::new(),
-        };
-        let rendered = format!("{credentials:?}");
-        assert!(rendered.contains("access"));
-        assert!(!rendered.contains("very-secret"));
-        assert!(!rendered.contains("session-token"));
-    }
-
-    #[tokio::test]
-    async fn user_agent_suffix_is_appended_after_signing() {
-        let request_headers: RecordedHeaders = Arc::new(Mutex::new(Vec::new()));
-        let connector = SharedHttpConnector::new(RecordingHeaderConnector {
-            request_headers: Arc::clone(&request_headers),
-        });
-        let http_client = http_client_fn(move |_settings, _components| connector.clone());
-
-        let mut spec = spec("s3.example.com", true);
-        spec.user_agent_suffix = "RustFS-Test/0.0";
-        spec.connect_timeout = Some(Duration::from_secs(5));
-        spec.read_timeout = Some(Duration::from_secs(5));
-        let config = build_remote_s3_config(&spec)
-            .await
-            .expect("spec should build")
-            .http_client(http_client)
-            .build();
-        S3Client::from_conf(config)
-            .head_bucket()
-            .bucket("bucket")
-            .send()
-            .await
-            .expect("recording connector should accept the request");
-
-        let recorded = request_headers.lock().expect("recorded header lock should not be poisoned");
-        let headers = &recorded[0];
-        let user_agent = headers
-            .iter()
-            .find(|(k, _)| k.eq_ignore_ascii_case("user-agent"))
-            .map(|(_, v)| v.as_str())
-            .expect("SDK request must carry a user-agent");
-        assert!(user_agent.ends_with(" RustFS-Test/0.0"), "user-agent was {user_agent}");
-        assert!(user_agent.starts_with("aws-sdk-rust/"), "SDK identity must be preserved: {user_agent}");
-        assert!(
-            headers.iter().any(|(k, _)| k.eq_ignore_ascii_case("authorization")),
-            "request must still be signed"
-        );
-    }
-}

crates/rio/Cargo.toml

@@ -90,15 +90,8 @@ s3s = { workspace = true, features = ["minio"] }
 hex-simd.workspace = true
 
 [dev-dependencies]
-tokio = { workspace = true, features = ["test-util"] }
 tokio-test = { workspace = true }
-criterion = { workspace = true, features = ["html_reports"] }
-proptest = { workspace = true }
 axum = { workspace = true }
 hyper = { workspace = true, features = ["http2", "server"] }
 hyper-util = { workspace = true, features = ["tokio"] }
 http-body-util = { workspace = true }
-
-[[bench]]
-name = "tee_reader"
-harness = false

crates/rio/benches/tee_reader.rs

@@ -1,103 +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.
-
-//! Throughput of `tee_reader` versus reading the same source directly:
-//! 64 MiB of data served in 1 MiB chunks, consumed with 1 MiB reads.
-
-use bytes::Bytes;
-use criterion::{Criterion, Throughput, criterion_group, criterion_main};
-use rustfs_rio::tee_reader;
-use std::hint::black_box;
-use std::io;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
-
-const CHUNK_BYTES: usize = 1024 * 1024;
-const TOTAL_BYTES: usize = 64 * 1024 * 1024;
-const TEE_BUFFER_BYTES: usize = 4 * CHUNK_BYTES;
-
-/// In-memory source that serves at most `CHUNK_BYTES` per poll.
-struct ChunkedSource {
-    data: Bytes,
-    pos: usize,
-}
-
-impl AsyncRead for ChunkedSource {
-    fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
-        let remaining = self.data.len() - self.pos;
-        let n = CHUNK_BYTES.min(remaining).min(buf.remaining());
-        buf.put_slice(&self.data[self.pos..self.pos + n]);
-        self.pos += n;
-        Poll::Ready(Ok(()))
-    }
-}
-
-async fn consume<R: AsyncRead + Unpin>(mut reader: R) -> usize {
-    let mut buf = vec![0u8; CHUNK_BYTES];
-    let mut total = 0;
-    loop {
-        let n = reader.read(&mut buf).await.expect("read");
-        if n == 0 {
-            return total;
-        }
-        total += n;
-    }
-}
-
-fn bench_tee_reader(c: &mut Criterion) {
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(2)
-        .enable_all()
-        .build()
-        .expect("build tokio runtime for tee_reader benchmark");
-    let data = Bytes::from(vec![0xA5u8; TOTAL_BYTES]);
-
-    let mut group = c.benchmark_group("tee_reader_64mib_1mib_chunks");
-    group.throughput(Throughput::Bytes(TOTAL_BYTES as u64));
-    group.sample_size(10);
-
-    group.bench_function("direct_read", |b| {
-        b.iter(|| {
-            let source = ChunkedSource {
-                data: data.clone(),
-                pos: 0,
-            };
-            let total = runtime.block_on(consume(source));
-            black_box(total)
-        })
-    });
-
-    group.bench_function("tee_primary_plus_secondary", |b| {
-        b.iter(|| {
-            let source = ChunkedSource {
-                data: data.clone(),
-                pos: 0,
-            };
-            let (primary, secondary) = tee_reader(source, TEE_BUFFER_BYTES);
-            let totals = runtime.block_on(async {
-                let secondary_task = tokio::spawn(consume(secondary));
-                let primary_total = consume(primary).await;
-                let secondary_total = secondary_task.await.expect("secondary task");
-                (primary_total, secondary_total)
-            });
-            black_box(totals)
-        })
-    });
-
-    group.finish();
-}
-
-criterion_group!(benches, bench_tee_reader);
-criterion_main!(benches);

crates/rio/src/lib.rs

@@ -118,12 +118,6 @@ pub use hardlimit_reader::HardLimitReader;
 
 mod hash_reader;
 pub use hash_reader::*;
-
-mod tee_reader;
-pub use tee_reader::{
-    DEFAULT_TEE_MAX_DRAIN_BYTES, TeeDrainLimitExceeded, TeeOptions, TeePrimary, TeeSecondary, TeeStream, tee_reader,
-    tee_reader_with_options,
-};
 mod checksum;
 pub use checksum::*;
 

crates/scanner/src/scanner.rs

@@ -552,21 +552,18 @@ pub(super) fn data_usage_info_has_persisted_baseline_identity(info: &DataUsageIn
 }
 
 pub(super) fn data_usage_info_is_bootstrap_pending(info: &DataUsageInfo) -> bool {
-    let Some(last_update) = info.last_update else {
+    if info.last_update.is_none() || info.scanner_cycle.is_some() {
         return false;
-    };
+    }
 
-    info == &scanner_usage_bootstrap_marker(last_update, info.scanner_epoch)
-}
-
-pub(super) fn scanner_usage_bootstrap_marker(last_update: std::time::SystemTime, scanner_epoch: Option<u64>) -> DataUsageInfo {
-    DataUsageInfo {
-        last_update: Some(last_update),
-        scanner_epoch,
+    let expected = DataUsageInfo {
+        last_update: info.last_update,
+        scanner_epoch: info.scanner_epoch,
         usage_snapshot_converged: Some(false),
         usage_snapshot_bootstrap_pending: true,
         ..Default::default()
-    }
+    };
+    info == &expected
 }
 
 fn usage_cache_needs_prompt_scan(authoritative: &DataUsageInfo, observed: Option<&DataUsageInfo>) -> bool {
@@ -918,8 +915,8 @@ fn prepare_cycle_for_usage_floor_bootstrap(
                 },
             )
         }
-        PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence => {
-            // The legacy incomplete fence proves only its leader epoch, not
+        PersistedUsageFloorStartup::RecoveredLegacyEmptyFence => {
+            // The legacy empty fence proves only its leader epoch, not
             // namespace coverage. Clear coverage while retaining the durable
             // cycle number so surviving caches cannot force a regression.
             let next = cycle_info.next;
@@ -1441,7 +1438,6 @@ async fn fence_scanner_epoch_after_cycle_timeout<Store, LockLost>(
     cycle_info: &mut CurrentCycle,
     cycle_revision: &mut DataUsageCacheRevision,
     leader_epoch: &mut u64,
-    allow_bootstrap_pending: bool,
     lock_lost: LockLost,
 ) -> bool
 where
@@ -1455,7 +1451,7 @@ where
         cycle_info,
         cycle_revision,
         leader_epoch,
-        allow_bootstrap_pending,
+        false,
         ScannerCycleResetPolicy::None,
     );
     tokio::pin!(claim);
@@ -1477,7 +1473,6 @@ struct ScannerCycleDeadlineState<'a> {
     cycle_revision: &'a mut DataUsageCacheRevision,
     leader_epoch: &'a mut u64,
     cycle_budget: &'a ScannerCycleBudget,
-    allow_bootstrap_pending: bool,
 }
 
 fn cycle_timeout_requires_recovery(worker_stopped: bool, cycle_state_persisted: bool, generation_fenced: bool) -> bool {
@@ -1499,7 +1494,6 @@ async fn handle_scanner_cycle_deadline<Store>(
         state.cycle_info,
         state.cycle_revision,
         state.leader_epoch,
-        state.allow_bootstrap_pending,
         guard.lock_lost_notified(),
     )
     .await;
@@ -2581,7 +2575,7 @@ async fn run_data_scanner_with_maintenance_state(
     match usage_floor_startup {
         PersistedUsageFloorStartup::Authoritative
         | PersistedUsageFloorStartup::BootstrapPending
-        | PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence => {}
+        | PersistedUsageFloorStartup::RecoveredLegacyEmptyFence => {}
         PersistedUsageFloorStartup::Missing => {
             if ctx.is_cancelled() || guard.is_lock_lost() {
                 global_metrics().set_cycle(None).await;
@@ -2676,8 +2670,8 @@ async fn run_data_scanner_with_maintenance_state(
         finish_scanner_leader_iteration(false, "epoch_claim_failed", "leadership epoch claim failed".to_string()).await;
         return Ok(());
     }
-    if usage_floor_startup == PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence
-        && let Err(err) = complete_legacy_incomplete_usage_floor_recovery(storeapi.clone(), leader_epoch).await
+    if usage_floor_startup == PersistedUsageFloorStartup::RecoveredLegacyEmptyFence
+        && let Err(err) = complete_legacy_empty_usage_floor_recovery(storeapi.clone(), leader_epoch).await
     {
         let error = err.to_string();
         warn!(
@@ -2750,7 +2744,6 @@ async fn run_data_scanner_with_maintenance_state(
                         cycle_revision: &mut cycle_revision,
                         leader_epoch: &mut leader_epoch,
                         cycle_budget: &cycle_budget,
-                        allow_bootstrap_pending: allow_usage_floor_bootstrap_pending,
                     },
                     worker_stopped,
                     &mut guard,
@@ -3040,7 +3033,6 @@ async fn run_data_scanner_with_maintenance_state(
                         cycle_revision: &mut cycle_revision,
                         leader_epoch: &mut leader_epoch,
                         cycle_budget: &cycle_budget,
-                        allow_bootstrap_pending: allow_usage_floor_bootstrap_pending,
                     },
                     worker_stopped,
                     &mut guard,

crates/scanner/src/scanner/cycle_state.rs

@@ -26,10 +26,7 @@ pub(super) const MAX_SCANNER_CYCLE_RECOVERY_RETRIES: u32 = 5;
 const METRIC_SCANNER_CYCLE_RECOVERY_REQUIRED: &str = "rustfs_scanner_cycle_recovery_required";
 const METRIC_SCANNER_CYCLE_RECOVERY_RETRY_COUNT: &str = "rustfs_scanner_cycle_recovery_retry_count";
 const USAGE_FLOOR_LOAD_FAILED: &str = "usage_floor_load_failed";
-// Keep the published status value stable for operators that already alert on
-// the empty-fence recovery introduced by backlog-2102. The same durable marker
-// now also covers strictly validated data-bearing legacy fences.
-const LEGACY_INCOMPLETE_USAGE_FLOOR_RECOVERY: &str = "legacy_empty_usage_floor";
+const LEGACY_EMPTY_USAGE_FLOOR_RECOVERY: &str = "legacy_empty_usage_floor";
 const CACHE_CYCLE_AHEAD: &str = "cache_cycle_ahead";
 
 const SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD: &str = "full-rebuild";
@@ -185,9 +182,9 @@ pub(super) fn clear_scanner_cache_cycle_ahead() {
     }
 }
 
-pub(super) fn record_legacy_incomplete_usage_floor_recovery_pending(leader_epoch: u64) {
+pub(super) fn record_legacy_empty_usage_floor_recovery_pending(leader_epoch: u64) {
     let previous = scanner_cycle_recovery_status();
-    let same_recovery = previous.classification.as_deref() == Some(LEGACY_INCOMPLETE_USAGE_FLOOR_RECOVERY)
+    let same_recovery = previous.classification.as_deref() == Some(LEGACY_EMPTY_USAGE_FLOOR_RECOVERY)
         && previous.leader_epoch == Some(leader_epoch);
     let now = unix_now_secs();
     let (first_detected_at_unix_secs, retry_count) = if same_recovery {
@@ -199,20 +196,20 @@ pub(super) fn record_legacy_incomplete_usage_floor_recovery_pending(leader_epoch
         path: DATA_USAGE_OBJ_NAME_PATH.clone(),
         quarantine_path: Some(DATA_USAGE_RECOVERY_PATH.clone()),
         state: "usage_floor_recovery_pending".to_string(),
-        classification: Some(LEGACY_INCOMPLETE_USAGE_FLOOR_RECOVERY.to_string()),
+        classification: Some(LEGACY_EMPTY_USAGE_FLOOR_RECOVERY.to_string()),
         leader_epoch: Some(leader_epoch),
         first_detected_at_unix_secs,
         last_attempt_at_unix_secs: Some(now),
         retry_count,
         max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES,
         retryable: true,
-        reason: Some("legacy incomplete usage floor recovery is awaiting a fenced leadership claim".to_string()),
+        reason: Some("legacy empty usage floor recovery is awaiting a fenced leadership claim".to_string()),
         ..Default::default()
     });
 }
 
-pub(super) fn clear_legacy_incomplete_usage_floor_recovery_status() {
-    if scanner_cycle_recovery_status().classification.as_deref() == Some(LEGACY_INCOMPLETE_USAGE_FLOOR_RECOVERY) {
+pub(super) fn clear_legacy_empty_usage_floor_recovery_status() {
+    if scanner_cycle_recovery_status().classification.as_deref() == Some(LEGACY_EMPTY_USAGE_FLOOR_RECOVERY) {
         set_scanner_cycle_recovery_status(recovery_status("healthy", None, false));
     }
 }
@@ -1434,101 +1431,6 @@ async fn delete_usage_state_reset_slot(
     }
 }
 
-#[derive(Clone, Copy)]
-pub(super) enum ScannerUsageBootstrapPublishContext {
-    Initial,
-    Recovery,
-    Reset,
-}
-
-enum ScannerUsageBootstrapPublishError {
-    Encode(serde_json::Error),
-    Reconcile(EcstoreError),
-    MissingEtag,
-    Save(EcstoreError),
-}
-
-impl ScannerUsageBootstrapPublishError {
-    fn into_scanner_error(self, context: ScannerUsageBootstrapPublishContext) -> ScannerError {
-        let message = match context {
-            ScannerUsageBootstrapPublishContext::Initial => match self {
-                Self::Encode(err) => format!("failed to encode scanner usage baseline bootstrap: {err}"),
-                Self::Reconcile(err) => format!("failed to reconcile scanner usage bootstrap: {err}"),
-                Self::MissingEtag => "scanner usage bootstrap returned no ETag and could not be confirmed".to_string(),
-                Self::Save(err) => format!("failed to persist scanner usage bootstrap: {err}"),
-            },
-            ScannerUsageBootstrapPublishContext::Recovery => match self {
-                Self::Encode(err) => format!("failed to encode recovered scanner usage bootstrap: {err}"),
-                Self::Reconcile(err) => format!("failed to reconcile recovered scanner usage bootstrap: {err}"),
-                Self::MissingEtag => "recovered scanner usage bootstrap returned no ETag and could not be confirmed".to_string(),
-                Self::Save(err) => format!("failed to recover legacy incomplete scanner usage floor: {err}"),
-            },
-            ScannerUsageBootstrapPublishContext::Reset => match self {
-                Self::Encode(err) => format!("failed to encode scanner usage reset bootstrap marker: {err}"),
-                Self::Reconcile(err) => format!("failed to reconcile scanner usage reset bootstrap marker: {err}"),
-                Self::MissingEtag => "scanner usage reset bootstrap returned no ETag and could not be confirmed".to_string(),
-                Self::Save(err) if scanner_publication_epoch_changed(&err) => {
-                    "scanner usage reset deferred by a movement epoch change".to_string()
-                }
-                Self::Save(EcstoreError::PreconditionFailed) => {
-                    "scanner usage reset primary slot changed before bootstrap publish".to_string()
-                }
-                Self::Save(err) => format!("failed to persist scanner usage reset bootstrap: {err}"),
-            },
-        };
-        ScannerError::Other(message)
-    }
-}
-
-pub(super) async fn publish_scanner_usage_bootstrap_primary(
-    storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
-    expected_revision: &DataUsageCacheRevision,
-    expected_publication_epoch: u64,
-    leader_epoch: Option<u64>,
-    context: ScannerUsageBootstrapPublishContext,
-) -> Result<(), ScannerError> {
-    async fn inner(
-        storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
-        expected_revision: &DataUsageCacheRevision,
-        expected_publication_epoch: u64,
-        leader_epoch: Option<u64>,
-    ) -> Result<(), ScannerUsageBootstrapPublishError> {
-        let marker = scanner_usage_bootstrap_marker(std::time::SystemTime::now(), leader_epoch);
-        let data = serde_json::to_vec(&marker).map_err(ScannerUsageBootstrapPublishError::Encode)?;
-        let save_result = save_config_with_publication_admission_for_epoch(
-            storeapi.clone(),
-            DATA_USAGE_OBJ_NAME_PATH.as_str(),
-            data.clone(),
-            expected_revision.preconditions(),
-            expected_publication_epoch,
-        )
-        .await;
-        if save_result
-            .as_ref()
-            .ok()
-            .and_then(|info| info.etag.as_deref())
-            .is_some_and(|etag| !etag.is_empty())
-        {
-            return Ok(());
-        }
-
-        let (persisted, revision) = read_config_with_revision(storeapi, DATA_USAGE_OBJ_NAME_PATH.as_str())
-            .await
-            .map_err(ScannerUsageBootstrapPublishError::Reconcile)?;
-        if persisted.as_deref() == Some(data.as_slice()) && matches!(revision, DataUsageCacheRevision::Etag(_)) {
-            return Ok(());
-        }
-        Err(match save_result {
-            Ok(_) => ScannerUsageBootstrapPublishError::MissingEtag,
-            Err(err) => ScannerUsageBootstrapPublishError::Save(err),
-        })
-    }
-
-    inner(storeapi, expected_revision, expected_publication_epoch, leader_epoch)
-        .await
-        .map_err(|err| err.into_scanner_error(context))
-}
-
 pub(super) async fn reset_scanner_usage_state_slots_for_full_rebuild(
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
     slots: &[ScannerUsageStateResetSlot],
@@ -1540,15 +1442,48 @@ pub(super) async fn reset_scanner_usage_state_slots_for_full_rebuild(
         .iter()
         .find(|slot| slot.path == DATA_USAGE_OBJ_NAME_PATH.as_str())
         .ok_or_else(|| ScannerError::Other("scanner usage reset primary slot was not inspected".to_string()))?;
-    publish_scanner_usage_bootstrap_primary(
+    let marker = DataUsageInfo {
+        last_update: Some(std::time::SystemTime::now()),
+        scanner_epoch: Some(leader_epoch),
+        usage_snapshot_converged: Some(false),
+        usage_snapshot_bootstrap_pending: true,
+        ..Default::default()
+    };
+    let data = serde_json::to_vec(&marker)
+        .map_err(|err| ScannerError::Other(format!("failed to encode scanner usage reset bootstrap marker: {err}")))?;
+    let save_result = save_config_with_publication_admission_for_epoch(
         storeapi.clone(),
-        &primary.revision,
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        data.clone(),
+        primary.revision.preconditions(),
         expected_epoch,
-        Some(leader_epoch),
-        ScannerUsageBootstrapPublishContext::Reset,
     )
-    .await?;
-    reset_paths.push(DATA_USAGE_OBJ_NAME_PATH.as_str().to_string());
+    .await;
+    if save_result
+        .as_ref()
+        .ok()
+        .and_then(|info| info.etag.as_deref())
+        .is_some_and(|etag| !etag.is_empty())
+    {
+        reset_paths.push(DATA_USAGE_OBJ_NAME_PATH.as_str().to_string());
+    } else {
+        let (persisted, revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage reset bootstrap marker: {err}")))?;
+        if persisted.as_deref() != Some(data.as_slice()) || !matches!(revision, DataUsageCacheRevision::Etag(_)) {
+            return Err(ScannerError::Other(match save_result {
+                Ok(_) => "scanner usage reset bootstrap returned no ETag and could not be confirmed".to_string(),
+                Err(err) if scanner_publication_epoch_changed(&err) => {
+                    "scanner usage reset deferred by a movement epoch change".to_string()
+                }
+                Err(EcstoreError::PreconditionFailed) => {
+                    "scanner usage reset primary slot changed before bootstrap publish".to_string()
+                }
+                Err(err) => format!("failed to persist scanner usage reset bootstrap: {err}"),
+            }));
+        }
+        reset_paths.push(DATA_USAGE_OBJ_NAME_PATH.as_str().to_string());
+    }
 
     for slot in slots.iter().filter(|slot| slot.path != DATA_USAGE_OBJ_NAME_PATH.as_str()) {
         if delete_usage_state_reset_slot(storeapi.clone(), slot, expected_epoch).await? {
@@ -1632,7 +1567,7 @@ pub async fn reset_scanner_usage_state_for_full_rebuild(
     }
 
     clear_scanner_usage_floor_failure();
-    clear_legacy_incomplete_usage_floor_recovery_status();
+    clear_legacy_empty_usage_floor_recovery_status();
     set_scanner_cycle_recovery_status(recovery_status("healthy", None, false));
     super::notify_scanner_cycle_recovery_wake();
     info!(
@@ -1679,48 +1614,33 @@ pub(super) enum PersistedUsageFloorStartup {
     Authoritative,
     Missing,
     BootstrapPending,
-    RecoveredLegacyIncompleteFence,
+    RecoveredLegacyEmptyFence,
 }
 
 #[derive(Clone, Debug)]
-struct LegacyIncompleteUsageFloorPrimary {
+struct LegacyEmptyUsageFloorPrimary {
     revision: DataUsageCacheRevision,
     epoch: u64,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(deny_unknown_fields)]
-struct LegacyIncompleteUsageFloorRecoveryMarker {
+struct LegacyEmptyUsageFloorRecoveryMarker {
     schema_version: u16,
     primary_revision: String,
     leader_epoch: u64,
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-struct LegacyIncompleteUsageFence {
-    claimable_epoch: Option<u64>,
-}
-
-impl LegacyIncompleteUsageFence {
-    fn new(claimable_epoch: Option<u64>) -> Self {
-        Self { claimable_epoch }
-    }
-
-    fn claimable_epoch(self) -> Option<u64> {
-        self.claimable_epoch
-    }
-}
-
-async fn read_legacy_incomplete_usage_floor_recovery_marker(
+async fn read_legacy_empty_usage_floor_recovery_marker(
     storeapi: Arc<impl ScannerObjectIO>,
-) -> Result<Option<(LegacyIncompleteUsageFloorRecoveryMarker, DataUsageCacheRevision)>, ScannerError> {
+) -> Result<Option<(LegacyEmptyUsageFloorRecoveryMarker, DataUsageCacheRevision)>, ScannerError> {
     let (data, revision) = read_config_with_revision(storeapi, DATA_USAGE_RECOVERY_PATH.as_str())
         .await
         .map_err(|err| ScannerError::Other(format!("failed to read scanner usage recovery marker: {err}")))?;
     let Some(data) = data else {
         return Ok(None);
     };
-    let marker = serde_json::from_slice::<LegacyIncompleteUsageFloorRecoveryMarker>(&data)
+    let marker = serde_json::from_slice::<LegacyEmptyUsageFloorRecoveryMarker>(&data)
         .map_err(|err| ScannerError::Other(format!("failed to decode scanner usage recovery marker: {err}")))?;
     if marker.schema_version != 1 || marker.primary_revision.is_empty() || marker.leader_epoch == 0 {
         return Err(ScannerError::Other("scanner usage recovery marker is invalid".to_string()));
@@ -1731,7 +1651,7 @@ async fn read_legacy_incomplete_usage_floor_recovery_marker(
     Ok(Some((marker, revision)))
 }
 
-async fn clear_legacy_incomplete_usage_floor_recovery_marker(
+async fn clear_legacy_empty_usage_floor_recovery_marker(
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
     marker_revision: &DataUsageCacheRevision,
     expected_publication_epoch: u64,
@@ -1765,11 +1685,11 @@ async fn clear_legacy_incomplete_usage_floor_recovery_marker(
     }
 }
 
-pub(super) async fn complete_legacy_incomplete_usage_floor_recovery(
+pub(super) async fn complete_legacy_empty_usage_floor_recovery(
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
     claimed_epoch: u64,
 ) -> Result<(), ScannerError> {
-    let Some((marker, marker_revision)) = read_legacy_incomplete_usage_floor_recovery_marker(storeapi.clone()).await? else {
+    let Some((marker, marker_revision)) = read_legacy_empty_usage_floor_recovery_marker(storeapi.clone()).await? else {
         return Ok(());
     };
     if claimed_epoch <= marker.leader_epoch {
@@ -1789,220 +1709,12 @@ pub(super) async fn complete_legacy_incomplete_usage_floor_recovery(
     let expected_publication_epoch = scanner_publication_epoch(storeapi.clone())
         .await
         .ok_or_else(|| ScannerError::Other("scanner usage recovery cleanup is blocked by data movement".to_string()))?;
-    clear_legacy_incomplete_usage_floor_recovery_marker(storeapi, &marker_revision, expected_publication_epoch).await?;
-    clear_legacy_incomplete_usage_floor_recovery_status();
+    clear_legacy_empty_usage_floor_recovery_marker(storeapi, &marker_revision, expected_publication_epoch).await?;
+    clear_legacy_empty_usage_floor_recovery_status();
     Ok(())
 }
 
-struct LegacyOptional<T> {
-    present: bool,
-    value: Option<T>,
-}
-
-impl<T> Default for LegacyOptional<T> {
-    fn default() -> Self {
-        Self {
-            present: false,
-            value: None,
-        }
-    }
-}
-
-fn deserialize_legacy_optional<'de, D, T>(deserializer: D) -> Result<LegacyOptional<T>, D::Error>
-where
-    D: serde::Deserializer<'de>,
-    T: Deserialize<'de>,
-{
-    Option::<T>::deserialize(deserializer).map(|value| LegacyOptional { present: true, value })
-}
-
-struct LegacyUniqueMap<V>(std::collections::HashMap<String, V>);
-
-impl<'de, V> Deserialize<'de> for LegacyUniqueMap<V>
-where
-    V: Deserialize<'de>,
-{
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        struct UniqueMapVisitor<V>(std::marker::PhantomData<V>);
-
-        impl<'de, V> serde::de::Visitor<'de> for UniqueMapVisitor<V>
-        where
-            V: Deserialize<'de>,
-        {
-            type Value = LegacyUniqueMap<V>;
-
-            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-                formatter.write_str("a JSON object without duplicate keys")
-            }
-
-            fn visit_map<A>(self, mut entries: A) -> Result<Self::Value, A::Error>
-            where
-                A: serde::de::MapAccess<'de>,
-            {
-                let mut values = std::collections::HashMap::new();
-                while let Some((key, value)) = entries.next_entry::<String, V>()? {
-                    match values.entry(key) {
-                        std::collections::hash_map::Entry::Vacant(entry) => {
-                            entry.insert(value);
-                        }
-                        std::collections::hash_map::Entry::Occupied(entry) => {
-                            return Err(serde::de::Error::custom(format!("duplicate map key `{}`", entry.key())));
-                        }
-                    }
-                }
-                Ok(LegacyUniqueMap(values))
-            }
-        }
-
-        deserializer.deserialize_map(UniqueMapVisitor(std::marker::PhantomData))
-    }
-}
-
-impl<V> LegacyUniqueMap<V> {
-    fn len(&self) -> usize {
-        self.0.len()
-    }
-
-    fn get(&self, key: &str) -> Option<&V> {
-        self.0.get(key)
-    }
-
-    fn iter(&self) -> impl Iterator<Item = (&String, &V)> {
-        self.0.iter()
-    }
-
-    fn values(&self) -> impl Iterator<Item = &V> {
-        self.0.values()
-    }
-}
-
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct LegacyBucketTargetUsageWire {
-    #[serde(rename = "replication_pending_size")]
-    _replication_pending_size: serde::de::IgnoredAny,
-    #[serde(rename = "replication_failed_size")]
-    _replication_failed_size: serde::de::IgnoredAny,
-    #[serde(rename = "replicated_size")]
-    _replicated_size: serde::de::IgnoredAny,
-    #[serde(rename = "replica_size")]
-    _replica_size: serde::de::IgnoredAny,
-    #[serde(rename = "replication_pending_count")]
-    _replication_pending_count: serde::de::IgnoredAny,
-    #[serde(rename = "replication_failed_count")]
-    _replication_failed_count: serde::de::IgnoredAny,
-    #[serde(rename = "replicated_count")]
-    _replicated_count: serde::de::IgnoredAny,
-}
-
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct LegacyBucketUsageWire {
-    size: u64,
-    #[serde(rename = "replication_pending_size_v1")]
-    _replication_pending_size_v1: serde::de::IgnoredAny,
-    #[serde(rename = "replication_failed_size_v1")]
-    _replication_failed_size_v1: serde::de::IgnoredAny,
-    #[serde(rename = "replicated_size_v1")]
-    _replicated_size_v1: serde::de::IgnoredAny,
-    #[serde(rename = "replication_pending_count_v1")]
-    _replication_pending_count_v1: serde::de::IgnoredAny,
-    #[serde(rename = "replication_failed_count_v1")]
-    _replication_failed_count_v1: serde::de::IgnoredAny,
-    objects_count: u64,
-    #[serde(rename = "object_size_histogram")]
-    _object_size_histogram: LegacyUniqueMap<u64>,
-    #[serde(rename = "object_versions_histogram")]
-    _object_versions_histogram: LegacyUniqueMap<u64>,
-    versions_count: u64,
-    delete_markers_count: u64,
-    #[serde(rename = "replica_size")]
-    _replica_size: serde::de::IgnoredAny,
-    #[serde(rename = "replica_count")]
-    _replica_count: serde::de::IgnoredAny,
-    #[serde(rename = "replication_info")]
-    _replication_info: LegacyUniqueMap<LegacyBucketTargetUsageWire>,
-}
-
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct LegacyDiskUsageStatusWire {
-    #[serde(rename = "disk_id")]
-    _disk_id: serde::de::IgnoredAny,
-    #[serde(rename = "pool_index")]
-    _pool_index: serde::de::IgnoredAny,
-    #[serde(rename = "set_index")]
-    _set_index: serde::de::IgnoredAny,
-    #[serde(rename = "disk_index")]
-    _disk_index: serde::de::IgnoredAny,
-    #[serde(rename = "last_update")]
-    _last_update: serde::de::IgnoredAny,
-    #[serde(rename = "snapshot_exists")]
-    _snapshot_exists: serde::de::IgnoredAny,
-}
-
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct LegacyTierStatsWire {
-    #[serde(rename = "total_size")]
-    _total_size: serde::de::IgnoredAny,
-    #[serde(rename = "num_versions")]
-    _num_versions: serde::de::IgnoredAny,
-    #[serde(rename = "num_objects")]
-    _num_objects: serde::de::IgnoredAny,
-}
-
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct LegacyAllTierStatsWire {
-    #[serde(rename = "tiers")]
-    _tiers: LegacyUniqueMap<LegacyTierStatsWire>,
-}
-
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct LegacyUsageWire {
-    #[serde(rename = "total_capacity")]
-    _total_capacity: serde::de::IgnoredAny,
-    #[serde(rename = "total_used_capacity")]
-    _total_used_capacity: serde::de::IgnoredAny,
-    #[serde(rename = "total_free_capacity")]
-    _total_free_capacity: serde::de::IgnoredAny,
-    #[serde(rename = "last_update")]
-    _last_update: serde::de::IgnoredAny,
-    #[serde(default, deserialize_with = "deserialize_legacy_optional")]
-    scanner_epoch: LegacyOptional<u64>,
-    objects_total_count: u64,
-    versions_total_count: u64,
-    delete_markers_total_count: u64,
-    objects_total_size: u64,
-    #[serde(rename = "replication_info")]
-    _replication_info: LegacyUniqueMap<LegacyBucketTargetUsageWire>,
-    #[serde(default, deserialize_with = "deserialize_legacy_optional")]
-    tier_stats: LegacyOptional<LegacyAllTierStatsWire>,
-    buckets_count: u64,
-    buckets_usage: LegacyUniqueMap<LegacyBucketUsageWire>,
-    usage_snapshot_complete: bool,
-    bucket_sizes: LegacyUniqueMap<u64>,
-    #[serde(rename = "disk_usage_status")]
-    _disk_usage_status: Vec<LegacyDiskUsageStatusWire>,
-}
-
-fn decode_legacy_usage_wire(data: &[u8], usage: &DataUsageInfo) -> Option<LegacyUsageWire> {
-    let wire = serde_json::from_slice::<LegacyUsageWire>(data).ok()?;
-    if wire.scanner_epoch.present != usage.scanner_epoch.is_some()
-        || wire.scanner_epoch.value != usage.scanner_epoch
-        || wire.tier_stats.present != usage.tier_stats.is_some()
-    {
-        return None;
-    }
-    Some(wire)
-}
-
-fn legacy_empty_usage_fence(data: &[u8], usage: &DataUsageInfo) -> Option<LegacyIncompleteUsageFence> {
+fn legacy_empty_usage_fence_epoch(data: &[u8], usage: &DataUsageInfo) -> Option<Option<u64>> {
     if usage.last_update.is_none() || usage.scanner_cycle.is_some() {
         return None;
     }
@@ -2018,88 +1730,55 @@ fn legacy_empty_usage_fence(data: &[u8], usage: &DataUsageInfo) -> Option<Legacy
         return None;
     }
 
+    let serde_json::Value::Object(fields) = serde_json::from_slice::<serde_json::Value>(data).ok()? else {
+        return None;
+    };
     // RUSTFS_COMPAT_TODO(backlog-2102): accept only the exact empty usage fence serialized by rc.2/rc.3. Remove after those releases are no longer supported direct-upgrade sources.
-    let wire = decode_legacy_usage_wire(data, usage)?;
-    Some(LegacyIncompleteUsageFence::new(wire.scanner_epoch.value))
-}
-
-fn legacy_incomplete_usage_fence(data: &[u8], usage: &DataUsageInfo) -> Option<LegacyIncompleteUsageFence> {
-    legacy_empty_usage_fence(data, usage).or_else(|| legacy_non_empty_usage_fence(data, usage))
-}
-
-// RUSTFS_COMPAT_TODO(backlog-2122): accept rc.1-rc.3 usage floors that were fenced before a scanner cycle completed. Remove after those releases are no longer supported direct-upgrade sources.
-fn legacy_non_empty_usage_fence(data: &[u8], usage: &DataUsageInfo) -> Option<LegacyIncompleteUsageFence> {
-    if usage.last_update.is_none()
-        || usage.scanner_cycle.is_some()
-        || usage.usage_snapshot_bootstrap_pending
-        || usage.usage_snapshot_complete
-        || usage.usage_snapshot_converged.is_some()
-        || usage.usage_snapshot_authoritative_baseline.is_some()
-        || !usage.usage_snapshot_set_states.is_empty()
-        || usage.usage_snapshot_partial
-        || usage.buckets_count == 0
-        || u64::try_from(usage.buckets_usage.len()).ok() != Some(usage.buckets_count)
+    const REQUIRED_FIELDS: &[&str] = &[
+        "total_capacity",
+        "total_used_capacity",
+        "total_free_capacity",
+        "last_update",
+        "objects_total_count",
+        "versions_total_count",
+        "delete_markers_total_count",
+        "objects_total_size",
+        "replication_info",
+        "buckets_count",
+        "buckets_usage",
+        "usage_snapshot_complete",
+        "bucket_sizes",
+        "disk_usage_status",
+    ];
+    let expected_len = REQUIRED_FIELDS.len() + if usage.scanner_epoch.is_some() { 1 } else { 0 };
+    if fields.len() != expected_len
+        || REQUIRED_FIELDS.iter().any(|field| !fields.contains_key(*field))
+        || (usage.scanner_epoch.is_some() != fields.contains_key("scanner_epoch"))
     {
         return None;
     }
-    if usage.scanner_epoch.is_some_and(|epoch| epoch == 0 || epoch >= u64::MAX - 1) {
-        return None;
-    }
-    let wire = decode_legacy_usage_wire(data, usage)?;
-    if wire.usage_snapshot_complete
-        || wire.buckets_count == 0
-        || u64::try_from(wire.buckets_usage.len()).ok() != Some(wire.buckets_count)
-        || wire.bucket_sizes.len() != wire.buckets_usage.len()
-        || wire
-            .buckets_usage
-            .iter()
-            .any(|(bucket, bucket_usage)| wire.bucket_sizes.get(bucket) != Some(&bucket_usage.size))
-    {
-        return None;
-    }
-    let (objects, versions, delete_markers, size) = wire.buckets_usage.values().try_fold(
-        (0_u64, 0_u64, 0_u64, 0_u64),
-        |(objects, versions, delete_markers, size), bucket| {
-            Some((
-                objects.checked_add(bucket.objects_count)?,
-                versions.checked_add(bucket.versions_count)?,
-                delete_markers.checked_add(bucket.delete_markers_count)?,
-                size.checked_add(bucket.size)?,
-            ))
-        },
-    )?;
-    if (objects, versions, delete_markers, size)
-        != (
-            wire.objects_total_count,
-            wire.versions_total_count,
-            wire.delete_markers_total_count,
-            wire.objects_total_size,
-        )
-    {
-        return None;
-    }
-    Some(LegacyIncompleteUsageFence::new(wire.scanner_epoch.value))
+    Some(usage.scanner_epoch)
 }
 
-async fn recover_legacy_incomplete_usage_floor(
+async fn recover_legacy_empty_usage_floor(
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
-    primary: LegacyIncompleteUsageFloorPrimary,
+    primary: LegacyEmptyUsageFloorPrimary,
     expected_publication_epoch: u64,
 ) -> Result<(), ScannerError> {
     let DataUsageCacheRevision::Etag(primary_revision) = &primary.revision else {
-        return Err(ScannerError::Other("legacy incomplete scanner usage floor has no revision".to_string()));
+        return Err(ScannerError::Other("legacy empty scanner usage floor has no revision".to_string()));
     };
-    let marker = LegacyIncompleteUsageFloorRecoveryMarker {
+    let marker = LegacyEmptyUsageFloorRecoveryMarker {
         schema_version: 1,
         primary_revision: primary_revision.clone(),
         leader_epoch: primary.epoch,
     };
     let marker_data = serde_json::to_vec(&marker)
         .map_err(|err| ScannerError::Other(format!("failed to encode scanner usage recovery marker: {err}")))?;
-    match read_legacy_incomplete_usage_floor_recovery_marker(storeapi.clone()).await? {
+    match read_legacy_empty_usage_floor_recovery_marker(storeapi.clone()).await? {
         Some((persisted, _)) if persisted != marker => {
             return Err(ScannerError::Other(
-                "scanner usage recovery marker conflicts with the persisted incomplete floor".to_string(),
+                "scanner usage recovery marker conflicts with the persisted empty floor".to_string(),
             ));
         }
         Some(_) => {}
@@ -2118,7 +1797,7 @@ async fn recover_legacy_incomplete_usage_floor(
                 .and_then(|info| info.etag.as_deref())
                 .is_some_and(|etag| !etag.is_empty())
             {
-                let persisted = read_legacy_incomplete_usage_floor_recovery_marker(storeapi.clone()).await?;
+                let persisted = read_legacy_empty_usage_floor_recovery_marker(storeapi.clone()).await?;
                 if persisted.as_ref().map(|(persisted, _)| persisted) != Some(&marker) {
                     return Err(ScannerError::Other(match marker_save {
                         Ok(_) => "scanner usage recovery marker returned no ETag and could not be confirmed".to_string(),
@@ -2129,26 +1808,52 @@ async fn recover_legacy_incomplete_usage_floor(
         }
     }
 
-    publish_scanner_usage_bootstrap_primary(
+    let marker = DataUsageInfo {
+        last_update: Some(std::time::SystemTime::now()),
+        scanner_epoch: Some(primary.epoch),
+        usage_snapshot_converged: Some(false),
+        usage_snapshot_bootstrap_pending: true,
+        ..Default::default()
+    };
+    let data = serde_json::to_vec(&marker)
+        .map_err(|err| ScannerError::Other(format!("failed to encode recovered scanner usage bootstrap: {err}")))?;
+    let save_result = save_config_with_publication_admission_for_epoch(
         storeapi.clone(),
-        &primary.revision,
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        data.clone(),
+        primary.revision.preconditions(),
         expected_publication_epoch,
-        Some(primary.epoch),
-        ScannerUsageBootstrapPublishContext::Recovery,
     )
-    .await?;
-    warn!(
-        target: "rustfs::scanner",
-        event = EVENT_SCANNER_PERSIST_STATE,
-        component = LOG_COMPONENT_SCANNER,
-        subsystem = LOG_SUBSYSTEM_RUNTIME,
-        // Keep the published state stable for existing empty-floor alerts.
-        state = "legacy_empty_usage_floor_recovered",
-        path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        scanner_epoch = primary.epoch,
-        "Scanner recovered a legacy incomplete usage floor"
-    );
-    Ok(())
+    .await;
+    if save_result
+        .as_ref()
+        .ok()
+        .and_then(|info| info.etag.as_deref())
+        .is_some_and(|etag| !etag.is_empty())
+    {
+        warn!(
+            target: "rustfs::scanner",
+            event = EVENT_SCANNER_PERSIST_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_RUNTIME,
+            state = "legacy_empty_usage_floor_recovered",
+            path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
+            scanner_epoch = primary.epoch,
+            "Scanner recovered a legacy empty usage floor"
+        );
+        return Ok(());
+    }
+
+    let (persisted, revision) = read_config_with_revision(storeapi, DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .map_err(|err| ScannerError::Other(format!("failed to reconcile recovered scanner usage bootstrap: {err}")))?;
+    if persisted.as_deref() == Some(data.as_slice()) && matches!(revision, DataUsageCacheRevision::Etag(_)) {
+        return Ok(());
+    }
+    Err(ScannerError::Other(match save_result {
+        Ok(_) => "recovered scanner usage bootstrap returned no ETag and could not be confirmed".to_string(),
+        Err(err) => format!("failed to recover legacy empty scanner usage floor: {err}"),
+    }))
 }
 
 pub(super) fn encode_scanner_cycle_state(
@@ -2340,8 +2045,8 @@ fn resolve_bootstrap_backup_slot(
             if backup_epoch >= primary_epoch {
                 update_persisted_usage_floor(resolution.floor, slot.usage, slot.backup_path)?;
             }
-        } else if let Some(fence) = legacy_incomplete_usage_fence(slot.data, slot.usage) {
-            if let Some(epoch) = fence.claimable_epoch() {
+        } else if let Some(epoch) = legacy_empty_usage_fence_epoch(slot.data, slot.usage) {
+            if let Some(epoch) = epoch {
                 resolution.floor.leader_epoch = resolution.floor.leader_epoch.max(epoch);
             }
         } else {
@@ -2351,12 +2056,10 @@ fn resolve_bootstrap_backup_slot(
         }
         return Ok(BootstrapBackupAction::Resume);
     }
-    let compatible_incomplete_fence = legacy_incomplete_usage_fence(slot.data, slot.usage).is_some_and(|fence| {
-        fence
-            .claimable_epoch()
-            .is_none_or(|epoch| slot.bootstrap_epoch.is_some_and(|bootstrap_epoch| epoch <= bootstrap_epoch))
+    let compatible_empty_fence = legacy_empty_usage_fence_epoch(slot.data, slot.usage).is_some_and(|epoch| {
+        epoch.is_none_or(|epoch| slot.bootstrap_epoch.is_some_and(|bootstrap_epoch| epoch <= bootstrap_epoch))
     });
-    if compatible_incomplete_fence {
+    if compatible_empty_fence {
         return Ok(BootstrapBackupAction::Resume);
     }
     if slot.recovered_bootstrap && data_usage_info_has_persisted_baseline_identity(slot.usage) {
@@ -2381,7 +2084,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
     let Some(read_epoch) = scanner_publication_epoch(storeapi.clone()).await else {
         return Err(ScannerError::Other("scanner usage floor read is blocked by data movement".to_string()));
     };
-    let recovery_marker = read_legacy_incomplete_usage_floor_recovery_marker(storeapi.clone()).await?;
+    let recovery_marker = read_legacy_empty_usage_floor_recovery_marker(storeapi.clone()).await?;
     let mut floor = PersistedUsageFloor::default();
     let mut found_any = false;
     let mut bootstrap_pending = false;
@@ -2396,7 +2099,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
     let mut invalid_baseline_epoch = recovery_marker.as_ref().map(|(marker, _)| marker.leader_epoch);
     let mut unrecoverable_baseline_path: Option<String> = None;
     let mut stale_authoritative_path: Option<String> = None;
-    let mut legacy_incomplete_primary: Option<LegacyIncompleteUsageFloorPrimary> = None;
+    let mut legacy_empty_primary: Option<LegacyEmptyUsageFloorPrimary> = None;
     for primary_path in [DATA_USAGE_OBJ_NAME_PATH.as_str(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()] {
         let backup_path = format!("{primary_path}.bkp");
         let is_v2_path = primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str();
@@ -2445,12 +2148,14 @@ pub(super) async fn persisted_usage_floor_for_startup(
                     } else if !data_usage_info_has_persisted_baseline_identity(&usage) {
                         invalid_baseline_path.get_or_insert_with(|| primary_path.to_string());
                         invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch);
-                        if let Some(fence) = legacy_incomplete_usage_fence(&data, &usage) {
-                            if is_v2_path && let Some(epoch) = fence.claimable_epoch() {
-                                legacy_incomplete_primary = Some(LegacyIncompleteUsageFloorPrimary { revision, epoch });
+                        match legacy_empty_usage_fence_epoch(&data, &usage) {
+                            Some(Some(epoch)) if is_v2_path => {
+                                legacy_empty_primary = Some(LegacyEmptyUsageFloorPrimary { revision, epoch });
+                            }
+                            Some(_) => {}
+                            None => {
+                                unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string());
                             }
-                        } else {
-                            unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string());
                         }
                         None
                     } else {
@@ -2538,7 +2243,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
                 if !data_usage_info_has_persisted_baseline_identity(&usage) {
                     invalid_baseline_path.get_or_insert_with(|| backup_path.clone());
                     invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch);
-                    if legacy_incomplete_usage_fence(&data, &usage).is_none() {
+                    if legacy_empty_usage_fence_epoch(&data, &usage).is_none() {
                         unrecoverable_baseline_path.get_or_insert_with(|| backup_path.clone());
                     }
                     // This is still persisted state, so it must not enable a
@@ -2589,18 +2294,17 @@ pub(super) async fn persisted_usage_floor_for_startup(
         if allow_missing_for_bootstrap
             && unrecoverable_baseline_path.is_none()
             && stale_authoritative_path.is_none()
-            && let Some(mut primary) = legacy_incomplete_primary
+            && let Some(mut primary) = legacy_empty_primary
         {
             primary.epoch = primary.epoch.max(invalid_baseline_epoch.unwrap_or_default());
-            let leader_epoch = primary.epoch;
-            recover_legacy_incomplete_usage_floor(storeapi.clone(), primary, read_epoch).await?;
-            record_legacy_incomplete_usage_floor_recovery_pending(leader_epoch);
+            recover_legacy_empty_usage_floor(storeapi.clone(), primary.clone(), read_epoch).await?;
+            record_legacy_empty_usage_floor_recovery_pending(primary.epoch);
             return Ok((
                 PersistedUsageFloor {
                     next_cycle: 0,
-                    leader_epoch,
+                    leader_epoch: primary.epoch,
                 },
-                PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence,
+                PersistedUsageFloorStartup::RecoveredLegacyEmptyFence,
             ));
         }
         if let Some(path) = stale_authoritative_path {
@@ -2613,7 +2317,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
         }
         if let Some(path) = invalid_baseline_path {
             return Err(ScannerError::Other(format!(
-                "persisted scanner usage floor from {path} has no authoritative baseline or newer valid backup; recover with POST /rustfs/admin/v3/scanner/usage-state/reset using mode full-rebuild"
+                "persisted scanner usage floor from {path} has no authoritative baseline or newer valid backup"
             )));
         }
         if !allow_missing_for_bootstrap {
@@ -2665,8 +2369,8 @@ pub(super) async fn persisted_usage_floor_for_startup(
             .as_ref()
             .map(|(marker, _)| marker.leader_epoch)
             .unwrap_or(floor.leader_epoch);
-        record_legacy_incomplete_usage_floor_recovery_pending(recovery_epoch);
-        PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence
+        record_legacy_empty_usage_floor_recovery_pending(recovery_epoch);
+        PersistedUsageFloorStartup::RecoveredLegacyEmptyFence
     } else if bootstrap_pending {
         if let Some(path) = unrecoverable_baseline_path {
             return Err(ScannerError::Other(format!(
@@ -2680,7 +2384,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
     if found_any && let Some((_, marker_revision)) = recovery_marker.as_ref() {
         drop(publication_admission);
         let marker_cleared =
-            match clear_legacy_incomplete_usage_floor_recovery_marker(storeapi.clone(), marker_revision, read_epoch).await {
+            match clear_legacy_empty_usage_floor_recovery_marker(storeapi.clone(), marker_revision, read_epoch).await {
                 Ok(()) => true,
                 Err(err) => {
                     warn!(
@@ -2703,7 +2407,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
             ));
         };
         if marker_cleared {
-            clear_legacy_incomplete_usage_floor_recovery_status();
+            clear_legacy_empty_usage_floor_recovery_status();
         }
         clear_scanner_usage_floor_failure();
         return Ok((floor, state));

crates/scanner/src/scanner/leadership.rs

@@ -185,14 +185,42 @@ pub(super) async fn initialize_usage_baseline_bootstrap(
             "scanner usage baseline bootstrap is blocked by data movement".to_string(),
         ));
     };
-    publish_scanner_usage_bootstrap_primary(
-        storeapi,
-        &DataUsageCacheRevision::Missing,
+    let baseline = DataUsageInfo {
+        last_update: Some(std::time::SystemTime::now()),
+        usage_snapshot_converged: Some(false),
+        usage_snapshot_bootstrap_pending: true,
+        ..Default::default()
+    };
+    let data = serde_json::to_vec(&baseline)
+        .map_err(|err| ScannerError::Other(format!("failed to encode scanner usage baseline bootstrap: {err}")))?;
+    let save_result = save_config_with_publication_admission_for_epoch(
+        storeapi.clone(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        data.clone(),
+        DataUsageCacheRevision::Missing.preconditions(),
         expected_epoch,
-        None,
-        ScannerUsageBootstrapPublishContext::Initial,
     )
-    .await
+    .await;
+    if save_result
+        .as_ref()
+        .ok()
+        .and_then(|info| info.etag.as_deref())
+        .is_some_and(|etag| !etag.is_empty())
+    {
+        return Ok(());
+    }
+
+    let (persisted, revision) = read_config_with_revision(storeapi, DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage bootstrap: {err}")))?;
+    if persisted.as_deref() == Some(data.as_slice()) && matches!(revision, DataUsageCacheRevision::Etag(_)) {
+        return Ok(());
+    }
+
+    Err(ScannerError::Other(match save_result {
+        Ok(_) => "scanner usage bootstrap returned no ETag and could not be confirmed".to_string(),
+        Err(err) => format!("failed to persist scanner usage bootstrap: {err}"),
+    }))
 }
 
 pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(

crates/scanner/src/scanner/tests.rs

@@ -462,7 +462,6 @@ async fn cycle_budget_persist_cursor_failure_is_recovery_required() {
         &mut cycle,
         &mut revision,
         &mut leader_epoch,
-        false,
         std::future::pending(),
     )
     .await;
@@ -479,52 +478,6 @@ async fn cycle_budget_persist_cursor_failure_is_recovery_required() {
     assert!(report.leader_lease_without_progress);
 }
 
-#[tokio::test]
-async fn cycle_budget_fence_accepts_bootstrap_pending_usage_marker() {
-    let store = Arc::new(MemoryConfigStore::default());
-    initialize_usage_baseline_bootstrap(store.clone())
-        .await
-        .expect("usage reset should publish a bootstrap marker");
-
-    let ctx = CancellationToken::new();
-    let mut revision = DataUsageCacheRevision::Missing;
-    let mut cycle = CurrentCycle {
-        current: 12,
-        next: 12,
-        ..Default::default()
-    };
-    let mut leader_epoch = 0;
-
-    let fenced = fence_scanner_epoch_after_cycle_timeout(
-        &ctx,
-        store.clone(),
-        &mut cycle,
-        &mut revision,
-        &mut leader_epoch,
-        true,
-        std::future::pending(),
-    )
-    .await;
-
-    assert!(fenced, "a valid reset bootstrap marker must not force cycle recovery after budget expiry");
-    assert!(!cycle_timeout_requires_recovery(true, true, fenced));
-    assert_eq!(leader_epoch, 1);
-
-    let persisted_cycle = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
-        .await
-        .expect("timeout fence should persist the next leader epoch");
-    let (_, persisted_epoch) = decode_scanner_cycle_state(&persisted_cycle).expect("persisted epoch fence should decode");
-    assert_eq!(persisted_epoch, 1);
-
-    let usage = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
-        .await
-        .expect("timeout fence should keep the bootstrap usage marker");
-    let usage = serde_json::from_slice::<DataUsageInfo>(&usage).expect("bootstrap marker should decode");
-    assert!(data_usage_info_is_bootstrap_pending(&usage));
-    assert_eq!(usage.scanner_epoch, Some(1));
-    assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
-}
-
 #[tokio::test]
 async fn cycle_budget_deadline_handler_fences_and_releases_guard() {
     let (_temp_dir, store) = setup_scanner_cycle_store().await;
@@ -562,7 +515,6 @@ async fn cycle_budget_deadline_handler_fences_and_releases_guard() {
             cycle_revision: &mut cycle_revision,
             leader_epoch: &mut leader_epoch,
             cycle_budget: &budget,
-            allow_bootstrap_pending: false,
         },
         true,
         &mut guard,
@@ -2285,53 +2237,6 @@ fn rc3_legacy_empty_usage_fence(epoch: Option<u64>) -> Vec<u8> {
     serde_json::to_vec(&value).expect("rc.3 legacy empty usage fence fixture should encode")
 }
 
-fn rc3_legacy_non_empty_usage_fence(epoch: Option<u64>) -> Vec<u8> {
-    // Pinned rc.3 field set. Leadership preserved this data and added only
-    // scanner_epoch when the producing scanner cycle had not completed.
-    const RC3_NON_EMPTY_USAGE_FENCE: &str = r#"{
-        "total_capacity":2000000000,
-        "total_used_capacity":1000000000,
-        "total_free_capacity":1000000000,
-        "last_update":{"secs_since_epoch":1,"nanos_since_epoch":0},
-        "objects_total_count":156382067,
-        "versions_total_count":156382070,
-        "delete_markers_total_count":3,
-        "objects_total_size":987654321,
-        "replication_info":{},
-        "buckets_count":1,
-        "buckets_usage":{
-            "photos":{
-                "size":987654321,
-                "replication_pending_size_v1":0,
-                "replication_failed_size_v1":0,
-                "replicated_size_v1":0,
-                "replication_pending_count_v1":0,
-                "replication_failed_count_v1":0,
-                "objects_count":156382067,
-                "object_size_histogram":{},
-                "object_versions_histogram":{},
-                "versions_count":156382070,
-                "delete_markers_count":3,
-                "replica_size":0,
-                "replica_count":0,
-                "replication_info":{}
-            }
-        },
-        "usage_snapshot_complete":false,
-        "bucket_sizes":{"photos":987654321},
-        "disk_usage_status":[]
-    }"#;
-    let mut value = serde_json::from_str::<serde_json::Value>(RC3_NON_EMPTY_USAGE_FENCE)
-        .expect("pinned rc.3 non-empty usage fence should decode");
-    if let Some(epoch) = epoch {
-        value
-            .as_object_mut()
-            .expect("legacy non-empty usage fence should be a JSON object")
-            .insert("scanner_epoch".to_string(), serde_json::Value::from(epoch));
-    }
-    serde_json::to_vec(&value).expect("rc.3 legacy non-empty usage fence fixture should encode")
-}
-
 #[tokio::test]
 async fn scanner_usage_floor_recovers_rc3_empty_fences_and_preserves_cycle_number() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -2360,7 +2265,7 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_preserves_cycle_numbe
             leader_epoch: 7,
         }
     );
-    assert_eq!(startup, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
+    assert_eq!(startup, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
 
     let primary = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
         .await
@@ -2374,7 +2279,7 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_preserves_cycle_numbe
         .await
         .expect("recovery marker should survive a restart before leadership claim");
     assert_eq!(restart_floor.leader_epoch, 7);
-    assert_eq!(restart_state, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
+    assert_eq!(restart_state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
     let mut cycle = CurrentCycle {
         current: 17_117,
         next: 17_118,
@@ -2463,7 +2368,7 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_preserves_cycle_numbe
         assert!(!persisted_reset.info.snapshot_complete);
         assert!(persisted_reset.cache.is_empty());
     }
-    complete_legacy_incomplete_usage_floor_recovery(store.clone(), leader_epoch)
+    complete_legacy_empty_usage_floor_recovery(store.clone(), leader_epoch)
         .await
         .expect("leadership claim should retire the recovery marker");
     assert!(matches!(
@@ -2477,277 +2382,6 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_preserves_cycle_numbe
     assert_eq!(claimed_state, PersistedUsageFloorStartup::BootstrapPending);
 }
 
-#[tokio::test]
-async fn scanner_usage_floor_recovers_rc3_non_empty_incomplete_fence() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let primary_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
-    store
-        .objects
-        .lock()
-        .await
-        .insert(primary_key.clone(), rc3_legacy_non_empty_usage_fence(Some(13)));
-    store.revisions.lock().await.insert(primary_key, 1);
-
-    save_config(
-        store.clone(),
-        LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        rc3_legacy_non_empty_usage_fence(None),
-    )
-    .await
-    .expect("legacy usage should persist");
-
-    let (floor, startup) = persisted_usage_floor_for_startup(store.clone(), true)
-        .await
-        .expect("rc.3 non-empty incomplete fence should enter recovery");
-    assert_eq!(
-        floor,
-        PersistedUsageFloor {
-            next_cycle: 0,
-            leader_epoch: 13,
-        }
-    );
-    assert_eq!(startup, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
-
-    let primary = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-        .await
-        .expect("recovered usage bootstrap should replace the old floor");
-    let pending = serde_json::from_slice::<DataUsageInfo>(&primary).expect("recovered usage bootstrap should decode");
-    assert!(data_usage_info_is_bootstrap_pending(&pending));
-    assert!(!data_usage_info_has_persisted_baseline_identity(&pending));
-    assert_eq!(pending.scanner_epoch, Some(13));
-    assert!(read_config(store, DATA_USAGE_RECOVERY_PATH.as_str()).await.is_ok());
-}
-
-#[tokio::test]
-async fn scanner_usage_floor_prefers_newer_backup_over_rc3_non_empty_incomplete_fence() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let primary_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
-    store
-        .objects
-        .lock()
-        .await
-        .insert(primary_key, rc3_legacy_non_empty_usage_fence(Some(13)));
-
-    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 2);
-    backup.scanner_epoch = Some(14);
-    backup.scanner_cycle = Some(9845);
-    save_config(
-        store.clone(),
-        &format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
-        serde_json::to_vec(&backup).expect("newer backup should encode"),
-    )
-    .await
-    .expect("newer backup should persist");
-
-    let (floor, startup) = persisted_usage_floor_for_startup(store.clone(), true)
-        .await
-        .expect("newer authoritative backup should win over the old incomplete floor");
-    assert_eq!(
-        floor,
-        PersistedUsageFloor {
-            next_cycle: 9846,
-            leader_epoch: 14,
-        }
-    );
-    assert_eq!(startup, PersistedUsageFloorStartup::Authoritative);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn scanner_usage_floor_rejects_noncanonical_non_empty_incomplete_fences() {
-    let base = serde_json::from_slice::<serde_json::Value>(&rc3_legacy_non_empty_usage_fence(Some(13)))
-        .expect("pinned rc.3 usage fence should decode");
-    let mut cases = Vec::new();
-
-    let mut unknown_top_level = base.clone();
-    unknown_top_level["future_field"] = serde_json::Value::Bool(true);
-    cases.push(("unknown top-level field", unknown_top_level));
-
-    let mut unknown_bucket_field = base.clone();
-    unknown_bucket_field["buckets_usage"]["photos"]["future_field"] = serde_json::Value::Bool(true);
-    cases.push(("unknown bucket field", unknown_bucket_field));
-
-    let mut wrong_bucket_size = base.clone();
-    wrong_bucket_size["bucket_sizes"]["photos"] = serde_json::Value::from(987_654_320_u64);
-    cases.push(("bucket size mismatch", wrong_bucket_size));
-
-    let mut wrong_total = base.clone();
-    wrong_total["objects_total_count"] = serde_json::Value::from(156_382_068_u64);
-    cases.push(("object total mismatch", wrong_total));
-
-    let mut wrong_versions = base.clone();
-    wrong_versions["versions_total_count"] = serde_json::Value::from(156_382_071_u64);
-    cases.push(("version total mismatch", wrong_versions));
-
-    let mut wrong_delete_markers = base.clone();
-    wrong_delete_markers["delete_markers_total_count"] = serde_json::Value::from(4_u64);
-    cases.push(("delete marker total mismatch", wrong_delete_markers));
-
-    let mut wrong_total_size = base.clone();
-    wrong_total_size["objects_total_size"] = serde_json::Value::from(987_654_320_u64);
-    cases.push(("object size total mismatch", wrong_total_size));
-
-    let mut wrong_cardinality = base.clone();
-    wrong_cardinality["buckets_count"] = serde_json::Value::from(2_u64);
-    cases.push(("bucket cardinality mismatch", wrong_cardinality));
-
-    let mut overflow = base.clone();
-    let mut overflow_bucket = overflow["buckets_usage"]["photos"].clone();
-    overflow_bucket["objects_count"] = serde_json::Value::from(u64::MAX);
-    overflow_bucket["size"] = serde_json::Value::from(0_u64);
-    overflow["buckets_usage"]["overflow"] = overflow_bucket;
-    overflow["bucket_sizes"]["overflow"] = serde_json::Value::from(0_u64);
-    overflow["buckets_count"] = serde_json::Value::from(2_u64);
-    cases.push(("checked total overflow", overflow));
-
-    let mut invalid_epoch = base;
-    invalid_epoch["scanner_epoch"] = serde_json::Value::from(0_u64);
-    cases.push(("invalid epoch", invalid_epoch));
-
-    for (case, value) in cases {
-        let store = Arc::new(MemoryConfigStore::default());
-        let primary_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
-        let original = serde_json::to_vec(&value).expect("noncanonical usage fixture should encode");
-        store.objects.lock().await.insert(primary_key.clone(), original.clone());
-        store.revisions.lock().await.insert(primary_key, 1);
-
-        let err = persisted_usage_floor_for_startup(store.clone(), true)
-            .await
-            .expect_err("noncanonical incomplete usage must remain fail-closed");
-        assert!(err.to_string().contains("usage-state/reset"), "unexpected error for {case}: {err}");
-        assert_eq!(
-            read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-                .await
-                .expect("rejected usage primary should remain"),
-            original,
-            "rejected primary changed for {case}"
-        );
-        assert!(
-            matches!(
-                read_config(store, DATA_USAGE_RECOVERY_PATH.as_str()).await,
-                Err(EcstoreError::ConfigNotFound)
-            ),
-            "recovery marker should not be written for {case}"
-        );
-    }
-}
-
-#[tokio::test]
-async fn scanner_usage_floor_rejects_duplicate_legacy_fields() {
-    let base = String::from_utf8(rc3_legacy_non_empty_usage_fence(Some(13))).expect("pinned rc.3 usage fence should be UTF-8");
-    let base_value = serde_json::from_str::<serde_json::Value>(&base).expect("pinned rc.3 usage fence should decode");
-    let duplicate_top_level = base.replacen(
-        "\"objects_total_count\":156382067",
-        "\"objects_total_count\":156382067,\"objects_total_count\":156382067",
-        1,
-    );
-    let duplicate_bucket = base.replacen("\"size\":987654321", "\"size\":987654321,\"size\":987654321", 1);
-    let bucket = serde_json::to_string(&base_value["buckets_usage"]["photos"]).expect("pinned rc.3 bucket usage should encode");
-    let bucket_map = format!("\"buckets_usage\":{{\"photos\":{bucket}}}");
-    let duplicate_bucket_key =
-        base.replacen(&bucket_map, &format!("\"buckets_usage\":{{\"photos\":{bucket},\"photos\":{bucket}}}"), 1);
-    let duplicate_bucket_size_key = base.replacen(
-        "\"bucket_sizes\":{\"photos\":987654321}",
-        "\"bucket_sizes\":{\"photos\":987654321,\"photos\":987654321}",
-        1,
-    );
-    let duplicate_histogram_key =
-        base.replacen("\"object_size_histogram\":{}", "\"object_size_histogram\":{\"small\":1,\"small\":1}", 1);
-    let mut duplicate_target_field = base.clone();
-    let target_map = "\"replication_info\":{\"target\":{\"replication_pending_size\":0,\"replication_failed_size\":0,\"replicated_size\":0,\"replica_size\":0,\"replication_pending_count\":0,\"replication_failed_count\":0,\"replicated_count\":0,\"replicated_count\":0}}";
-    let target_offset = duplicate_target_field
-        .rfind("\"replication_info\":{}")
-        .expect("pinned fixture should contain bucket replication info");
-    duplicate_target_field.replace_range(target_offset..target_offset + "\"replication_info\":{}".len(), target_map);
-    let target = "{\"replication_pending_size\":0,\"replication_failed_size\":0,\"replicated_size\":0,\"replica_size\":0,\"replication_pending_count\":0,\"replication_failed_count\":0,\"replicated_count\":0}";
-    let mut duplicate_target_key = base.clone();
-    let target_offset = duplicate_target_key
-        .rfind("\"replication_info\":{}")
-        .expect("pinned fixture should contain bucket replication info");
-    let target_map = format!("\"replication_info\":{{\"target\":{target},\"target\":{target}}}");
-    duplicate_target_key.replace_range(target_offset..target_offset + "\"replication_info\":{}".len(), &target_map);
-
-    let tier = "{\"total_size\":0,\"num_versions\":0,\"num_objects\":0}";
-    let duplicate_tier_key = base.replacen(
-        '{',
-        &format!("{{\"tier_stats\":{{\"tiers\":{{\"STANDARD\":{tier},\"STANDARD\":{tier}}}}},"),
-        1,
-    );
-
-    for (case, original, expected_error) in [
-        ("top-level field", duplicate_top_level.into_bytes(), "duplicate field"),
-        ("bucket field", duplicate_bucket.into_bytes(), "duplicate field"),
-        ("replication target field", duplicate_target_field.into_bytes(), "duplicate field"),
-        ("bucket map key", duplicate_bucket_key.into_bytes(), "usage-state/reset"),
-        ("bucket size map key", duplicate_bucket_size_key.into_bytes(), "usage-state/reset"),
-        ("histogram map key", duplicate_histogram_key.into_bytes(), "usage-state/reset"),
-        ("replication target map key", duplicate_target_key.into_bytes(), "usage-state/reset"),
-        ("tier map key", duplicate_tier_key.into_bytes(), "usage-state/reset"),
-    ] {
-        let store = Arc::new(MemoryConfigStore::default());
-        let primary_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
-        store.objects.lock().await.insert(primary_key.clone(), original.clone());
-        store.revisions.lock().await.insert(primary_key, 1);
-
-        let err = match persisted_usage_floor_for_startup(store.clone(), true).await {
-            Err(err) => err,
-            Ok(result) => panic!("duplicate {case} must remain fail-closed: {result:?}"),
-        };
-        assert!(err.to_string().contains(expected_error), "unexpected error for {case}: {err}");
-        assert_eq!(
-            read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-                .await
-                .expect("rejected duplicate-field primary should remain"),
-            original,
-            "rejected primary changed for {case}"
-        );
-        assert!(
-            matches!(
-                read_config(store, DATA_USAGE_RECOVERY_PATH.as_str()).await,
-                Err(EcstoreError::ConfigNotFound)
-            ),
-            "recovery marker should not be written for {case}"
-        );
-    }
-}
-
-#[tokio::test]
-async fn scanner_usage_floor_rejects_current_schema_incomplete_snapshot() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let primary_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
-    let mut current = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 2);
-    current.usage_snapshot_complete = false;
-    current.scanner_epoch = Some(13);
-    current.scanner_cycle = None;
-    let original = serde_json::to_vec(&current).expect("current incomplete usage should encode");
-    assert!(
-        serde_json::from_slice::<serde_json::Value>(&original)
-            .expect("current incomplete usage should decode")
-            .get("usage_snapshot_partial")
-            .is_some()
-    );
-    store.objects.lock().await.insert(primary_key.clone(), original.clone());
-    store.revisions.lock().await.insert(primary_key, 1);
-
-    let err = persisted_usage_floor_for_startup(store.clone(), true)
-        .await
-        .expect_err("current schema incomplete usage must remain fail-closed");
-    assert!(err.to_string().contains("usage-state/reset"), "unexpected error: {err}");
-    assert_eq!(
-        read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-            .await
-            .expect("rejected current usage primary should remain"),
-        original
-    );
-    assert!(matches!(
-        read_config(store, DATA_USAGE_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
 #[tokio::test]
 async fn scanner_usage_floor_recovery_preserves_newer_authoritative_companion_floor() {
     for companion_path in [
@@ -2796,7 +2430,7 @@ async fn scanner_usage_floor_recovery_preserves_newer_authoritative_companion_fl
             .expect("a newer authoritative companion should advance the recovery floor");
         assert_eq!(floor.leader_epoch, 8, "unexpected companion path: {companion_path}");
         assert_eq!(floor.next_cycle, 12, "unexpected companion path: {companion_path}");
-        assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
+        assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
     }
 }
 
@@ -2851,7 +2485,7 @@ async fn scanner_usage_floor_recovery_fences_non_authoritative_legacy_backup() {
                 .expect("an exact empty backup should contribute its epoch fence");
             assert_eq!(floor.leader_epoch, 9);
             assert_eq!(floor.next_cycle, 12);
-            assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
+            assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
         }
     }
 }
@@ -2880,7 +2514,7 @@ async fn scanner_usage_floor_recovery_resumes_after_marker_only_crash_point() {
         .await
         .expect("the durable marker should resume the primary conversion");
     assert_eq!(floor.leader_epoch, 7);
-    assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
+    assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
     let recovered = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
         .await
         .expect("recovered bootstrap should replace the legacy primary");
@@ -2906,7 +2540,7 @@ async fn scanner_usage_floor_recovery_reconciles_marker_post_commit_error() {
         .await
         .expect("a committed recovery marker should reconcile after an ambiguous error");
     assert_eq!(floor.leader_epoch, 7);
-    assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
+    assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
     assert!(read_config(store, DATA_USAGE_RECOVERY_PATH.as_str()).await.is_ok());
 }
 
@@ -2945,7 +2579,7 @@ async fn scanner_usage_floor_recovery_reconciles_marker_delete_post_commit_error
         .await
         .insert(memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_RECOVERY_PATH.as_str()));
 
-    complete_legacy_incomplete_usage_floor_recovery(store.clone(), leader_epoch)
+    complete_legacy_empty_usage_floor_recovery(store.clone(), leader_epoch)
         .await
         .expect("a committed marker delete should reconcile after an ambiguous error");
     assert!(matches!(
@@ -2987,12 +2621,12 @@ async fn scanner_usage_floor_recovery_retry_budget_uses_marker_epoch_identity()
         .await
         .expect("claimed bootstrap should retain its recovery identity");
     assert_eq!(floor.leader_epoch, 8);
-    assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyIncompleteFence);
+    assert_eq!(state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
     let status = scanner_cycle_recovery_status();
     assert_eq!(status.leader_epoch, Some(7));
     assert_eq!(status.retry_count, 3);
     assert_eq!(status.first_detected_at_unix_secs, first_detected);
-    clear_legacy_incomplete_usage_floor_recovery_status();
+    clear_legacy_empty_usage_floor_recovery_status();
 }
 
 #[tokio::test]
@@ -3148,7 +2782,7 @@ fn scanner_usage_floor_failure_is_exposed_and_cleared() {
 #[test]
 #[serial]
 fn scanner_usage_floor_recovery_stays_retryable_until_claim_cleanup() {
-    record_legacy_incomplete_usage_floor_recovery_pending(7);
+    record_legacy_empty_usage_floor_recovery_pending(7);
     let pending = scanner_cycle_recovery_status();
     assert_eq!(pending.state, "usage_floor_recovery_pending");
     assert_eq!(pending.classification.as_deref(), Some("legacy_empty_usage_floor"));
@@ -3158,12 +2792,12 @@ fn scanner_usage_floor_recovery_stays_retryable_until_claim_cleanup() {
     let first_detected = pending.first_detected_at_unix_secs;
 
     assert!(record_scanner_cycle_recovery_retry(3));
-    record_legacy_incomplete_usage_floor_recovery_pending(7);
+    record_legacy_empty_usage_floor_recovery_pending(7);
     let retried = scanner_cycle_recovery_status();
     assert_eq!(retried.retry_count, 3);
     assert_eq!(retried.first_detected_at_unix_secs, first_detected);
 
-    clear_legacy_incomplete_usage_floor_recovery_status();
+    clear_legacy_empty_usage_floor_recovery_status();
     assert_eq!(scanner_cycle_recovery_status().state, "healthy");
 }
 

docs/architecture/compat-cleanup-register.md

@@ -13,7 +13,6 @@
 
 - `tokio-tar-extension-limits` bounded archive parser hardening: Snowball extraction depends on per-entry and cumulative GNU long-name, GNU long-link, and PAX extension limits; physical-entry, GNU sparse-map, and sparse-continuation limits; cancellation-safe sparse parsing; and fused entry streams after parser errors. The released tokio-tar API does not provide this complete boundary. Keep the reviewed fork pin until astral-sh/tokio-tar#118 is merged and one published tokio-tar release contains every listed capability with the Snowball regression fixtures passing against that release.
 - `backlog-2102` rc.2/rc.3 empty scanner usage floor recovery: old DeleteBucket cleanup could synthesize an empty incomplete v2 usage primary/backup before leadership added an epoch, while newer scanners require a durable authoritative baseline identity. New scanners recognize only that exact serialized empty-fence shape, preserve its epoch through a CAS-protected recovery marker, and rebuild namespace coverage without treating zero usage as authoritative. Remove this recovery path and marker after rc.2 and rc.3 are no longer supported direct-upgrade sources.
-- `backlog-2122` rc.1-rc.3 non-empty scanner usage floor recovery: leadership fencing in those releases can stamp scanner_epoch onto a real bucket-usage snapshot before any scanner cycle completed, leaving a non-empty floor with no scanner_cycle and no authoritative baseline identity. New scanners recognize only this consistent incomplete fenced shape, preserve the epoch through the CAS-protected recovery marker, and rebuild namespace coverage without treating the old usage data as authoritative. Remove this recovery path after rc.1, rc.2, and rc.3 are no longer supported direct-upgrade sources.
 - `s3gate-metadata-xml` persisted bucket XML migration: mixed-version site-replication peers, retained `.metadata.bin` objects, and backup archives can all carry XML written by the s3s codec, so the gateway migration must keep the legacy codec available until every stored form has crossed a verified rewrite boundary. Remove the legacy s3s parser and serializer only after the minimum supported direct-upgrade release reads and writes every persisted XML configuration family through the gateway codec, every supported mixed-version site-replication topology has completed its writer upgrade, and migration tooling has verified or rewritten every retained bucket metadata object and restorable backup archive.
 - `rustfs-6339` legacy bucket policy ID casing: earlier RustFS releases persisted the top-level policy identifier as "ID", while current writes use the S3-compatible "Id" spelling. Readers accept both spellings so retained bucket metadata remains usable after upgrade. Remove the legacy alias after migration tooling has rewritten every retained bucket policy using "ID".
 - `table-publication-fence-v1` table publication fencing: nodes that predate table and table-bucket publication fences can mutate live files while a new node is publishing a catalog pointer. New nodes retain exact object guards until the operator confirms that every serving node uses the new fences. Fleet confirmation also requires non-overlapping active warehouse prefixes and lifecycle workers that exclude table buckets. Remove the exact live-file fallback and the fleet-confirmation gate after the minimum supported RustFS release acquires table fences for registered-table mutations and table-bucket fences for unresolved-prefix mutations.

docs/operations/outbound-connection-policy.md

@@ -18,7 +18,7 @@ RustFS validates every operator-configured outbound destination to close a serve
 | Target configuration validation (startup and admin API) | Full policy | `crates/targets/src/config/common.rs` `validate_outbound_http_url`; `rustfs/src/admin/handlers/target_descriptor.rs` |
 | OIDC discovery, JWKS, and token requests | Full policy | A blocked provider logs `OIDC provider discovery blocked by outbound policy` naming the origin to allowlist (`crates/iam/src/oidc.rs`) |
 | Object Lambda targets | Full policy | `rustfs/src/admin/router.rs` `outbound_policy` |
-| Bucket replication targets | Literal check, relaxed | Private addresses are always allowed; loopback only with `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET=true` (`crates/ecstore/src/bucket/remote_s3_client.rs` `validate_remote_endpoint`, shared with on-demand migration sources) |
+| Bucket replication targets | Literal check, relaxed | Private addresses are always allowed; loopback only with `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET=true` (`crates/ecstore/src/bucket/bucket_target_sys.rs` `validate_replication_target_endpoint`) |
 | Site replication peers | Literal check | `rustfs/src/site_replication/mod.rs` |
 | Tiering warm backends (S3, MinIO, RustFS, Azure, GCS, Aliyun, Tencent, Huawei, R2) | Literal check | `crates/ecstore/src/services/tier/warm_backend.rs` `validate_endpoint`; the RustFS provider adds a debug-only, env-gated loopback exception for e2e tests |
 | Keystone `auth_url` | Literal check | `crates/keystone/src/config.rs` |

scripts/error-other-format-baseline.txt

@@ -5,6 +5,7 @@
 # scripts/check_error_other_format_ratchet.sh --update-baseline. A PR that
 # raises a count or adds a file is introducing a new quorum-bucketing hazard
 # and must carry an explicit exemption rationale in its description.
+2|crates/ecstore/src/bucket/bucket_target_sys.rs
 4|crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
 3|crates/ecstore/src/bucket/lifecycle/durable_namespace.rs
 2|crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs