feat(ecstore): add on-demand migration runtime OnDemandMigrationSys

Per-node runtime for On-Demand Migration (rustfs/backlog#2152): turns each
bucket's persisted config into a live SourceClient guarded by a three-state
circuit breaker, a TTL negative cache, per-key singleflight, a pull
concurrency semaphore and lock-free counters with a serializable snapshot.

- sys.rs: OnceLock singleton; `apply` installs/rebuilds/removes bucket state
  (config compared by value, counters preserved across rebuilds, old
  cancellation token fired); `publish` is the metadata publish-hook entry
  (sync removal, spawned install, generation-ordered so a slow older install
  cannot overwrite a newer one); `resolve(bucket, key)` judges module switch,
  bucket state, prefix filter, client availability, negative cache, breaker.
- breaker.rs: Closed/Open/HalfOpen with fixed constants (5 failures / 30 s
  window / 30 s open / 1 probe); NotFound resets, AccessDenied is neutral.
- negative_cache.rs: moka sync cache keyed by local key, ttl=0 disables.
- stats.rs: requests_total{op,outcome}, pulled_bytes/objects, pull_failures,
  inflight/queue gauges, log-bucket latency histogram, last_source_error;
  snake_case snapshot pinned by a golden JSON test.
- Anonymous sources surface as a typed `OdmStateError::AnonymousUnsupported`
  until the shared client builder gains an anonymous mode.
- rustfs: `RUSTFS_ON_DEMAND_MIGRATION_ENABLED` module switch (default false)
  published to module_switches and injected into ecstore before bucket
  metadata loads; hook registered at the same point.

crates/ecstore/Cargo.toml

@@ -166,7 +166,7 @@ uuid = { workspace = true, features = ["v4", "fast-rng", "serde", "macro-diagnos
 reed-solomon-erasure = { workspace = true, features = ["simd-accel"] }
 reed-solomon-simd = { workspace = true }
 lazy_static.workspace = true
-moka = { workspace = true, features = ["future"] }
+moka = { workspace = true, features = ["future", "sync"] }
 rustfs-lock.workspace = true
 rustfs-io-metrics.workspace = true
 regex = { workspace = true }
@@ -185,7 +185,7 @@ hyper-rustls = { workspace = true, default-features = false, features = ["native
 hostname.workspace = true
 rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] }
 rustls-pki-types.workspace = true
-tokio = { workspace = true, features = ["io-util", "sync", "signal", "fs", "rt-multi-thread"] }
+tokio = { workspace = true, features = ["io-util", "sync", "signal", "fs", "rt-multi-thread", "time"] }
 tonic = { workspace = true, features = ["gzip", "deflate"] }
 xxhash-rust = { workspace = true, features = ["xxh64", "xxh3"] }
 tower = { workspace = true, features = ["timeout"] }

crates/ecstore/src/api/mod.rs

@@ -128,7 +128,6 @@ pub mod bucket {
     }
 
     pub mod metadata {
-        pub use crate::bucket::metadata::BUCKET_DURABILITY_CONFIG;
         pub use crate::bucket::metadata::{
             BUCKET_ACCELERATE_CONFIG, BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_LOGGING_CONFIG,
             BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_QUOTA_CONFIG_FILE,
@@ -137,6 +136,7 @@ pub mod bucket {
             BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BUCKET_WEBSITE_CONFIG, BucketMetadata, OBJECT_LOCK_CONFIG,
             load_bucket_metadata, table_catalog_path_hash,
         };
+        pub use crate::bucket::metadata::{BUCKET_DURABILITY_CONFIG, BUCKET_ON_DEMAND_MIGRATION_CONFIG};
     }
 
     pub mod durability {
@@ -145,6 +145,29 @@ pub mod bucket {
         };
     }
 
+    pub mod on_demand_migration {
+        pub use crate::bucket::on_demand_migration::{
+            ApplyOutcome, BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION,
+            Breaker, BreakerState, BreakerTransition, BreakerVerdict, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, GaugeGuard,
+            LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup,
+            OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason,
+            PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS,
+            SourceLatencySnapshot, source_client_spec,
+        };
+        pub use crate::bucket::on_demand_migration::{
+            ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
+            OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy,
+            SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+        };
+        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 metadata_sys {
         #[cfg(feature = "test-util")]
         pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
@@ -154,11 +177,11 @@ pub mod bucket {
             delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy,
             get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
             get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
-            get_object_lock_config, get_object_lock_config_state, get_public_access_block_config, get_quota_config,
-            get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config,
-            get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata,
-            set_bucket_metadata, update, update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation,
-            update_quota_if_incarnation, update_under_transaction_lock,
+            get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config, get_public_access_block_config,
+            get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config,
+            get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata,
+            remove_bucket_metadata, set_bucket_metadata, update, update_bucket_targets_under_transaction_lock,
+            update_config_with, update_if_incarnation, update_quota_if_incarnation, update_under_transaction_lock,
         };
     }
 
@@ -199,6 +222,13 @@ 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,17 +15,13 @@
 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;
@@ -37,28 +33,17 @@ 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, 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 aws_sdk_s3::{Client as S3Client, operation::head_object::HeadObjectOutput};
+use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
 use futures::{StreamExt, stream};
-use http::{HeaderMap, HeaderName, HeaderValue, StatusCode, Uri};
-use hyper_util::client::legacy::Client as HyperClient;
-use hyper_util::rt::{TokioExecutor, TokioTimer};
+use http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
 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,
@@ -70,12 +55,10 @@ 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;
@@ -84,7 +67,6 @@ 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;
@@ -92,72 +74,50 @@ 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";
 
-#[derive(Clone)]
-struct RemoteTargetCredentialsProvider {
-    credentials: SdkCredentials,
+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(),
+    }
 }
 
-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)));
+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: "",
         }
-        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>>;
@@ -1058,57 +1018,17 @@ impl BucketTargetSys {
             });
         };
 
-        let creds = remote_target_sdk_credentials(credentials, &target.reset_id, SystemTime::now()).map_err(|error| {
-            BucketTargetError::RemoteTargetConnectionErr {
+        let spec = RemoteS3EndpointSpec::from(target);
+        let client = build_remote_s3_client(&spec)
+            .await
+            .map_err(|err| BucketTargetError::RemoteTargetConnectionErr {
                 bucket: target.target_bucket.clone(),
                 access_key: credentials.access_key.clone(),
-                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();
+                error: err.to_string(),
+            })?;
 
         Ok(TargetClient {
-            endpoint,
+            endpoint: spec.endpoint_url(),
             credentials: target.credentials.clone(),
             bucket: target.target_bucket.clone(),
             storage_class: target.storage_class.clone(),
@@ -1118,7 +1038,7 @@ impl BucketTargetSys {
             secure: target.secure,
             health_check_duration: target.health_check_duration,
             replicate_sync: target.replication_sync,
-            client: Arc::new(S3Client::from_conf(config)),
+            client: Arc::new(client),
         })
     }
 
@@ -1281,327 +1201,6 @@ 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() {
@@ -2707,7 +2306,24 @@ 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
@@ -2934,8 +2550,8 @@ mod tests {
             expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")),
         };
 
-        let sdk_credentials =
-            remote_target_sdk_credentials(&credentials, "account", now).expect("unexpired temporary credentials should build");
+        let sdk_credentials = remote_sdk_credentials(&remote_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));
@@ -2951,7 +2567,7 @@ mod tests {
             expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")),
         };
 
-        let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now())
+        let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::now())
             .expect("Go zero expiration should remain compatible with static credentials");
 
         assert!(sdk_credentials.session_token().is_none());
@@ -2969,14 +2585,14 @@ mod tests {
         };
 
         assert_eq!(
-            remote_target_sdk_credentials(&credentials, "", SystemTime::UNIX_EPOCH + Duration::from_secs(1_000))
+            remote_sdk_credentials(&remote_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_target_sdk_credentials(&credentials, "", expiration)
+            remote_sdk_credentials(&remote_credentials(&credentials, ""), expiration)
                 .expect_err("credentials expire at the exact expiration boundary"),
             EXPIRED_REMOTE_TARGET_CREDENTIALS
         );
@@ -3036,7 +2652,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_target_sdk_credentials(&credentials, "", SystemTime::now())
+        let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::now())
             .expect("unexpired temporary credentials should build");
         let client = S3Client::from_conf(
             S3Config::builder()
@@ -3211,6 +2827,46 @@ 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")
     }
@@ -3220,16 +2876,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_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());
+            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());
         }
     }
 
     #[test]
     fn replication_endpoint_rejects_loopback_without_opt_in() {
         // Default (production) behaviour: loopback IP and localhost host both rejected.
-        let err = validate_replication_target_endpoint_inner(&parse_url("http://127.0.0.1:9000"), false)
+        let err = validate_remote_endpoint_inner(&parse_url("http://127.0.0.1:9000"), false)
             .expect_err("loopback IP must be rejected by default");
         assert!(matches!(
             err,
@@ -3238,7 +2894,7 @@ mod tests {
                 ..
             }
         ));
-        let err = validate_replication_target_endpoint_inner(&parse_url("http://localhost:9000"), false)
+        let err = validate_remote_endpoint_inner(&parse_url("http://localhost:9000"), false)
             .expect_err("localhost must be rejected by default");
         assert!(matches!(
             err,
@@ -3253,15 +2909,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_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());
+        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());
     }
 
     #[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_replication_target_endpoint_inner(&parse_url("http://169.254.169.254/latest/meta-data"), true)
+        let err = validate_remote_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,
@@ -3270,7 +2926,7 @@ mod tests {
                 ..
             }
         ));
-        let err = validate_replication_target_endpoint_inner(&parse_url("http://[fe80::1]:9000"), true)
+        let err = validate_remote_endpoint_inner(&parse_url("http://[fe80::1]:9000"), true)
             .expect_err("link-local must stay rejected even with loopback opt-in");
         assert!(matches!(
             err,
@@ -4279,12 +3935,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_target(&BucketTarget {
+        let client = build_aws_s3_http_client_for_spec(&RemoteS3EndpointSpec::from(&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/metadata.rs

@@ -270,6 +270,7 @@ pub const BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG: &str = "public-access-block.xml";
 pub const BUCKET_ACL_CONFIG: &str = "bucket-acl.json";
 pub const BUCKET_TABLE_CONFIG: &str = "table-bucket.json";
 pub const BUCKET_DURABILITY_CONFIG: &str = "durability.json";
+pub const BUCKET_ON_DEMAND_MIGRATION_CONFIG: &str = "on-demand-migration.json";
 pub const BUCKET_TABLE_RESERVED_PREFIX: &str = ".rustfs-table";
 pub const BUCKET_TABLE_CATALOG_META_PREFIX: &str = "s3tables/catalog";
 pub const BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX: &str = "table-buckets";
@@ -321,6 +322,7 @@ pub struct BucketMetadata {
     pub bucket_acl_config_json: Vec<u8>,
     pub table_bucket_config_json: Vec<u8>,
     pub durability_config_json: Vec<u8>,
+    pub on_demand_migration_config_json: Vec<u8>,
 
     pub policy_config_updated_at: OffsetDateTime,
     pub object_lock_config_updated_at: OffsetDateTime,
@@ -342,6 +344,7 @@ pub struct BucketMetadata {
     pub bucket_acl_config_updated_at: OffsetDateTime,
     pub table_bucket_config_updated_at: OffsetDateTime,
     pub durability_config_updated_at: OffsetDateTime,
+    pub on_demand_migration_config_updated_at: OffsetDateTime,
 
     pub new_field_updated_at: OffsetDateTime,
 
@@ -393,6 +396,7 @@ impl Default for BucketMetadata {
             bucket_acl_config_json: Default::default(),
             table_bucket_config_json: Default::default(),
             durability_config_json: Default::default(),
+            on_demand_migration_config_json: Default::default(),
             policy_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             object_lock_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             encryption_config_updated_at: OffsetDateTime::UNIX_EPOCH,
@@ -413,6 +417,7 @@ impl Default for BucketMetadata {
             bucket_acl_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             table_bucket_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             durability_config_updated_at: OffsetDateTime::UNIX_EPOCH,
+            on_demand_migration_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             new_field_updated_at: OffsetDateTime::UNIX_EPOCH,
             policy_config: Default::default(),
             notification_config: Default::default(),
@@ -477,6 +482,23 @@ impl BucketMetadata {
     /// Absent/empty/unparsable payloads all mean "no override" (the bucket
     /// follows the global durability mode); a parse failure is logged so a
     /// corrupted entry cannot silently change fsync behavior.
+    /// Parsed on-demand migration config, if one is stored.
+    ///
+    /// `Ok(None)` means no config (absent or cleared). A stored payload that
+    /// does not parse is an error, never a default: the runtime must not
+    /// pull from a source it cannot describe.
+    pub fn on_demand_migration_config(
+        &self,
+    ) -> std::result::Result<
+        Option<super::on_demand_migration::OnDemandMigrationConfig>,
+        super::on_demand_migration::OnDemandMigrationConfigError,
+    > {
+        if self.on_demand_migration_config_json.is_empty() {
+            return Ok(None);
+        }
+        super::on_demand_migration::OnDemandMigrationConfig::from_json(&self.on_demand_migration_config_json).map(Some)
+    }
+
     pub fn durability_config(&self) -> Option<super::durability::BucketDurabilityConfig> {
         if self.durability_config_json.is_empty() {
             return None;
@@ -555,6 +577,9 @@ impl BucketMetadata {
                 "BucketAclConfigJSON" | "BucketAclConfigJson" => self.bucket_acl_config_json = read_msgp_bin(rd)?,
                 "TableBucketConfigJSON" | "TableBucketConfigJson" => self.table_bucket_config_json = read_msgp_bin(rd)?,
                 "DurabilityConfigJSON" | "DurabilityConfigJson" => self.durability_config_json = read_msgp_bin(rd)?,
+                "OnDemandMigrationConfigJSON" | "OnDemandMigrationConfigJson" => {
+                    self.on_demand_migration_config_json = read_msgp_bin(rd)?
+                }
                 "CorsConfigUpdatedAt" => self.cors_config_updated_at = read_msgp_time_value(rd)?,
                 "LoggingConfigUpdatedAt" => self.logging_config_updated_at = read_msgp_time_value(rd)?,
                 "WebsiteConfigUpdatedAt" => self.website_config_updated_at = read_msgp_time_value(rd)?,
@@ -564,6 +589,7 @@ impl BucketMetadata {
                 "BucketAclConfigUpdatedAt" => self.bucket_acl_config_updated_at = read_msgp_time_value(rd)?,
                 "TableBucketConfigUpdatedAt" => self.table_bucket_config_updated_at = read_msgp_time_value(rd)?,
                 "DurabilityConfigUpdatedAt" => self.durability_config_updated_at = read_msgp_time_value(rd)?,
+                "OnDemandMigrationConfigUpdatedAt" => self.on_demand_migration_config_updated_at = read_msgp_time_value(rd)?,
                 other => {
                     tracing::debug!(field = %other, "BucketMetadata decode_from: skipping unknown field");
                     skip_msgp_value(rd)?;
@@ -576,8 +602,8 @@ impl BucketMetadata {
 
     /// Encode to msgp bytes. Field order follows MinIO BucketMetadata for compatibility.
     pub fn encode_to<W: Write>(&self, wr: &mut W) -> Result<()> {
-        // Map size: MinIO fields (25) + RustFS extensions (19)
-        let map_len: u32 = 44;
+        // Map size: MinIO fields (25) + RustFS extensions (21)
+        let map_len: u32 = 46;
         rmp::encode::write_map_len(wr, map_len)?;
 
         // MinIO field order (same as Go struct)
@@ -637,6 +663,7 @@ impl BucketMetadata {
         write_bin_field(wr, "BucketAclConfigJSON", &self.bucket_acl_config_json)?;
         write_bin_field(wr, "TableBucketConfigJSON", &self.table_bucket_config_json)?;
         write_bin_field(wr, "DurabilityConfigJSON", &self.durability_config_json)?;
+        write_bin_field(wr, "OnDemandMigrationConfigJSON", &self.on_demand_migration_config_json)?;
         rmp::encode::write_str(wr, "CorsConfigUpdatedAt")?;
         write_msgp_time(wr, self.cors_config_updated_at)?;
         rmp::encode::write_str(wr, "LoggingConfigUpdatedAt")?;
@@ -655,6 +682,8 @@ impl BucketMetadata {
         write_msgp_time(wr, self.table_bucket_config_updated_at)?;
         rmp::encode::write_str(wr, "DurabilityConfigUpdatedAt")?;
         write_msgp_time(wr, self.durability_config_updated_at)?;
+        rmp::encode::write_str(wr, "OnDemandMigrationConfigUpdatedAt")?;
+        write_msgp_time(wr, self.on_demand_migration_config_updated_at)?;
 
         Ok(())
     }
@@ -756,6 +785,9 @@ impl BucketMetadata {
         if self.durability_config_updated_at == OffsetDateTime::UNIX_EPOCH {
             self.durability_config_updated_at = self.created
         }
+        if self.on_demand_migration_config_updated_at == OffsetDateTime::UNIX_EPOCH {
+            self.on_demand_migration_config_updated_at = self.created
+        }
     }
 
     pub fn update_config(&mut self, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
@@ -871,6 +903,17 @@ impl BucketMetadata {
                 self.durability_config_json = data;
                 self.durability_config_updated_at = updated;
             }
+            BUCKET_ON_DEMAND_MIGRATION_CONFIG => {
+                // Structural check only (shape, unknown fields); the
+                // deployment-relative rules run in the admin handler with a
+                // `ValidationContext`. A blob this build cannot read must not
+                // be persisted for every later reader to trip over.
+                if !data.is_empty() {
+                    super::on_demand_migration::OnDemandMigrationConfig::from_json(&data).map_err(Error::other)?;
+                }
+                self.on_demand_migration_config_json = data;
+                self.on_demand_migration_config_updated_at = updated;
+            }
             _ => return Err(Error::other(format!("config file not found : {config_file}"))),
         }
 
@@ -1779,6 +1822,117 @@ mod test {
         assert!(!bm.table_bucket_enabled());
     }
 
+    const ODM_JSON: &[u8] = br#"{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#;
+
+    /// rustfs/backlog#2148: the on-demand migration config is a RustFS
+    /// extension entry that round-trips through `update_config` and the
+    /// msgpack codec, clears on delete, and never parses corruption into a
+    /// default.
+    #[test]
+    fn on_demand_migration_config_round_trips_and_tracks_updates() {
+        use crate::bucket::on_demand_migration::{OnDemandMigrationConfig, OnDemandMigrationConfigError};
+
+        let mut bm = BucketMetadata::new("odm-bucket");
+        assert_eq!(bm.on_demand_migration_config(), Ok(None), "fresh metadata carries no config");
+
+        let expected = OnDemandMigrationConfig::from_json(ODM_JSON).unwrap();
+        bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .expect("valid config is accepted");
+        assert_ne!(bm.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(bm.on_demand_migration_config(), Ok(Some(expected.clone())));
+
+        let back = BucketMetadata::unmarshal(&bm.marshal_msg().unwrap()).unwrap();
+        assert_eq!(back.on_demand_migration_config_json, bm.on_demand_migration_config_json);
+        assert_eq!(
+            back.on_demand_migration_config_updated_at.unix_timestamp(),
+            bm.on_demand_migration_config_updated_at.unix_timestamp()
+        );
+        assert_eq!(back.on_demand_migration_config(), Ok(Some(expected)));
+
+        // A blob this build cannot read is rejected at the write boundary
+        // rather than persisted for every reader to trip over.
+        let before = bm.on_demand_migration_config_json.clone();
+        assert!(
+            bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec())
+                .is_err()
+        );
+        assert_eq!(bm.on_demand_migration_config_json, before, "a rejected update leaves the blob untouched");
+
+        // Delete clears the entry.
+        let stamped = bm.on_demand_migration_config_updated_at;
+        bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, Vec::new()).unwrap();
+        assert!(bm.on_demand_migration_config_json.is_empty());
+        assert_eq!(bm.on_demand_migration_config(), Ok(None));
+        assert!(bm.on_demand_migration_config_updated_at >= stamped);
+
+        // Corruption that bypassed `update_config` (disk, another writer)
+        // is a typed error, never a default.
+        bm.on_demand_migration_config_json = b"not-json".to_vec();
+        assert!(matches!(bm.on_demand_migration_config(), Err(OnDemandMigrationConfigError::Malformed(_))));
+    }
+
+    /// rustfs/backlog#2148: a `.metadata.bin` written before the on-demand
+    /// migration keys existed decodes with an empty blob and an epoch
+    /// timestamp that `default_timestamps` back-fills from `created`.
+    #[test]
+    fn on_demand_migration_config_absent_in_legacy_blob_defaults_to_created() {
+        let blob = decode_hex(include_str!("../../tests/fixtures/minio/bucket_metadata.blob.hex"));
+        let mut bm = BucketMetadata::unmarshal(&blob[4..]).expect("unmarshal MinIO bucket metadata");
+        assert!(bm.on_demand_migration_config_json.is_empty());
+        assert_eq!(bm.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(bm.on_demand_migration_config(), Ok(None));
+
+        bm.default_timestamps();
+        assert_ne!(bm.created, OffsetDateTime::UNIX_EPOCH, "fixture must carry a real creation time");
+        assert_eq!(bm.on_demand_migration_config_updated_at, bm.created);
+
+        // A metadata blob from this build with no config set stays
+        // indistinguishable from the legacy one for these fields.
+        let fresh = BucketMetadata::unmarshal(&BucketMetadata::new("fresh").marshal_msg().unwrap()).unwrap();
+        assert!(fresh.on_demand_migration_config_json.is_empty());
+        assert_eq!(fresh.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH);
+    }
+
+    /// rustfs/backlog#2148: a reader that predates the two on-demand
+    /// migration keys takes `decode_from`'s unknown-field branch, which is
+    /// `skip_msgp_value`. Walk the new-format blob with exactly that
+    /// primitive and prove both keys are skipped without desynchronising the
+    /// stream, so the fields that follow them still decode.
+    #[test]
+    fn old_decoder_skips_on_demand_migration_fields_without_desync() {
+        let mut bm = BucketMetadata::new("odm-skip");
+        bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .unwrap();
+        bm.update_config(BUCKET_DURABILITY_CONFIG, br#"{"mode":"relaxed"}"#.to_vec())
+            .unwrap();
+        let buf = bm.marshal_msg().unwrap();
+
+        let mut rd = std::io::Cursor::new(buf.as_slice());
+        let fields = rmp::decode::read_map_len(&mut rd).unwrap();
+        let mut skipped = Vec::new();
+        let mut durability_json = Vec::new();
+        for _ in 0..fields {
+            let key_len = rmp::decode::read_str_len(&mut rd).unwrap();
+            let mut key = vec![0u8; key_len as usize];
+            rd.read_exact(&mut key).unwrap();
+            let key = String::from_utf8(key).unwrap();
+            match key.as_str() {
+                // The field an old reader knows that is encoded *after* the
+                // unknown JSON key and *before* the unknown timestamp key.
+                "DurabilityConfigJSON" => durability_json = read_msgp_bin(&mut rd).unwrap(),
+                other => {
+                    if other.starts_with("OnDemandMigration") {
+                        skipped.push(other.to_string());
+                    }
+                    skip_msgp_value(&mut rd).unwrap();
+                }
+            }
+        }
+        assert_eq!(skipped, ["OnDemandMigrationConfigJSON", "OnDemandMigrationConfigUpdatedAt"]);
+        assert_eq!(durability_json, br#"{"mode":"relaxed"}"#);
+        assert_eq!(rd.position() as usize, buf.len(), "old-style walk must consume the blob exactly");
+    }
+
     /// HP-5b (rustfs/backlog#938): the durability override is a RustFS
     /// extension entry and must survive an encode/decode round trip.
     #[test]

crates/ecstore/src/bucket/metadata_sys.rs

@@ -19,6 +19,7 @@ use super::quota::BucketQuota;
 use super::target::BucketTargets;
 use crate::bucket::bucket_target_sys::BucketTargetSys;
 use crate::bucket::metadata::{load_bucket_metadata_parse, load_bucket_metadata_parse_with_presence};
+use crate::bucket::on_demand_migration::{ON_DEMAND_MIGRATION_CONFIG_HOOK, OnDemandMigrationConfig};
 use crate::bucket::utils::is_meta_bucketname;
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result, is_err_bucket_not_found, is_err_strict_volume_not_found};
@@ -384,6 +385,42 @@ fn clear_bucket_durability(bucket: &str) {
     crate::disk::local::bucket_durability::set(bucket, None);
 }
 
+/// Publish the bucket's on-demand migration config (or its absence) to the
+/// runtime registered in `ON_DEMAND_MIGRATION_CONFIG_HOOK`.
+///
+/// Called from the same five cache-install paths as
+/// [`sync_bucket_durability`]. A stored payload this build cannot parse is
+/// published as `None`: the runtime must stop pulling for that bucket rather
+/// than keep an older config or guess.
+fn sync_on_demand_migration(bucket: &str, bm: &BucketMetadata) {
+    let Some(hook) = ON_DEMAND_MIGRATION_CONFIG_HOOK.get() else {
+        return;
+    };
+    match bm.on_demand_migration_config() {
+        Ok(config) => hook(bucket, config.as_ref()),
+        Err(err) => {
+            warn!(
+                event = "bucket_metadata_parse_failed",
+                component = "ecstore",
+                subsystem = "bucket_metadata",
+                bucket = %bucket,
+                config = "on_demand_migration",
+                error = %err,
+                "Failed to parse bucket metadata config"
+            );
+            hook(bucket, None);
+        }
+    }
+}
+
+/// Withdraw a bucket's on-demand migration config when its metadata leaves
+/// the cache.
+fn clear_on_demand_migration(bucket: &str) {
+    if let Some(hook) = ON_DEMAND_MIGRATION_CONFIG_HOOK.get() {
+        hook(bucket, None);
+    }
+}
+
 pub async fn get(bucket: &str) -> Result<Arc<BucketMetadata>> {
     let sys = get_bucket_metadata_sys()?;
     let lock = sys.read().await;
@@ -970,6 +1007,16 @@ pub async fn get_durability_config(
     Ok((bm.durability_config(), bm.durability_config_updated_at))
 }
 
+/// The bucket's on-demand migration config with its update time, or
+/// `Ok(None)` when the bucket has none. A stored payload that does not parse
+/// is a typed error (`OnDemandMigrationConfigError` inside `Error::Io`).
+pub async fn get_on_demand_migration_config(bucket: &str) -> Result<Option<(OnDemandMigrationConfig, OffsetDateTime)>> {
+    let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
+    let bucket_meta_sys = bucket_meta_sys_lock.read().await;
+
+    bucket_meta_sys.get_on_demand_migration_config(bucket).await
+}
+
 pub async fn get_quota_config(bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1492,6 +1539,7 @@ impl BucketMetadataSys {
                 if removed {
                     BucketTargetSys::get().delete(bucket).await;
                     clear_bucket_durability(bucket);
+                    clear_on_demand_migration(bucket);
                 }
             }
             return Ok(());
@@ -1529,6 +1577,7 @@ impl BucketMetadataSys {
                 self.missing_buckets.invalidate(bucket).await;
                 sync_bucket_target_sys(bucket, &bm).await;
                 sync_bucket_durability(bucket, &bm);
+                sync_on_demand_migration(bucket, &bm);
             }
             MetadataLoadMode::Initial => {
                 let _publish_guard = self
@@ -1575,6 +1624,7 @@ impl BucketMetadataSys {
             if removed {
                 BucketTargetSys::get().delete(bucket).await;
                 clear_bucket_durability(bucket);
+                clear_on_demand_migration(bucket);
             }
             return Ok(());
         }
@@ -1597,6 +1647,7 @@ impl BucketMetadataSys {
         self.missing_buckets.invalidate(bucket).await;
         sync_bucket_target_sys(bucket, &metadata).await;
         sync_bucket_durability(bucket, &metadata);
+        sync_on_demand_migration(bucket, &metadata);
         Ok(())
     }
 
@@ -1624,6 +1675,7 @@ impl BucketMetadataSys {
             self.missing_buckets.invalidate(&bucket).await;
             sync_bucket_target_sys(&bucket, &bm).await;
             sync_bucket_durability(&bucket, &bm);
+            sync_on_demand_migration(&bucket, &bm);
         }
     }
 
@@ -1644,6 +1696,7 @@ impl BucketMetadataSys {
         if removed {
             BucketTargetSys::get().delete(bucket).await;
             clear_bucket_durability(bucket);
+            clear_on_demand_migration(bucket);
         }
         removed || removed_fabricated
     }
@@ -1933,6 +1986,7 @@ impl BucketMetadataSys {
                 self.missing_buckets.invalidate(bucket).await;
                 sync_bucket_target_sys(bucket, &bm).await;
                 sync_bucket_durability(bucket, &bm);
+                sync_on_demand_migration(bucket, &bm);
             } else {
                 let exists = self
                     .bucket_exists(bucket, &guard, "lazy bucket metadata existence check")
@@ -2271,6 +2325,7 @@ impl BucketMetadataSys {
         self.missing_buckets.invalidate(bucket).await;
         sync_bucket_target_sys(bucket, &metadata).await;
         sync_bucket_durability(bucket, &metadata);
+        sync_on_demand_migration(bucket, &metadata);
         Ok(BucketMetadataAuthority::Authoritative(metadata))
     }
 
@@ -2463,6 +2518,17 @@ impl BucketMetadataSys {
             Err(Error::ConfigNotFound)
         }
     }
+
+    /// See [`get_on_demand_migration_config`].
+    pub async fn get_on_demand_migration_config(
+        &self,
+        bucket: &str,
+    ) -> Result<Option<(OnDemandMigrationConfig, OffsetDateTime)>> {
+        let (bm, _) = self.get_config(bucket).await?;
+
+        let config = bm.on_demand_migration_config().map_err(Error::other)?;
+        Ok(config.map(|config| (config, bm.on_demand_migration_config_updated_at)))
+    }
 }
 
 /// Test-only fixture shared with sibling modules (e.g. the quota checker
@@ -4043,6 +4109,151 @@ mod tests {
         assert_eq!(bucket_durability::lookup(bucket), None);
     }
 
+    const ODM_JSON: &[u8] = br#"{"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#;
+
+    /// Every `(bucket, config)` the recording hook has seen. Tests filter by
+    /// their own bucket name; the hook is process-wide and set once.
+    static ODM_HOOK_CALLS: std::sync::Mutex<Vec<(String, Option<OnDemandMigrationConfig>)>> = std::sync::Mutex::new(Vec::new());
+
+    fn install_recording_odm_hook() {
+        ON_DEMAND_MIGRATION_CONFIG_HOOK.get_or_init(|| {
+            Box::new(|bucket, config| {
+                ODM_HOOK_CALLS.lock().unwrap().push((bucket.to_string(), config.cloned()));
+            })
+        });
+    }
+
+    fn odm_hook_calls(bucket: &str) -> Vec<Option<OnDemandMigrationConfig>> {
+        ODM_HOOK_CALLS
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|(name, _)| name == bucket)
+            .map(|(_, config)| config.clone())
+            .collect()
+    }
+
+    /// rustfs/backlog#2148: the accessor reports absence as `Ok(None)` and a
+    /// stored payload it cannot parse as a typed error, never as a default
+    /// and never as `ConfigNotFound`.
+    #[tokio::test]
+    async fn get_on_demand_migration_config_distinguishes_absent_from_corrupt() {
+        use crate::bucket::on_demand_migration::OnDemandMigrationConfigError;
+
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = BucketMetadataSys::new(ecstore);
+        let bucket = "odm-accessor";
+
+        sys.set(bucket.to_string(), Arc::new(BucketMetadata::new(bucket))).await;
+        assert_eq!(sys.get_on_demand_migration_config(bucket).await.unwrap(), None);
+
+        let mut corrupt = BucketMetadata::new(bucket);
+        corrupt.on_demand_migration_config_json = br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec();
+        sys.set(bucket.to_string(), Arc::new(corrupt)).await;
+        let err = sys
+            .get_on_demand_migration_config(bucket)
+            .await
+            .expect_err("corrupt config must not read as a default");
+        assert_ne!(err, Error::ConfigNotFound, "corruption must not be reported as absence");
+        let typed = match &err {
+            Error::Io(io) => io
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<OnDemandMigrationConfigError>()),
+            _ => None,
+        };
+        assert!(
+            matches!(typed, Some(OnDemandMigrationConfigError::Malformed(_))),
+            "typed parse error must survive the Result boundary, got: {err:?}"
+        );
+
+        let mut valid = BucketMetadata::new(bucket);
+        valid
+            .update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .unwrap();
+        let stamped = valid.on_demand_migration_config_updated_at;
+        sys.set(bucket.to_string(), Arc::new(valid)).await;
+        let (config, updated_at) = sys
+            .get_on_demand_migration_config(bucket)
+            .await
+            .unwrap()
+            .expect("stored config is returned");
+        assert_eq!(config, OnDemandMigrationConfig::from_json(ODM_JSON).unwrap());
+        assert_eq!(updated_at, stamped);
+    }
+
+    /// rustfs/backlog#2148: the publish hook fires on every path that
+    /// installs bucket metadata into the cache (set, initial load, peer
+    /// reload, refresh loop, lazy load) and withdraws on removal, mirroring
+    /// `sync_bucket_durability`.
+    #[tokio::test]
+    async fn on_demand_migration_hook_fires_on_every_cache_install_path() {
+        install_recording_odm_hook();
+
+        let (dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let bucket = "odm-hook-paths";
+        for dir in &dirs {
+            std::fs::create_dir_all(dir.path().join(bucket)).expect("physical bucket should exist");
+        }
+        let expected = OnDemandMigrationConfig::from_json(ODM_JSON).unwrap();
+        let expect_publish = |before: usize, label: &str| {
+            let calls = odm_hook_calls(bucket);
+            assert_eq!(calls.len(), before + 1, "{label} must publish exactly once");
+            assert_eq!(calls.last().unwrap().as_ref(), Some(&expected), "{label} must publish the stored config");
+        };
+
+        // set (via persist_new_and_set, which installs through `set`).
+        let mut bm = BucketMetadata::new(bucket);
+        bm.update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .unwrap();
+        let writer = BucketMetadataSys::new(ecstore.clone());
+        let before = odm_hook_calls(bucket).len();
+        writer.persist_new_and_set(bm).await.expect("metadata should persist");
+        expect_publish(before, "set");
+
+        // init (initial load on a cold system).
+        let mut cold = BucketMetadataSys::new(ecstore.clone());
+        let before = odm_hook_calls(bucket).len();
+        cold.init(vec![bucket.to_string()]).await;
+        assert!(cold.get(bucket).await.is_ok(), "initial load must cache the bucket");
+        expect_publish(before, "init");
+
+        // peer reload.
+        let before = odm_hook_calls(bucket).len();
+        cold.reload_from_store(bucket).await.expect("peer reload should publish");
+        expect_publish(before, "peer reload");
+
+        // refresh loop.
+        let before = odm_hook_calls(bucket).len();
+        let mut failed = HashSet::new();
+        cold.concurrent_load(&[bucket.to_string()], &mut failed, MetadataLoadMode::Refresh)
+            .await;
+        assert!(failed.is_empty(), "refresh must succeed");
+        expect_publish(before, "refresh loop");
+
+        // lazy load on another cold system.
+        let lazy = BucketMetadataSys::new(ecstore);
+        let before = odm_hook_calls(bucket).len();
+        let (_, loaded) = lazy.get_config(bucket).await.expect("lazy load should publish");
+        assert!(loaded, "the lazy path must have gone to disk");
+        expect_publish(before, "lazy load");
+
+        // Removal withdraws the config.
+        let before = odm_hook_calls(bucket).len();
+        assert!(lazy.remove(bucket).await);
+        let calls = odm_hook_calls(bucket);
+        assert_eq!(calls.len(), before + 1, "remove must withdraw exactly once");
+        assert_eq!(calls.last().unwrap(), &None);
+
+        // A corrupt payload is withdrawn, never published as a config.
+        let mut corrupt = BucketMetadata::new(bucket);
+        corrupt.on_demand_migration_config_json = b"not-json".to_vec();
+        let before = odm_hook_calls(bucket).len();
+        lazy.set(bucket.to_string(), Arc::new(corrupt)).await;
+        let calls = odm_hook_calls(bucket);
+        assert_eq!(calls.len(), before + 1);
+        assert_eq!(calls.last().unwrap(), &None, "unreadable config must publish absence");
+    }
+
     #[tokio::test]
     async fn refresh_wait_exits_when_cancelled() {
         let cancel_token = CancellationToken::new();

crates/ecstore/src/bucket/mod.rs

@@ -26,8 +26,10 @@ 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/breaker.rs

@@ -0,0 +1,362 @@
+// 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.
+
+//! Per-bucket three-state circuit breaker protecting an on-demand migration
+//! source (rustfs/backlog#2152).
+//!
+//! `Closed` lets every request through and counts consecutive failures
+//! inside a sliding window; reaching the threshold opens the breaker. `Open`
+//! rejects everything until the open duration elapses, then moves to
+//! `HalfOpen`, which admits a single probe: success closes the breaker,
+//! failure re-opens it. Timing uses `tokio::time::Instant` so tests can drive
+//! it with `tokio::time::pause`.
+//!
+//! Only transport-level failures count (`Throttled`, `Timeout`, `Connect`,
+//! `ServerError`). `NotFound` is a healthy answer and resets the failure
+//! streak; `AccessDenied`, `Unsupported` and `Other` are configuration or
+//! object problems that neither open nor close the breaker.
+
+use super::source_client::SourceError;
+use parking_lot::Mutex;
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+use tokio::time::Instant;
+
+/// Consecutive counted failures that open the breaker.
+pub const BREAKER_FAILURE_THRESHOLD: u32 = 5;
+/// Failures further apart than this do not accumulate.
+pub const BREAKER_FAILURE_WINDOW: Duration = Duration::from_secs(30);
+/// How long an open breaker rejects before admitting a probe.
+pub const BREAKER_OPEN_DURATION: Duration = Duration::from_secs(30);
+/// Probes admitted while half-open.
+pub const BREAKER_HALF_OPEN_MAX_PROBES: u32 = 1;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum BreakerState {
+    Closed,
+    Open,
+    HalfOpen,
+}
+
+impl BreakerState {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            BreakerState::Closed => "closed",
+            BreakerState::Open => "open",
+            BreakerState::HalfOpen => "half_open",
+        }
+    }
+}
+
+/// A state change the caller may want to log.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct BreakerTransition {
+    pub from: BreakerState,
+    pub to: BreakerState,
+}
+
+/// How a source result is scored by the breaker.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum BreakerVerdict {
+    /// Resets the failure streak; closes a half-open breaker.
+    Success,
+    /// Counts toward the threshold; re-opens a half-open breaker.
+    Failure,
+    /// Leaves the breaker untouched.
+    Neutral,
+}
+
+impl BreakerVerdict {
+    /// `None` is a successful source call.
+    pub fn for_result(error: Option<&SourceError>) -> Self {
+        match error {
+            None | Some(SourceError::NotFound) => BreakerVerdict::Success,
+            Some(SourceError::Throttled | SourceError::Timeout | SourceError::Connect(_) | SourceError::ServerError(_)) => {
+                BreakerVerdict::Failure
+            }
+            Some(SourceError::AccessDenied | SourceError::Unsupported(_) | SourceError::Other(_)) => BreakerVerdict::Neutral,
+        }
+    }
+}
+
+#[derive(Debug)]
+struct Inner {
+    state: BreakerState,
+    consecutive_failures: u32,
+    last_failure_at: Option<Instant>,
+    opened_at: Option<Instant>,
+    half_open_probes: u32,
+}
+
+#[derive(Debug)]
+pub struct Breaker {
+    inner: Mutex<Inner>,
+}
+
+impl Default for Breaker {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Breaker {
+    pub fn new() -> Self {
+        Self {
+            inner: Mutex::new(Inner {
+                state: BreakerState::Closed,
+                consecutive_failures: 0,
+                last_failure_at: None,
+                opened_at: None,
+                half_open_probes: 0,
+            }),
+        }
+    }
+
+    /// Current state after applying the open-duration timeout.
+    pub fn state(&self) -> BreakerState {
+        let mut inner = self.inner.lock();
+        Self::advance(&mut inner, Instant::now());
+        inner.state
+    }
+
+    /// Whether a request may reach the source right now. Consumes the
+    /// half-open probe budget when it grants one.
+    pub fn allow_request(&self) -> bool {
+        let mut inner = self.inner.lock();
+        Self::advance(&mut inner, Instant::now());
+        match inner.state {
+            BreakerState::Closed => true,
+            BreakerState::Open => false,
+            BreakerState::HalfOpen => {
+                if inner.half_open_probes < BREAKER_HALF_OPEN_MAX_PROBES {
+                    inner.half_open_probes += 1;
+                    true
+                } else {
+                    false
+                }
+            }
+        }
+    }
+
+    /// Scores a source result; returns the transition it caused, if any.
+    pub fn record(&self, verdict: BreakerVerdict) -> Option<BreakerTransition> {
+        match verdict {
+            BreakerVerdict::Success => self.record_success(),
+            BreakerVerdict::Failure => self.record_failure(),
+            BreakerVerdict::Neutral => None,
+        }
+    }
+
+    pub fn record_success(&self) -> Option<BreakerTransition> {
+        let mut inner = self.inner.lock();
+        let now = Instant::now();
+        Self::advance(&mut inner, now);
+        inner.consecutive_failures = 0;
+        inner.last_failure_at = None;
+        match inner.state {
+            BreakerState::Closed => None,
+            // A success while open can only come from a request admitted
+            // before the breaker opened; it says nothing about recovery.
+            BreakerState::Open => None,
+            BreakerState::HalfOpen => Some(Self::transition(&mut inner, BreakerState::Closed, now)),
+        }
+    }
+
+    pub fn record_failure(&self) -> Option<BreakerTransition> {
+        let mut inner = self.inner.lock();
+        let now = Instant::now();
+        Self::advance(&mut inner, now);
+        match inner.state {
+            BreakerState::Closed => {
+                let within_window = inner
+                    .last_failure_at
+                    .is_some_and(|last| now.saturating_duration_since(last) <= BREAKER_FAILURE_WINDOW);
+                inner.consecutive_failures = if within_window { inner.consecutive_failures + 1 } else { 1 };
+                inner.last_failure_at = Some(now);
+                if inner.consecutive_failures >= BREAKER_FAILURE_THRESHOLD {
+                    Some(Self::transition(&mut inner, BreakerState::Open, now))
+                } else {
+                    None
+                }
+            }
+            BreakerState::Open => None,
+            BreakerState::HalfOpen => Some(Self::transition(&mut inner, BreakerState::Open, now)),
+        }
+    }
+
+    fn advance(inner: &mut Inner, now: Instant) {
+        if inner.state == BreakerState::Open
+            && inner
+                .opened_at
+                .is_some_and(|opened| now.saturating_duration_since(opened) >= BREAKER_OPEN_DURATION)
+        {
+            Self::transition(inner, BreakerState::HalfOpen, now);
+        }
+    }
+
+    fn transition(inner: &mut Inner, to: BreakerState, now: Instant) -> BreakerTransition {
+        let from = inner.state;
+        inner.state = to;
+        match to {
+            BreakerState::Open => {
+                inner.opened_at = Some(now);
+                inner.half_open_probes = 0;
+            }
+            BreakerState::HalfOpen => {
+                inner.half_open_probes = 0;
+            }
+            BreakerState::Closed => {
+                inner.opened_at = None;
+                inner.half_open_probes = 0;
+                inner.consecutive_failures = 0;
+                inner.last_failure_at = None;
+            }
+        }
+        BreakerTransition { from, to }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn server_error() -> SourceError {
+        SourceError::ServerError(503)
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn five_failures_open_then_half_open_after_timeout() {
+        let breaker = Breaker::new();
+        for i in 0..BREAKER_FAILURE_THRESHOLD - 1 {
+            assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&server_error()))), None, "failure {i}");
+            assert_eq!(breaker.state(), BreakerState::Closed);
+        }
+        assert_eq!(
+            breaker.record(BreakerVerdict::for_result(Some(&server_error()))),
+            Some(BreakerTransition {
+                from: BreakerState::Closed,
+                to: BreakerState::Open
+            })
+        );
+        assert_eq!(breaker.state(), BreakerState::Open);
+        assert!(!breaker.allow_request());
+
+        tokio::time::advance(BREAKER_OPEN_DURATION - Duration::from_secs(1)).await;
+        assert!(!breaker.allow_request());
+        assert_eq!(breaker.state(), BreakerState::Open);
+
+        tokio::time::advance(Duration::from_secs(1)).await;
+        assert_eq!(breaker.state(), BreakerState::HalfOpen);
+        assert!(breaker.allow_request(), "one probe is admitted");
+        assert!(!breaker.allow_request(), "second probe is rejected");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn half_open_probe_success_closes_and_failure_reopens() {
+        let breaker = Breaker::new();
+        for _ in 0..BREAKER_FAILURE_THRESHOLD {
+            breaker.record_failure();
+        }
+        tokio::time::advance(BREAKER_OPEN_DURATION).await;
+        assert!(breaker.allow_request());
+        assert_eq!(
+            breaker.record_failure(),
+            Some(BreakerTransition {
+                from: BreakerState::HalfOpen,
+                to: BreakerState::Open
+            })
+        );
+        assert!(!breaker.allow_request());
+
+        tokio::time::advance(BREAKER_OPEN_DURATION).await;
+        assert!(breaker.allow_request());
+        assert_eq!(
+            breaker.record_success(),
+            Some(BreakerTransition {
+                from: BreakerState::HalfOpen,
+                to: BreakerState::Closed
+            })
+        );
+        assert_eq!(breaker.state(), BreakerState::Closed);
+        assert!(breaker.allow_request());
+        // The streak restarts from zero after closing.
+        for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 {
+            assert_eq!(breaker.record_failure(), None);
+        }
+        assert_eq!(breaker.state(), BreakerState::Closed);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn failures_outside_window_do_not_accumulate() {
+        let breaker = Breaker::new();
+        for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 {
+            breaker.record_failure();
+        }
+        tokio::time::advance(BREAKER_FAILURE_WINDOW + Duration::from_secs(1)).await;
+        assert_eq!(breaker.record_failure(), None, "stale streak restarts at one");
+        assert_eq!(breaker.state(), BreakerState::Closed);
+    }
+
+    #[test]
+    fn not_found_and_access_denied_do_not_count() {
+        let breaker = Breaker::new();
+        for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 {
+            breaker.record(BreakerVerdict::for_result(Some(&server_error())));
+        }
+        assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::AccessDenied))), None);
+        assert_eq!(breaker.state(), BreakerState::Closed);
+        // AccessDenied is neutral: the streak is still one short of opening.
+        assert_eq!(
+            breaker.record(BreakerVerdict::for_result(Some(&SourceError::Unsupported("sse-c".into())))),
+            None
+        );
+        assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::Other("x".into())))), None);
+        // NotFound is a healthy answer and resets the streak entirely.
+        assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::NotFound))), None);
+        for _ in 0..BREAKER_FAILURE_THRESHOLD - 1 {
+            assert_eq!(breaker.record(BreakerVerdict::for_result(Some(&SourceError::Timeout))), None);
+        }
+        assert_eq!(breaker.state(), BreakerState::Closed);
+    }
+
+    #[test]
+    fn verdicts_cover_every_source_error_class() {
+        assert_eq!(BreakerVerdict::for_result(None), BreakerVerdict::Success);
+        assert_eq!(BreakerVerdict::for_result(Some(&SourceError::NotFound)), BreakerVerdict::Success);
+        for failure in [
+            SourceError::Throttled,
+            SourceError::Timeout,
+            SourceError::Connect("refused".into()),
+            SourceError::ServerError(500),
+        ] {
+            assert_eq!(BreakerVerdict::for_result(Some(&failure)), BreakerVerdict::Failure, "{failure:?}");
+        }
+        for neutral in [
+            SourceError::AccessDenied,
+            SourceError::Unsupported("sse-c".into()),
+            SourceError::Other("x".into()),
+        ] {
+            assert_eq!(BreakerVerdict::for_result(Some(&neutral)), BreakerVerdict::Neutral, "{neutral:?}");
+        }
+    }
+
+    #[test]
+    fn state_labels_are_stable() {
+        assert_eq!(BreakerState::Closed.as_str(), "closed");
+        assert_eq!(BreakerState::Open.as_str(), "open");
+        assert_eq!(BreakerState::HalfOpen.as_str(), "half_open");
+        assert_eq!(serde_json::to_string(&BreakerState::HalfOpen).unwrap(), "\"half_open\"");
+    }
+}

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

@@ -0,0 +1,47 @@
+// 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): a bucket can name an external S3-compatible
+//! source bucket; GET misses are served from that source and backfilled
+//! locally. This module owns the bucket-level configuration model
+//! (`on-demand-migration.json` in the bucket metadata file), the source
+//! client, and the per-node runtime (`sys`) that turns configs into live
+//! clients guarded by a breaker, a negative cache, singleflight and a pull
+//! concurrency limit (rustfs/backlog#2147).
+
+pub mod breaker;
+pub mod config;
+pub mod negative_cache;
+pub mod source_client;
+pub mod stats;
+pub mod sys;
+
+pub use breaker::{
+    BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, Breaker,
+    BreakerState, BreakerTransition, BreakerVerdict,
+};
+pub use config::{
+    ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
+    OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig,
+    SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+};
+pub use negative_cache::{NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache};
+pub use stats::{
+    GaugeGuard, LastSourceError, LatencyBucketSnapshot, OdmOp, OdmOutcome, OdmStats, OdmStatsSnapshot, PullFailureReason,
+    PullPath, SOURCE_LATENCY_BUCKET_BOUNDS_MS, SourceLatencySnapshot,
+};
+pub use sys::{
+    ApplyOutcome, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, OdmBucketSnapshot, OdmLookup, OdmStateError,
+    OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_client_spec,
+};

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

@@ -0,0 +1,130 @@
+// 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.
+
+//! Per-bucket cache of keys the source answered 404 for
+//! (rustfs/backlog#2152). A hit short-circuits the source lookup for
+//! `policy.negative_cache_ttl_secs`; a TTL of zero disables the cache.
+//!
+//! Entries are never invalidated on a local PUT: once the object exists
+//! locally the handler never consults ODM for it, so a stale negative entry
+//! is harmless.
+
+use std::time::Duration;
+
+/// Upper bound on remembered keys per bucket; LRU eviction beyond it.
+pub const NEGATIVE_CACHE_MAX_ENTRIES: u64 = 100_000;
+
+#[derive(Debug)]
+pub struct NegativeCache {
+    cache: Option<moka::sync::Cache<String, ()>>,
+    ttl: Duration,
+}
+
+impl NegativeCache {
+    /// `ttl == 0` builds a disabled cache that never records anything.
+    pub fn new(ttl: Duration) -> Self {
+        Self::with_capacity(ttl, NEGATIVE_CACHE_MAX_ENTRIES)
+    }
+
+    pub fn with_capacity(ttl: Duration, max_entries: u64) -> Self {
+        let cache = (!ttl.is_zero()).then(|| {
+            moka::sync::Cache::builder()
+                .max_capacity(max_entries)
+                .time_to_live(ttl)
+                .build()
+        });
+        Self { cache, ttl }
+    }
+
+    pub fn is_enabled(&self) -> bool {
+        self.cache.is_some()
+    }
+
+    pub fn ttl(&self) -> Duration {
+        self.ttl
+    }
+
+    /// Whether `key` is currently remembered as absent on the source.
+    pub fn contains(&self, key: &str) -> bool {
+        self.cache.as_ref().is_some_and(|cache| cache.get(key).is_some())
+    }
+
+    /// Remembers `key` as absent; no-op when disabled.
+    pub fn insert(&self, key: &str) {
+        if let Some(cache) = &self.cache {
+            cache.insert(key.to_string(), ());
+        }
+    }
+
+    /// Forgets `key` (e.g. after an admin-triggered backfill found it).
+    pub fn remove(&self, key: &str) {
+        if let Some(cache) = &self.cache {
+            cache.invalidate(key);
+        }
+    }
+
+    /// Approximate live entry count, for status snapshots only.
+    pub fn len(&self) -> u64 {
+        self.cache.as_ref().map_or(0, |cache| {
+            cache.run_pending_tasks();
+            cache.entry_count()
+        })
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn entry_expires_after_ttl() {
+        let cache = NegativeCache::new(Duration::from_millis(80));
+        assert!(cache.is_enabled());
+        cache.insert("a/x");
+        assert!(cache.contains("a/x"));
+        assert!(!cache.contains("a/y"));
+        std::thread::sleep(Duration::from_millis(160));
+        assert!(!cache.contains("a/x"), "entry must expire after the TTL");
+    }
+
+    #[test]
+    fn zero_ttl_disables_the_cache() {
+        let cache = NegativeCache::new(Duration::ZERO);
+        assert!(!cache.is_enabled());
+        cache.insert("a/x");
+        assert!(!cache.contains("a/x"));
+        assert!(cache.is_empty());
+    }
+
+    #[test]
+    fn remove_forgets_a_key() {
+        let cache = NegativeCache::new(Duration::from_secs(30));
+        cache.insert("a/x");
+        cache.remove("a/x");
+        assert!(!cache.contains("a/x"));
+    }
+
+    #[test]
+    fn capacity_bounds_entries() {
+        let cache = NegativeCache::with_capacity(Duration::from_secs(30), 4);
+        for i in 0..64 {
+            cache.insert(&format!("k{i}"));
+        }
+        assert!(cache.len() <= 4, "len {} exceeds capacity", cache.len());
+    }
+}

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

@@ -0,0 +1,527 @@
+// 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.
+
+//! Per-bucket on-demand migration counters (rustfs/backlog#2152).
+//!
+//! `OdmStats` is lock-free and survives config rebuilds; `snapshot()` turns
+//! it into the serializable `OdmStatsSnapshot` that the metrics collector
+//! and the admin status route (ODM-10/14/15) consume. Field names and label
+//! values are a wire contract: the golden JSON test below pins them.
+
+use super::breaker::BreakerState;
+use super::source_client::SourceError;
+use parking_lot::Mutex;
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use time::OffsetDateTime;
+
+/// Request operations that can enter ODM.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum OdmOp {
+    Get,
+    Head,
+}
+
+impl OdmOp {
+    pub const ALL: [OdmOp; 2] = [OdmOp::Get, OdmOp::Head];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            OdmOp::Get => "get",
+            OdmOp::Head => "head",
+        }
+    }
+}
+
+/// How a request that entered ODM ended. `local_hit` is deliberately absent:
+/// requests served locally never reach the runtime.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum OdmOutcome {
+    SourceHit,
+    SourceMiss,
+    SourceError,
+    BreakerOpen,
+    NegativeCached,
+    Filtered,
+    Unsupported,
+}
+
+impl OdmOutcome {
+    pub const ALL: [OdmOutcome; 7] = [
+        OdmOutcome::SourceHit,
+        OdmOutcome::SourceMiss,
+        OdmOutcome::SourceError,
+        OdmOutcome::BreakerOpen,
+        OdmOutcome::NegativeCached,
+        OdmOutcome::Filtered,
+        OdmOutcome::Unsupported,
+    ];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            OdmOutcome::SourceHit => "source_hit",
+            OdmOutcome::SourceMiss => "source_miss",
+            OdmOutcome::SourceError => "source_error",
+            OdmOutcome::BreakerOpen => "breaker_open",
+            OdmOutcome::NegativeCached => "negative_cached",
+            OdmOutcome::Filtered => "filtered",
+            OdmOutcome::Unsupported => "unsupported",
+        }
+    }
+}
+
+/// Which pipeline stored a pulled object locally.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PullPath {
+    /// Streamed to the client and written locally in one pass.
+    Inline,
+    /// Pulled by a background task after a partial/large read.
+    Background,
+    /// Pulled by the backfill job.
+    Backfill,
+}
+
+impl PullPath {
+    pub const ALL: [PullPath; 3] = [PullPath::Inline, PullPath::Background, PullPath::Backfill];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            PullPath::Inline => "inline",
+            PullPath::Background => "background",
+            PullPath::Backfill => "backfill",
+        }
+    }
+}
+
+/// Why a pull did not produce a local object.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum PullFailureReason {
+    SourceNotFound,
+    SourceAccessDenied,
+    SourceThrottled,
+    SourceTimeout,
+    SourceConnect,
+    SourceServerError,
+    SourceUnsupported,
+    SourceOther,
+    /// Source bytes did not match the ETag advertised by HEAD/GET.
+    EtagMismatch,
+    /// The local write (internal PUT) failed.
+    LocalWrite,
+    /// The bucket state was removed or the process is shutting down.
+    Canceled,
+    /// The background pull queue was full.
+    QueueFull,
+}
+
+impl PullFailureReason {
+    pub const ALL: [PullFailureReason; 12] = [
+        PullFailureReason::SourceNotFound,
+        PullFailureReason::SourceAccessDenied,
+        PullFailureReason::SourceThrottled,
+        PullFailureReason::SourceTimeout,
+        PullFailureReason::SourceConnect,
+        PullFailureReason::SourceServerError,
+        PullFailureReason::SourceUnsupported,
+        PullFailureReason::SourceOther,
+        PullFailureReason::EtagMismatch,
+        PullFailureReason::LocalWrite,
+        PullFailureReason::Canceled,
+        PullFailureReason::QueueFull,
+    ];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            PullFailureReason::SourceNotFound => "source_not_found",
+            PullFailureReason::SourceAccessDenied => "source_access_denied",
+            PullFailureReason::SourceThrottled => "source_throttled",
+            PullFailureReason::SourceTimeout => "source_timeout",
+            PullFailureReason::SourceConnect => "source_connect",
+            PullFailureReason::SourceServerError => "source_server_error",
+            PullFailureReason::SourceUnsupported => "source_unsupported",
+            PullFailureReason::SourceOther => "source_other",
+            PullFailureReason::EtagMismatch => "etag_mismatch",
+            PullFailureReason::LocalWrite => "local_write",
+            PullFailureReason::Canceled => "canceled",
+            PullFailureReason::QueueFull => "queue_full",
+        }
+    }
+}
+
+impl From<&SourceError> for PullFailureReason {
+    fn from(err: &SourceError) -> Self {
+        match err {
+            SourceError::NotFound => PullFailureReason::SourceNotFound,
+            SourceError::AccessDenied => PullFailureReason::SourceAccessDenied,
+            SourceError::Throttled => PullFailureReason::SourceThrottled,
+            SourceError::Timeout => PullFailureReason::SourceTimeout,
+            SourceError::Connect(_) => PullFailureReason::SourceConnect,
+            SourceError::ServerError(_) => PullFailureReason::SourceServerError,
+            SourceError::Unsupported(_) => PullFailureReason::SourceUnsupported,
+            SourceError::Other(_) => PullFailureReason::SourceOther,
+        }
+    }
+}
+
+/// Upper bounds (milliseconds) of the source latency histogram buckets; the
+/// implicit last bucket is unbounded. Roughly logarithmic from 5 ms to 60 s.
+pub const SOURCE_LATENCY_BUCKET_BOUNDS_MS: [u64; 14] = [
+    5, 10, 20, 50, 100, 200, 500, 1_000, 2_000, 5_000, 10_000, 20_000, 30_000, 60_000,
+];
+
+#[derive(Debug, Default)]
+struct LatencyHistogram {
+    /// One counter per bound plus one for the overflow bucket.
+    buckets: [AtomicU64; SOURCE_LATENCY_BUCKET_BOUNDS_MS.len() + 1],
+    count: AtomicU64,
+    sum_ms: AtomicU64,
+}
+
+impl LatencyHistogram {
+    fn observe(&self, latency: Duration) {
+        let ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX);
+        let index = SOURCE_LATENCY_BUCKET_BOUNDS_MS
+            .iter()
+            .position(|bound| ms <= *bound)
+            .unwrap_or(SOURCE_LATENCY_BUCKET_BOUNDS_MS.len());
+        self.buckets[index].fetch_add(1, Ordering::Relaxed);
+        self.count.fetch_add(1, Ordering::Relaxed);
+        self.sum_ms.fetch_add(ms, Ordering::Relaxed);
+    }
+
+    fn snapshot(&self) -> SourceLatencySnapshot {
+        let mut cumulative = 0;
+        let buckets = SOURCE_LATENCY_BUCKET_BOUNDS_MS
+            .iter()
+            .zip(self.buckets.iter())
+            .map(|(bound, counter)| {
+                cumulative += counter.load(Ordering::Relaxed);
+                LatencyBucketSnapshot {
+                    le_ms: *bound,
+                    count: cumulative,
+                }
+            })
+            .collect();
+        SourceLatencySnapshot {
+            buckets,
+            count: self.count.load(Ordering::Relaxed),
+            sum_ms: self.sum_ms.load(Ordering::Relaxed),
+        }
+    }
+}
+
+/// The most recent source failure, kept for operators: class only, never the
+/// key or the message (which may echo attacker-controlled input).
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LastSourceError {
+    pub class: String,
+    #[serde(with = "time::serde::rfc3339")]
+    pub at: OffsetDateTime,
+}
+
+#[derive(Debug, Default)]
+pub struct OdmStats {
+    requests_total: [[AtomicU64; OdmOutcome::ALL.len()]; OdmOp::ALL.len()],
+    pulled_bytes_total: AtomicU64,
+    pulled_objects_total: [AtomicU64; PullPath::ALL.len()],
+    pull_failures_total: [AtomicU64; PullFailureReason::ALL.len()],
+    inflight_pulls: AtomicU64,
+    queue_depth: AtomicU64,
+    source_latency: LatencyHistogram,
+    last_source_error: Mutex<Option<LastSourceError>>,
+}
+
+impl OdmStats {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn record_request(&self, op: OdmOp, outcome: OdmOutcome) {
+        self.requests_total[op as usize][outcome as usize].fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn record_pulled_bytes(&self, bytes: u64) {
+        self.pulled_bytes_total.fetch_add(bytes, Ordering::Relaxed);
+    }
+
+    pub fn record_pulled_object(&self, path: PullPath) {
+        self.pulled_objects_total[path as usize].fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn record_pull_failure(&self, reason: PullFailureReason) {
+        self.pull_failures_total[reason as usize].fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn record_source_latency(&self, latency: Duration) {
+        self.source_latency.observe(latency);
+    }
+
+    pub fn record_source_error(&self, err: &SourceError) {
+        self.record_source_error_at(err, OffsetDateTime::now_utc());
+    }
+
+    pub fn record_source_error_at(&self, err: &SourceError, at: OffsetDateTime) {
+        *self.last_source_error.lock() = Some(LastSourceError {
+            class: err.class_label().to_string(),
+            at,
+        });
+    }
+
+    pub fn last_source_error(&self) -> Option<LastSourceError> {
+        self.last_source_error.lock().clone()
+    }
+
+    pub fn inflight_pulls(&self) -> u64 {
+        self.inflight_pulls.load(Ordering::Relaxed)
+    }
+
+    pub fn queue_depth(&self) -> u64 {
+        self.queue_depth.load(Ordering::Relaxed)
+    }
+
+    /// RAII increment of `inflight_pulls`.
+    pub fn inflight_guard(self: &Arc<Self>) -> GaugeGuard {
+        GaugeGuard::new(Arc::clone(self), OdmGauge::InflightPulls)
+    }
+
+    /// RAII increment of `queue_depth`.
+    pub fn queue_guard(self: &Arc<Self>) -> GaugeGuard {
+        GaugeGuard::new(Arc::clone(self), OdmGauge::QueueDepth)
+    }
+
+    fn gauge(&self, gauge: OdmGauge) -> &AtomicU64 {
+        match gauge {
+            OdmGauge::InflightPulls => &self.inflight_pulls,
+            OdmGauge::QueueDepth => &self.queue_depth,
+        }
+    }
+
+    /// Read-only, side-effect-free copy of every counter. The breaker lives
+    /// next to the stats in the bucket state; its state is passed in so the
+    /// snapshot stays a single document.
+    pub fn snapshot(&self, breaker_state: BreakerState) -> OdmStatsSnapshot {
+        let mut requests_total = BTreeMap::new();
+        for op in OdmOp::ALL {
+            let mut by_outcome = BTreeMap::new();
+            for outcome in OdmOutcome::ALL {
+                by_outcome.insert(
+                    outcome.as_str().to_string(),
+                    self.requests_total[op as usize][outcome as usize].load(Ordering::Relaxed),
+                );
+            }
+            requests_total.insert(op.as_str().to_string(), by_outcome);
+        }
+        let pulled_objects_total = PullPath::ALL
+            .iter()
+            .map(|path| {
+                (
+                    path.as_str().to_string(),
+                    self.pulled_objects_total[*path as usize].load(Ordering::Relaxed),
+                )
+            })
+            .collect();
+        let pull_failures_total = PullFailureReason::ALL
+            .iter()
+            .map(|reason| {
+                (
+                    reason.as_str().to_string(),
+                    self.pull_failures_total[*reason as usize].load(Ordering::Relaxed),
+                )
+            })
+            .collect();
+        OdmStatsSnapshot {
+            requests_total,
+            pulled_bytes_total: self.pulled_bytes_total.load(Ordering::Relaxed),
+            pulled_objects_total,
+            pull_failures_total,
+            inflight_pulls: self.inflight_pulls(),
+            queue_depth: self.queue_depth(),
+            source_latency: self.source_latency.snapshot(),
+            last_source_error: self.last_source_error(),
+            breaker_state,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum OdmGauge {
+    InflightPulls,
+    QueueDepth,
+}
+
+/// Increments a gauge on creation and decrements it on drop. Owns its
+/// `OdmStats` so it can live inside the pull slot handed to callers.
+#[derive(Debug)]
+pub struct GaugeGuard {
+    stats: Arc<OdmStats>,
+    gauge: OdmGauge,
+}
+
+impl GaugeGuard {
+    fn new(stats: Arc<OdmStats>, gauge: OdmGauge) -> Self {
+        stats.gauge(gauge).fetch_add(1, Ordering::Relaxed);
+        Self { stats, gauge }
+    }
+}
+
+impl Drop for GaugeGuard {
+    fn drop(&mut self) {
+        self.stats.gauge(self.gauge).fetch_sub(1, Ordering::Relaxed);
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LatencyBucketSnapshot {
+    /// Upper bound of the bucket in milliseconds.
+    pub le_ms: u64,
+    /// Cumulative observations at or below `le_ms`.
+    pub count: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SourceLatencySnapshot {
+    pub buckets: Vec<LatencyBucketSnapshot>,
+    /// Total observations, including those above the last bound.
+    pub count: u64,
+    pub sum_ms: u64,
+}
+
+/// Serializable copy of [`OdmStats`]. Every key is snake_case and every
+/// label set is fixed, so consumers can rely on the document shape.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OdmStatsSnapshot {
+    /// `op -> outcome -> count`.
+    pub requests_total: BTreeMap<String, BTreeMap<String, u64>>,
+    pub pulled_bytes_total: u64,
+    /// `path -> count`.
+    pub pulled_objects_total: BTreeMap<String, u64>,
+    /// `reason -> count`.
+    pub pull_failures_total: BTreeMap<String, u64>,
+    pub inflight_pulls: u64,
+    pub queue_depth: u64,
+    pub source_latency: SourceLatencySnapshot,
+    pub last_source_error: Option<LastSourceError>,
+    pub breaker_state: BreakerState,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+    use time::macros::datetime;
+
+    #[test]
+    fn snapshot_matches_golden_json() {
+        let stats = Arc::new(OdmStats::new());
+        stats.record_request(OdmOp::Get, OdmOutcome::SourceHit);
+        stats.record_request(OdmOp::Get, OdmOutcome::SourceHit);
+        stats.record_request(OdmOp::Head, OdmOutcome::NegativeCached);
+        stats.record_pulled_bytes(4096);
+        stats.record_pulled_object(PullPath::Inline);
+        stats.record_pull_failure(PullFailureReason::from(&SourceError::Timeout));
+        stats.record_source_latency(Duration::from_millis(3));
+        stats.record_source_latency(Duration::from_millis(750));
+        stats.record_source_latency(Duration::from_secs(90));
+        stats.record_source_error_at(&SourceError::ServerError(502), datetime!(2026-09-02 10:00:00 UTC));
+        let _inflight = stats.inflight_guard();
+        let _queued = stats.queue_guard();
+
+        let snapshot = stats.snapshot(BreakerState::HalfOpen);
+        let actual = serde_json::to_value(&snapshot).unwrap();
+        let expected = json!({
+            "requests_total": {
+                "get": {
+                    "breaker_open": 0, "filtered": 0, "negative_cached": 0, "source_error": 0,
+                    "source_hit": 2, "source_miss": 0, "unsupported": 0
+                },
+                "head": {
+                    "breaker_open": 0, "filtered": 0, "negative_cached": 1, "source_error": 0,
+                    "source_hit": 0, "source_miss": 0, "unsupported": 0
+                }
+            },
+            "pulled_bytes_total": 4096,
+            "pulled_objects_total": { "backfill": 0, "background": 0, "inline": 1 },
+            "pull_failures_total": {
+                "canceled": 0, "etag_mismatch": 0, "local_write": 0, "queue_full": 0,
+                "source_access_denied": 0, "source_connect": 0, "source_not_found": 0, "source_other": 0,
+                "source_server_error": 0, "source_throttled": 0, "source_timeout": 1, "source_unsupported": 0
+            },
+            "inflight_pulls": 1,
+            "queue_depth": 1,
+            "source_latency": {
+                "buckets": [
+                    { "le_ms": 5, "count": 1 }, { "le_ms": 10, "count": 1 }, { "le_ms": 20, "count": 1 },
+                    { "le_ms": 50, "count": 1 }, { "le_ms": 100, "count": 1 }, { "le_ms": 200, "count": 1 },
+                    { "le_ms": 500, "count": 1 }, { "le_ms": 1000, "count": 2 }, { "le_ms": 2000, "count": 2 },
+                    { "le_ms": 5000, "count": 2 }, { "le_ms": 10000, "count": 2 }, { "le_ms": 20000, "count": 2 },
+                    { "le_ms": 30000, "count": 2 }, { "le_ms": 60000, "count": 2 }
+                ],
+                "count": 3,
+                "sum_ms": 90753
+            },
+            "last_source_error": { "class": "server_error", "at": "2026-09-02T10:00:00Z" },
+            "breaker_state": "half_open"
+        });
+        assert_eq!(actual, expected);
+
+        let round_trip: OdmStatsSnapshot = serde_json::from_value(actual).unwrap();
+        assert_eq!(round_trip, snapshot);
+    }
+
+    #[test]
+    fn gauges_return_to_zero_when_guards_drop() {
+        let stats = Arc::new(OdmStats::new());
+        {
+            let _a = stats.inflight_guard();
+            let _b = stats.inflight_guard();
+            let _c = stats.queue_guard();
+            assert_eq!(stats.inflight_pulls(), 2);
+            assert_eq!(stats.queue_depth(), 1);
+        }
+        assert_eq!(stats.inflight_pulls(), 0);
+        assert_eq!(stats.queue_depth(), 0);
+    }
+
+    #[test]
+    fn pull_failure_reason_covers_every_source_error_class() {
+        let cases = [
+            (SourceError::NotFound, PullFailureReason::SourceNotFound),
+            (SourceError::AccessDenied, PullFailureReason::SourceAccessDenied),
+            (SourceError::Throttled, PullFailureReason::SourceThrottled),
+            (SourceError::Timeout, PullFailureReason::SourceTimeout),
+            (SourceError::Connect("x".into()), PullFailureReason::SourceConnect),
+            (SourceError::ServerError(500), PullFailureReason::SourceServerError),
+            (SourceError::Unsupported("x".into()), PullFailureReason::SourceUnsupported),
+            (SourceError::Other("x".into()), PullFailureReason::SourceOther),
+        ];
+        for (err, reason) in cases {
+            assert_eq!(PullFailureReason::from(&err), reason, "{err:?}");
+            assert_eq!(serde_json::to_string(&reason).unwrap(), format!("\"{}\"", reason.as_str()));
+        }
+    }
+
+    #[test]
+    fn label_lists_are_exhaustive_and_unique() {
+        let outcomes: std::collections::BTreeSet<_> = OdmOutcome::ALL.iter().map(|o| o.as_str()).collect();
+        assert_eq!(outcomes.len(), OdmOutcome::ALL.len());
+        let reasons: std::collections::BTreeSet<_> = PullFailureReason::ALL.iter().map(|r| r.as_str()).collect();
+        assert_eq!(reasons.len(), PullFailureReason::ALL.len());
+        let paths: std::collections::BTreeSet<_> = PullPath::ALL.iter().map(|p| p.as_str()).collect();
+        assert_eq!(paths.len(), PullPath::ALL.len());
+    }
+}

crates/ecstore/src/bucket/remote_s3_client.rs

@@ -0,0 +1,789 @@
+// 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"
+        );
+    }
+}

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/bucket_target_sys.rs` `validate_replication_target_endpoint`) |
+| 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) |
 | 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` |

rustfs/src/module_switches.rs

@@ -35,9 +35,14 @@ pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
 pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
 pub(crate) const ENV_BITROT_SELFTEST_ENABLE: &str = "RUSTFS_BITROT_SELFTEST_ENABLE";
 pub(crate) const ENV_BITROT_SELFTEST_STRICT: &str = "RUSTFS_BITROT_SELFTEST_STRICT";
+/// On-demand migration module switch (rustfs/backlog#2152). Off until GA
+/// (rustfs/backlog#2163) so every intermediate PR ships dark.
+pub(crate) const ENV_ON_DEMAND_MIGRATION_ENABLED: &str = "RUSTFS_ON_DEMAND_MIGRATION_ENABLED";
+pub(crate) const DEFAULT_ON_DEMAND_MIGRATION_ENABLED: bool = false;
 
 static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
 static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
+static ON_DEMAND_MIGRATION_MODULE_ENABLED: AtomicBool = AtomicBool::new(DEFAULT_ON_DEMAND_MIGRATION_ENABLED);
 
 /// Whether the data scanner is enabled, defaulting to on.
 pub(crate) fn scanner_enabled_from_env() -> bool {
@@ -80,3 +85,51 @@ pub fn is_notify_module_enabled() -> bool {
 pub(crate) fn set_notify_module_enabled(enabled: bool) {
     NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
 }
+
+/// Whether the on-demand migration module is enabled, defaulting to off.
+/// Read once at startup by `startup_bucket_metadata` and published below.
+pub(crate) fn on_demand_migration_enabled_from_env() -> bool {
+    rustfs_utils::get_env_bool(ENV_ON_DEMAND_MIGRATION_ENABLED, DEFAULT_ON_DEMAND_MIGRATION_ENABLED)
+}
+
+/// Last published on-demand migration module state.
+pub fn is_on_demand_migration_module_enabled() -> bool {
+    ON_DEMAND_MIGRATION_MODULE_ENABLED.load(Ordering::Relaxed)
+}
+
+/// Publish the on-demand migration module state resolved at startup. The
+/// ecstore runtime receives the same value through
+/// `OnDemandMigrationSys::set_module_enabled`, since ecstore cannot read
+/// this crate.
+pub(crate) fn set_on_demand_migration_module_enabled(enabled: bool) {
+    ON_DEMAND_MIGRATION_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn on_demand_migration_switch_defaults_off_and_follows_env() {
+        temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, None::<&str>, || {
+            assert!(!on_demand_migration_enabled_from_env());
+        });
+        temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, Some("true"), || {
+            assert!(on_demand_migration_enabled_from_env());
+        });
+        temp_env::with_var(ENV_ON_DEMAND_MIGRATION_ENABLED, Some("not-a-bool"), || {
+            assert!(!on_demand_migration_enabled_from_env(), "unparsable values keep the default");
+        });
+    }
+
+    #[test]
+    fn on_demand_migration_switch_publishes_to_the_cell() {
+        // The cell is process-global; restore it so sibling tests observe the default.
+        let before = is_on_demand_migration_module_enabled();
+        set_on_demand_migration_module_enabled(true);
+        assert!(is_on_demand_migration_module_enabled());
+        set_on_demand_migration_module_enabled(false);
+        assert!(!is_on_demand_migration_module_enabled());
+        set_on_demand_migration_module_enabled(before);
+    }
+}

rustfs/src/startup_bucket_metadata.rs

@@ -12,10 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::module_switches::{on_demand_migration_enabled_from_env, set_on_demand_migration_module_enabled};
 use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions};
 use crate::storage_api::startup::bucket_metadata::{
-    ECStore, Error as StorageError, Result as StorageResult, get_global_replication_pool, init_bucket_metadata_sys,
-    reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config,
+    ECStore, Error as StorageError, OnDemandMigrationSys, Result as StorageResult, get_global_replication_pool,
+    init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config,
 };
 use std::{
     io::{Error as IoError, Result as IoResult},
@@ -24,11 +25,13 @@ use std::{
 };
 use tokio_util::sync::CancellationToken;
 
+const EVENT_ON_DEMAND_MIGRATION_RUNTIME_INITIALIZED: &str = "on_demand_migration_runtime_initialized";
 const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_CANCELED: &str = "replication_resync_startup_background_canceled";
 const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_COMPLETED: &str = "replication_resync_startup_background_completed";
 const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_FAILED: &str = "replication_resync_startup_background_failed";
 const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_STARTED: &str = "replication_resync_startup_background_started";
 const LOG_COMPONENT_STARTUP_BUCKET_METADATA: &str = "startup_bucket_metadata";
+const LOG_SUBSYSTEM_ON_DEMAND_MIGRATION: &str = "on_demand_migration";
 const LOG_SUBSYSTEM_REPLICATION: &str = "replication";
 const METRIC_REPLICATION_RESYNC_STARTUP_BACKGROUND_DURATION_SECONDS: &str =
     "rustfs_replication_resync_startup_background_duration_seconds";
@@ -58,6 +61,7 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>, c
     let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
 
     try_migrate_bucket_metadata(store.clone()).await;
+    init_on_demand_migration_runtime();
     init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
     try_migrate_iam_config(store).await;
     spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx.clone(), false);
@@ -79,12 +83,33 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: Cance
     try_migrate_bucket_metadata(store.clone()).await;
 
     try_migrate_iam_config(store.clone()).await;
+    init_on_demand_migration_runtime();
     init_bucket_metadata_sys(store, buckets.clone()).await;
     spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx, true);
 
     Ok(buckets)
 }
 
+/// Publishes the on-demand migration module switch and registers the
+/// runtime's config hook before bucket metadata is loaded, so every cache
+/// install path (initial load included) reaches `OnDemandMigrationSys`
+/// (rustfs/backlog#2152). Idempotent across embedded and server startups.
+fn init_on_demand_migration_runtime() {
+    let enabled = on_demand_migration_enabled_from_env();
+    set_on_demand_migration_module_enabled(enabled);
+    let sys = OnDemandMigrationSys::get();
+    sys.set_module_enabled(enabled);
+    let hook_registered = sys.register_config_hook();
+    tracing::info!(
+        event = EVENT_ON_DEMAND_MIGRATION_RUNTIME_INITIALIZED,
+        component = LOG_COMPONENT_STARTUP_BUCKET_METADATA,
+        subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION,
+        state = if enabled { "enabled" } else { "disabled" },
+        hook_registered,
+        "On-demand migration runtime initialized"
+    );
+}
+
 fn spawn_bucket_resync_startup_reconcile(buckets: Vec<String>, ctx: CancellationToken, init_resync_after_reconcile: bool) {
     tokio::spawn(async move {
         describe_bucket_resync_startup_background_metrics();

rustfs/src/storage/storage_api.rs

@@ -407,8 +407,8 @@ pub(crate) mod ecstore_bucket {
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::bucket::lifecycle::tier_delete_journal::test_util::install_all_v6_fleet_capability_proof;
     pub(crate) use rustfs_ecstore::api::bucket::{
-        bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, policy_sys,
-        replication, tagging, target, utils,
+        bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, on_demand_migration,
+        policy_sys, replication, tagging, target, utils,
     };
     pub(crate) use rustfs_ecstore::api::bucket::{quota, versioning, versioning_sys};
 }

rustfs/src/storage_api.rs

@@ -290,6 +290,7 @@ pub(crate) mod startup {
             }
         }
 
+        pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::OnDemandMigrationSys;
         pub(crate) use crate::storage::storage_api::{
             ECStore, Error, Result, get_global_replication_pool, init_bucket_metadata_sys,
             reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config,

scripts/error-other-format-baseline.txt

@@ -5,7 +5,6 @@
 # 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