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.

.config/e2e-full-selection.txt

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

crates/e2e_test/src/fake_s3_target/README.md

@@ -1,15 +1,11 @@
 # Programmable fake S3 target
 
-This module is the shared failure-injection boundary for replication end-to-end tests and the programmable external source for on-demand-migration (ODM) tests. It runs an in-process, path-style S3 endpoint backed by `s3s`; no production crate depends on it.
+This module is the shared failure-injection boundary for replication end-to-end tests. It runs an in-process, path-style S3 endpoint backed by `s3s`; no production crate depends on it.
 
 `FakeS3Target::start()` creates the listener. Add target buckets with `create_bucket`, point a RustFS remote target at `address()`, use `FAKE_ACCESS_KEY` / `FAKE_SECRET_KEY`, then enqueue per-operation faults with `inject`. Faults for one operation are consumed in FIFO order and do not consume faults queued for another operation. A fault is consumed only after `s3s` verifies the full request signature, so anonymous, other-access-key, and bad-signature traffic cannot disturb a script.
 
-Supported data operations are HeadBucket, GetBucketVersioning, ListObjectsV2, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets created with `create_bucket` are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
+Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
 
-`create_bucket_with_mode(name, BucketMode::Unversioned)` models a plain migration source: PUT overwrites in place, DELETE removes the key without a delete marker, GetBucketVersioning reports no status, and no `x-amz-version-id` is returned by PUT, GET, HEAD, tagging, or multipart completion. The only `versionId` such a bucket accepts is `null`; any other value is rejected with `InvalidArgument`. The mode is fixed at creation.
+Fault actions cover HTTP 401/403/503 responses, pre-dispatch delay, connection abort when a logical request-body threshold is reached, streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions. Each record also journals a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract.
 
-ListObjectsV2 lists current versions only (a key whose newest version is a delete marker is hidden) in byte order and supports `prefix`, `delimiter`, `max-keys` (clamped to 1000), `start-after`, and `continuation-token`; common prefixes count toward `max-keys`, `IsTruncated` / `NextContinuationToken` / `KeyCount` follow S3, and continuation tokens are opaque. `encoding-type` and `fetch-owner` are accepted but ignored, and ListObjects (v1) is not implemented. GET and HEAD honor `Range` in the `bytes=first-last`, `bytes=first-`, and `bytes=-suffix` forms with a 206 status, exact `Content-Range`, and `Accept-Ranges: bytes`; unsatisfiable ranges answer 416 `InvalidRange` with `Content-Range: bytes */<length>`. PUT and CreateMultipartUpload accept `Content-Type`, `Content-Encoding`, `Content-Disposition`, `Content-Language`, `Cache-Control`, `Expires`, and `x-amz-meta-*` (names stored lowercased), and HEAD/GET replay them verbatim together with `Last-Modified` and the ETag (hex MD5 for single PUTs, `<md5-of-part-md5s>-<parts>` for multipart objects). `put_seed_object` stores an object directly, bypassing the wire, the fault script, and the journal, so a source can be seeded without polluting the assertions a scenario later makes.
-
-Fault actions cover HTTP 401/403/503 responses (`Status`), any 4xx/5xx status paired with the matching S3 error code (`ResponseStatus`), pre-dispatch delay, holding a fully computed successful response before its first byte (`Stall`), connection abort when a logical request-body threshold is reached, GetObject bodies cut off after N bytes while `Content-Length` announces the full size (`TruncateBodyAt`), streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions and `count_requests(operation, key)` counts entries for one exact key. Each record journals the `Range` and `User-Agent` request headers, the ListObjectsV2 `prefix` and `continuation-token` query values, and a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract.
-
-The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type and each standard object header at 1 KiB. By default a PUT or uploaded part is capped at 64 MiB and a completed multipart object and all stored object/part data are capped at 128 MiB; `FakeS3Target::start_with_options(FakeS3TargetOptions { max_object_bytes })` raises the object cap up to 256 MiB, and the total budget then becomes twice the object cap (never below 128 MiB). Body drain, body-permit waits, delay, stall, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound.
+The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type at 1 KiB. A PUT or uploaded part is capped at 64 MiB; a completed multipart object and all stored object/part data are capped at 128 MiB. Body drain, body-permit waits, delay, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound.

crates/e2e_test/src/lib.rs

@@ -23,17 +23,10 @@ pub mod common;
 #[cfg(test)]
 pub mod chaos;
 
-// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8)
-// and on-demand-migration source scenarios (backlog#2151).
+// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8).
 #[cfg(test)]
 pub mod fake_s3_target;
 
-// On-demand migration (backlog#2147): shared two-server environment, admin
-// wrappers, and the harness self-test (backlog#2151). Behavior scenarios are
-// added by later ODM tasks.
-#[cfg(test)]
-pub mod on_demand_migration;
-
 // Socket-level network fault-injection proxy for black-box cluster tests
 // (backlog#1325 network fault-injection block): latency / blackhole / one-way
 // partition on the wire between nodes. Serves #1312/#1319 (lock-plane one-way

crates/e2e_test/src/on_demand_migration/common.rs

@@ -1,452 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Shared environment for on-demand migration (ODM) end-to-end tests.
-//!
-//! [`OdmTestEnv`] pairs one RustFS server under test with one in-process
-//! programmable S3 source ([`FakeS3Target`]). Admin calls target the route
-//! convention fixed by the tracking plan
-//! (`/rustfs/admin/v3/on-demand-migration/{bucket}`, JSON bodies); the
-//! server side lands with ODM-07, so until then the wrappers compile but are
-//! not exercised by the harness self-test.
-
-use crate::common::{RustFSTestEnvironment, signed_request};
-use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FakeS3TargetOptions, SeedMetadata};
-use aws_config::retry::RetryConfig;
-use aws_sdk_s3::Client;
-use aws_sdk_s3::config::{Credentials, Region};
-use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
-use bytes::Bytes;
-use serde::Serialize;
-use std::fmt;
-
-pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
-
-/// Module switch the server reads at startup (`false` before GA). The harness
-/// turns it on so scenario tests exercise the feature without repeating it.
-pub const ODM_MODULE_SWITCH_ENV: &str = "RUSTFS_ON_DEMAND_MIGRATION_ENABLED";
-/// Admin route prefix; the bucket name is appended as one path segment.
-pub const ODM_ADMIN_ROUTE: &str = "/rustfs/admin/v3/on-demand-migration";
-/// Region the fake source is addressed with (it accepts any SigV4 region).
-pub const FAKE_SOURCE_REGION: &str = "us-east-1";
-
-/// Wire form of the bucket-level ODM configuration (ODM-01 model). Every
-/// field is public so a scenario can tweak one knob and serialize the rest
-/// with the documented defaults.
-#[derive(Debug, Clone, Serialize)]
-pub struct OdmSourceSpec {
-    pub version: u32,
-    pub enabled: bool,
-    pub source: OdmSource,
-    pub filter: OdmFilter,
-    pub policy: OdmPolicy,
-}
-
-#[derive(Debug, Clone, Serialize)]
-pub struct OdmSource {
-    pub provider: String,
-    pub endpoint: String,
-    pub region: String,
-    pub bucket: String,
-    pub path_style: String,
-    pub credentials: Option<OdmCredentials>,
-    pub tls: OdmTls,
-}
-
-#[derive(Clone, Serialize)]
-pub struct OdmCredentials {
-    pub access_key: String,
-    pub secret_key: String,
-    pub session_token: Option<String>,
-}
-
-impl fmt::Debug for OdmCredentials {
-    /// Test logs are captured into CI artifacts; keep the secret out of them.
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("OdmCredentials")
-            .field("access_key", &self.access_key)
-            .field("secret_key", &"REDACTED")
-            .field("session_token", &self.session_token.as_ref().map(|_| "REDACTED"))
-            .finish()
-    }
-}
-
-#[derive(Debug, Clone, Default, Serialize)]
-pub struct OdmTls {
-    pub skip_verify: bool,
-    pub ca_cert_pem: Option<String>,
-}
-
-#[derive(Debug, Clone, Default, Serialize)]
-pub struct OdmFilter {
-    pub prefix: Option<String>,
-    pub source_prefix: Option<String>,
-}
-
-#[derive(Debug, Clone, Serialize)]
-pub struct OdmPolicy {
-    pub head: String,
-    pub range_get: String,
-    pub source_error: String,
-    pub respect_local_delete_marker: bool,
-    pub preserve_etag: bool,
-    pub copy_tags: bool,
-    pub emit_events: bool,
-    pub negative_cache_ttl_secs: u64,
-    pub inline_max_bytes: u64,
-    pub multipart_part_size_bytes: u64,
-    pub max_concurrent_pulls: u32,
-    pub pull_queue_capacity: u32,
-    pub source_timeout: OdmSourceTimeout,
-    pub bandwidth_limit_bytes_per_sec: Option<u64>,
-}
-
-#[derive(Debug, Clone, Serialize)]
-pub struct OdmSourceTimeout {
-    pub connect_ms: u64,
-    pub first_byte_ms: u64,
-    pub idle_ms: u64,
-}
-
-impl Default for OdmPolicy {
-    /// The ODM-01 defaults verbatim.
-    fn default() -> Self {
-        Self {
-            head: "proxy".to_string(),
-            range_get: "serve_and_backfill".to_string(),
-            source_error: "propagate".to_string(),
-            respect_local_delete_marker: true,
-            preserve_etag: true,
-            copy_tags: false,
-            emit_events: true,
-            negative_cache_ttl_secs: 30,
-            inline_max_bytes: 16 * 1024 * 1024,
-            multipart_part_size_bytes: 64 * 1024 * 1024,
-            max_concurrent_pulls: 8,
-            pull_queue_capacity: 1024,
-            source_timeout: OdmSourceTimeout {
-                connect_ms: 5_000,
-                first_byte_ms: 15_000,
-                idle_ms: 30_000,
-            },
-            bandwidth_limit_bytes_per_sec: None,
-        }
-    }
-}
-
-impl OdmSourceSpec {
-    /// Enabled configuration pointing at a bucket on the fake source with the
-    /// fixture credentials, path-style addressing, and default policy.
-    pub fn for_fake_source(source: &FakeS3Target, source_bucket: impl Into<String>) -> Self {
-        Self::new(
-            "s3",
-            source.endpoint(),
-            FAKE_SOURCE_REGION,
-            source_bucket,
-            FAKE_ACCESS_KEY,
-            FAKE_SECRET_KEY,
-        )
-    }
-
-    /// Enabled configuration pointing at a bucket on a second RustFS server
-    /// (see [`start_source_rustfs`]).
-    pub fn for_rustfs_source(source: &RustFSTestEnvironment, source_bucket: impl Into<String>) -> Self {
-        Self::new(
-            "rustfs",
-            &source.url,
-            FAKE_SOURCE_REGION,
-            source_bucket,
-            &source.access_key,
-            &source.secret_key,
-        )
-    }
-
-    fn new(
-        provider: &str,
-        endpoint: &str,
-        region: &str,
-        source_bucket: impl Into<String>,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Self {
-        Self {
-            version: 1,
-            enabled: true,
-            source: OdmSource {
-                provider: provider.to_string(),
-                endpoint: endpoint.to_string(),
-                region: region.to_string(),
-                bucket: source_bucket.into(),
-                path_style: "path".to_string(),
-                credentials: Some(OdmCredentials {
-                    access_key: access_key.to_string(),
-                    secret_key: secret_key.to_string(),
-                    session_token: None,
-                }),
-                tls: OdmTls::default(),
-            },
-            filter: OdmFilter::default(),
-            policy: OdmPolicy::default(),
-        }
-    }
-
-    pub fn to_json(&self) -> serde_json::Value {
-        serde_json::to_value(self).expect("ODM source spec serializes")
-    }
-}
-
-/// Backfill job control (ODM-12 route shape).
-#[derive(Debug, Clone)]
-pub enum BackfillOp {
-    Start(BackfillRequest),
-    Cancel,
-    Status,
-}
-
-#[derive(Debug, Clone, Default, Serialize)]
-pub struct BackfillRequest {
-    pub prefix: Option<String>,
-    pub skip_existing: Option<String>,
-    pub dry_run: bool,
-}
-
-/// Status plus raw body of an admin call, so a scenario can assert on the
-/// HTTP status first and only then parse the JSON.
-#[derive(Debug, Clone)]
-pub struct AdminResponse {
-    pub status: u16,
-    pub body: String,
-}
-
-impl AdminResponse {
-    pub fn json(&self) -> Result<serde_json::Value, BoxError> {
-        Ok(serde_json::from_str(&self.body)?)
-    }
-}
-
-/// One object to seed into the source.
-#[derive(Clone)]
-pub struct SeedObject {
-    pub key: String,
-    pub body: Bytes,
-    pub metadata: SeedMetadata,
-}
-
-impl SeedObject {
-    pub fn new(key: impl Into<String>, body: impl Into<Bytes>) -> Self {
-        Self {
-            key: key.into(),
-            body: body.into(),
-            metadata: SeedMetadata::new(),
-        }
-    }
-
-    pub fn with_metadata(mut self, metadata: SeedMetadata) -> Self {
-        self.metadata = metadata;
-        self
-    }
-}
-
-/// RustFS under test plus its fake S3 source.
-pub struct OdmTestEnv {
-    pub rustfs: RustFSTestEnvironment,
-    pub source: FakeS3Target,
-    /// S3 client for the RustFS under test.
-    pub client: Client,
-}
-
-impl OdmTestEnv {
-    /// Start a fake source with default limits and a RustFS server with the
-    /// ODM module switch enabled.
-    pub async fn start() -> Result<Self, BoxError> {
-        Self::start_with_options(FakeS3TargetOptions::default()).await
-    }
-
-    pub async fn start_with_options(options: FakeS3TargetOptions) -> Result<Self, BoxError> {
-        let source = FakeS3Target::start_with_options(options).await?;
-        let mut rustfs = RustFSTestEnvironment::new().await?;
-        rustfs
-            .start_rustfs_server_with_env(vec![], &[(ODM_MODULE_SWITCH_ENV, "true")])
-            .await?;
-        let client = rustfs.create_s3_client();
-        Ok(Self { rustfs, source, client })
-    }
-
-    /// S3 client addressing the fake source directly, for assertions on the
-    /// source's own state. Retries are off so a scripted fault is consumed by
-    /// exactly the request the test issued.
-    pub fn source_client(&self) -> Client {
-        fake_source_client(&self.source)
-    }
-
-    /// Enabled ODM configuration for `source_bucket` on the fake source.
-    pub fn fake_source_spec(&self, source_bucket: impl Into<String>) -> OdmSourceSpec {
-        OdmSourceSpec::for_fake_source(&self.source, source_bucket)
-    }
-
-    /// `PUT /rustfs/admin/v3/on-demand-migration/{bucket}` with the JSON spec.
-    pub async fn configure_source(&self, bucket: &str, spec: &OdmSourceSpec) -> Result<AdminResponse, BoxError> {
-        self.admin(http::Method::PUT, &format!("/{bucket}"), Some(spec.to_json()))
-            .await
-    }
-
-    /// Same as [`Self::configure_source`] with `dry-run=true`: validate and
-    /// probe without persisting.
-    pub async fn validate_source(&self, bucket: &str, spec: &OdmSourceSpec) -> Result<AdminResponse, BoxError> {
-        self.admin(http::Method::PUT, &format!("/{bucket}?dry-run=true"), Some(spec.to_json()))
-            .await
-    }
-
-    /// `GET .../{bucket}`: redacted configuration, 404 when unconfigured.
-    pub async fn get_config(&self, bucket: &str) -> Result<AdminResponse, BoxError> {
-        self.admin(http::Method::GET, &format!("/{bucket}"), None).await
-    }
-
-    /// `DELETE .../{bucket}`: remove the configuration (idempotent).
-    pub async fn disable(&self, bucket: &str) -> Result<AdminResponse, BoxError> {
-        self.admin(http::Method::DELETE, &format!("/{bucket}"), None).await
-    }
-
-    /// `GET .../{bucket}/status`: runtime snapshot.
-    pub async fn status(&self, bucket: &str) -> Result<AdminResponse, BoxError> {
-        self.admin(http::Method::GET, &format!("/{bucket}/status"), None).await
-    }
-
-    /// Backfill control: `POST .../{bucket}/backfill?op=start|cancel` or
-    /// `GET .../{bucket}/backfill` for the checkpoint.
-    pub async fn backfill(&self, bucket: &str, op: BackfillOp) -> Result<AdminResponse, BoxError> {
-        match op {
-            BackfillOp::Start(request) => {
-                self.admin(
-                    http::Method::POST,
-                    &format!("/{bucket}/backfill?op=start"),
-                    Some(serde_json::to_value(request)?),
-                )
-                .await
-            }
-            BackfillOp::Cancel => {
-                self.admin(http::Method::POST, &format!("/{bucket}/backfill?op=cancel"), None)
-                    .await
-            }
-            BackfillOp::Status => self.admin(http::Method::GET, &format!("/{bucket}/backfill"), None).await,
-        }
-    }
-
-    async fn admin(
-        &self,
-        method: http::Method,
-        path_and_query: &str,
-        body: Option<serde_json::Value>,
-    ) -> Result<AdminResponse, BoxError> {
-        let url = format!("{}{ODM_ADMIN_ROUTE}{path_and_query}", self.rustfs.url);
-        let body = body.map(|value| serde_json::to_vec(&value)).transpose()?;
-        let content_type = body.is_some().then_some("application/json");
-        let response = signed_request(method, &url, &self.rustfs.access_key, &self.rustfs.secret_key, body, content_type).await?;
-        Ok(AdminResponse {
-            status: response.status().as_u16(),
-            body: response.text().await?,
-        })
-    }
-
-    /// Store objects directly in the fake source (no wire traffic, no journal
-    /// entries). Returns the ETags in input order.
-    pub fn seed_source(&self, source_bucket: &str, objects: &[SeedObject]) -> Vec<String> {
-        objects
-            .iter()
-            .map(|object| {
-                self.source
-                    .put_seed_object(source_bucket, object.key.clone(), object.body.clone(), &object.metadata)
-            })
-            .collect()
-    }
-
-    /// Whether `key` is listed by the RustFS under test. Listing is served from
-    /// local state only, so this does not trigger a migration the way GET or
-    /// HEAD would.
-    pub async fn local_key_listed(&self, bucket: &str, key: &str) -> Result<bool, BoxError> {
-        let listed = self
-            .client
-            .list_objects_v2()
-            .bucket(bucket)
-            .prefix(key)
-            .max_keys(1)
-            .send()
-            .await?;
-        Ok(listed.contents().iter().any(|object| object.key() == Some(key)))
-    }
-
-    /// Panics unless `key` is stored locally with exactly `expected` bytes.
-    /// Presence is checked through listing first so a missing object fails
-    /// here instead of being pulled from the source by the GET.
-    pub async fn assert_local_present(&self, bucket: &str, key: &str, expected: &[u8]) {
-        assert!(
-            self.local_key_listed(bucket, key)
-                .await
-                .unwrap_or_else(|error| panic!("listing {bucket}/{key} failed: {error}")),
-            "{bucket}/{key} must be present locally"
-        );
-        let body = self
-            .client
-            .get_object()
-            .bucket(bucket)
-            .key(key)
-            .send()
-            .await
-            .unwrap_or_else(|error| panic!("GET {bucket}/{key} failed: {error}"))
-            .body
-            .collect()
-            .await
-            .unwrap_or_else(|error| panic!("reading {bucket}/{key} failed: {error}"))
-            .into_bytes();
-        assert_eq!(body.as_ref(), expected, "{bucket}/{key} local content mismatch");
-    }
-
-    /// Panics if `key` is listed locally.
-    pub async fn assert_local_absent(&self, bucket: &str, key: &str) {
-        assert!(
-            !self
-                .local_key_listed(bucket, key)
-                .await
-                .unwrap_or_else(|error| panic!("listing {bucket}/{key} failed: {error}")),
-            "{bucket}/{key} must be absent locally"
-        );
-    }
-}
-
-/// S3 client for the fake source with retries disabled (see
-/// [`OdmTestEnv::source_client`]).
-pub fn fake_source_client(source: &FakeS3Target) -> Client {
-    let credentials = Credentials::new(FAKE_ACCESS_KEY, FAKE_SECRET_KEY, None, None, "odm-fake-source");
-    Client::from_conf(
-        aws_sdk_s3::Config::builder()
-            .credentials_provider(credentials)
-            .region(Region::new(FAKE_SOURCE_REGION))
-            .endpoint_url(source.endpoint())
-            .force_path_style(true)
-            .behavior_version_latest()
-            .retry_config(RetryConfig::standard().with_max_attempts(1))
-            .http_client(SmithyHttpClientBuilder::new().build_http())
-            .build(),
-    )
-}
-
-/// Start a second, fully independent RustFS process (own port, data
-/// directory, and default credentials) to act as a real S3 source. It is
-/// spawned the same way `reliant::tiering` starts its cold tier; the process
-/// is stopped and its directory removed when the returned environment drops.
-pub async fn start_source_rustfs() -> Result<RustFSTestEnvironment, BoxError> {
-    let mut source = RustFSTestEnvironment::new().await?;
-    source.start_rustfs_server_without_cleanup(vec![]).await?;
-    Ok(source)
-}

crates/e2e_test/src/on_demand_migration/harness_self_test.rs

@@ -1,606 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Self-test of the ODM harness (rustfs/backlog#2151): the fake source's
-//! migration-facing surface (ListObjectsV2 paging, `Range`, unversioned
-//! buckets, metadata replay, fault actions) and the two-server environment.
-//! No ODM behavior is exercised here.
-
-use super::common::{OdmTestEnv, SeedObject, fake_source_client, start_source_rustfs};
-use crate::fake_s3_target::{BucketMode, FakeS3Target, FakeS3TargetOptions, FaultAction, Operation, SeedMetadata};
-use aws_sdk_s3::Client;
-use aws_sdk_s3::error::ProvideErrorMetadata;
-use aws_sdk_s3::primitives::{ByteStream, DateTime};
-use bytes::Bytes;
-use std::collections::BTreeSet;
-use std::time::{Duration, Instant};
-
-type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
-
-const SOURCE_BUCKET: &str = "odm-source";
-
-/// Position-dependent payload so a misaligned range read is caught.
-fn payload(len: usize) -> Bytes {
-    (0..len).map(|index| (index % 251) as u8).collect::<Vec<u8>>().into()
-}
-
-async fn fake_source() -> Result<(FakeS3Target, Client), Box<dyn std::error::Error + Send + Sync>> {
-    let source = FakeS3Target::start().await?;
-    source.create_bucket(SOURCE_BUCKET);
-    let client = fake_source_client(&source);
-    Ok((source, client))
-}
-
-/// Full ListObjectsV2 traversal. Returns `(keys, common prefixes, pages)` and
-/// checks the page shape on the way: every page except the last is full and
-/// truncated, the last carries no continuation token.
-async fn list_all(
-    client: &Client,
-    prefix: Option<&str>,
-    delimiter: Option<&str>,
-    start_after: Option<&str>,
-    max_keys: i32,
-) -> Result<(Vec<String>, Vec<String>, usize), Box<dyn std::error::Error + Send + Sync>> {
-    let mut keys = Vec::new();
-    let mut prefixes = Vec::new();
-    let mut pages = 0usize;
-    let mut token: Option<String> = None;
-    loop {
-        let page = client
-            .list_objects_v2()
-            .bucket(SOURCE_BUCKET)
-            .set_prefix(prefix.map(str::to_string))
-            .set_delimiter(delimiter.map(str::to_string))
-            .set_start_after(start_after.map(str::to_string))
-            .max_keys(max_keys)
-            .set_continuation_token(token.clone())
-            .send()
-            .await?;
-        pages += 1;
-        let page_keys: Vec<String> = page
-            .contents()
-            .iter()
-            .filter_map(|object| object.key().map(str::to_string))
-            .collect();
-        let page_prefixes: Vec<String> = page
-            .common_prefixes()
-            .iter()
-            .filter_map(|common| common.prefix().map(str::to_string))
-            .collect();
-        let entries = page_keys.len() + page_prefixes.len();
-        assert_eq!(page.key_count(), Some(entries as i32), "KeyCount must count keys and prefixes");
-        assert_eq!(page.continuation_token(), token.as_deref(), "the request token must be echoed");
-        keys.extend(page_keys);
-        prefixes.extend(page_prefixes);
-        if page.is_truncated() == Some(true) {
-            assert_eq!(entries as i32, max_keys, "every truncated page must be full");
-            token = Some(
-                page.next_continuation_token()
-                    .expect("truncated page must carry a continuation token")
-                    .to_string(),
-            );
-        } else {
-            assert!(page.next_continuation_token().is_none(), "final page must not carry a token");
-            return Ok((keys, prefixes, pages));
-        }
-    }
-}
-
-#[tokio::test]
-async fn fake_source_list_objects_v2_paginates_with_delimiter() -> TestResult {
-    let (source, client) = fake_source().await?;
-    let mut expected_keys = BTreeSet::new();
-    for directory in 0..30 {
-        for file in 0..30 {
-            expected_keys.insert(format!("d{directory:02}/k{file:03}"));
-        }
-    }
-    for index in 0..100 {
-        expected_keys.insert(format!("top-{index:03}"));
-    }
-    assert_eq!(expected_keys.len(), 1000);
-    for key in &expected_keys {
-        source.put_seed_object(SOURCE_BUCKET, key.clone(), Bytes::from(key.clone()), &SeedMetadata::new());
-    }
-    // A key whose current version is a delete marker must stay hidden.
-    client
-        .put_object()
-        .bucket(SOURCE_BUCKET)
-        .key("hidden/marker")
-        .body(ByteStream::from_static(b"gone"))
-        .send()
-        .await?;
-    client
-        .delete_object()
-        .bucket(SOURCE_BUCKET)
-        .key("hidden/marker")
-        .send()
-        .await?;
-    let expected_sorted: Vec<String> = expected_keys.iter().cloned().collect();
-    let expected_prefixes: Vec<String> = (0..30).map(|directory| format!("d{directory:02}/")).collect();
-    let expected_top: Vec<String> = (0..100).map(|index| format!("top-{index:03}")).collect();
-
-    // Flat traversal in byte order, 1000 keys in pages of 7.
-    let (keys, prefixes, pages) = list_all(&client, None, None, None, 7).await?;
-    assert_eq!(keys, expected_sorted);
-    assert!(prefixes.is_empty());
-    assert_eq!(pages, 143);
-
-    // Delimiter folding: 30 common prefixes then 100 top-level keys, pages of 7.
-    let (keys, prefixes, pages) = list_all(&client, None, Some("/"), None, 7).await?;
-    assert_eq!(prefixes, expected_prefixes);
-    assert_eq!(keys, expected_top);
-    assert_eq!(pages, 19);
-
-    // Empty prefix equals no prefix.
-    let (keys, _, _) = list_all(&client, Some(""), None, None, 1000).await?;
-    assert_eq!(keys, expected_sorted);
-
-    // No match: empty, not truncated, no token.
-    let (keys, prefixes, pages) = list_all(&client, Some("zzz/"), Some("/"), None, 7).await?;
-    assert!(keys.is_empty() && prefixes.is_empty());
-    assert_eq!(pages, 1);
-    let (keys, _, _) = list_all(&client, Some("hidden/"), None, None, 7).await?;
-    assert!(keys.is_empty(), "a current delete marker must hide its key");
-
-    // Exact page boundary: 30 keys under one directory, max-keys=30 -> one
-    // untruncated page.
-    let (keys, prefixes, pages) = list_all(&client, Some("d05/"), Some("/"), None, 30).await?;
-    assert_eq!(keys.len(), 30);
-    assert!(prefixes.is_empty());
-    assert_eq!(pages, 1);
-
-    // start-after skips keys at or before the marker.
-    let (keys, _, _) = list_all(&client, None, None, Some("top-097"), 1000).await?;
-    assert_eq!(keys, ["top-098", "top-099"]);
-
-    // max-keys is clamped to 1000; exactly 1000 keys fit in one page.
-    let (keys, _, pages) = list_all(&client, None, None, None, 5000).await?;
-    assert_eq!(keys.len(), 1000);
-    assert_eq!(pages, 1);
-
-    let listings: Vec<_> = source
-        .requests()
-        .into_iter()
-        .filter(|record| record.operation == Operation::ListObjectsV2)
-        .collect();
-    assert!(listings.len() >= 143 + 19);
-    assert!(listings.iter().any(|record| record.prefix.as_deref() == Some("d05/")));
-    assert!(
-        listings.iter().any(|record| record.continuation_token.is_some()),
-        "resumed pages must journal their continuation token"
-    );
-    assert!(listings.iter().all(|record| record.user_agent.is_some()));
-    source.shutdown().await;
-    Ok(())
-}
-
-#[tokio::test]
-async fn fake_source_range_get_variants_and_416() -> TestResult {
-    let (source, client) = fake_source().await?;
-    let body = payload(1000);
-    source.put_seed_object(SOURCE_BUCKET, "ranged", body.clone(), &SeedMetadata::new());
-
-    for (range, expected_range, expected_slice) in [
-        ("bytes=10-19", "bytes 10-19/1000", &body[10..20]),
-        ("bytes=990-", "bytes 990-999/1000", &body[990..]),
-        ("bytes=-5", "bytes 995-999/1000", &body[995..]),
-        ("bytes=0-5000", "bytes 0-999/1000", &body[..]),
-    ] {
-        let output = client
-            .get_object()
-            .bucket(SOURCE_BUCKET)
-            .key("ranged")
-            .range(range)
-            .send()
-            .await?;
-        assert_eq!(output.content_range(), Some(expected_range), "{range}");
-        assert_eq!(output.accept_ranges(), Some("bytes"), "{range}");
-        assert_eq!(output.content_length(), Some(expected_slice.len() as i64), "{range}");
-        let collected = output.body.collect().await?.into_bytes();
-        assert_eq!(collected.as_ref(), expected_slice, "{range}");
-    }
-    let head = client
-        .head_object()
-        .bucket(SOURCE_BUCKET)
-        .key("ranged")
-        .range("bytes=10-19")
-        .send()
-        .await?;
-    assert_eq!(head.content_range(), Some("bytes 10-19/1000"));
-    assert_eq!(head.content_length(), Some(10));
-
-    for range in ["bytes=1000-", "bytes=-0"] {
-        let error = client
-            .get_object()
-            .bucket(SOURCE_BUCKET)
-            .key("ranged")
-            .range(range)
-            .send()
-            .await
-            .expect_err("unsatisfiable range must fail");
-        let response = error.raw_response().expect("416 must retain the raw response");
-        assert_eq!(response.status().as_u16(), 416, "{range}");
-        assert_eq!(response.headers().get("content-range"), Some("bytes */1000"), "{range}");
-        assert_eq!(error.code(), Some("InvalidRange"), "{range}");
-    }
-
-    let ranged = source
-        .requests()
-        .into_iter()
-        .find(|record| record.operation == Operation::GetObject && record.range.as_deref() == Some("bytes=10-19"))
-        .expect("the Range header must be journaled verbatim");
-    assert_eq!(ranged.key.as_deref(), Some("ranged"));
-    assert!(source.count_requests(Operation::GetObject, "ranged") >= 6);
-    source.shutdown().await;
-    Ok(())
-}
-
-#[tokio::test]
-async fn fake_source_unversioned_bucket_overwrites_and_deletes() -> TestResult {
-    let (source, client) = fake_source().await?;
-    source.create_bucket_with_mode("plain-source", BucketMode::Unversioned);
-    let versioning = client.get_bucket_versioning().bucket("plain-source").send().await?;
-    assert!(versioning.status().is_none(), "unversioned bucket must report no versioning status");
-
-    let first = client
-        .put_object()
-        .bucket("plain-source")
-        .key("doc")
-        .body(ByteStream::from_static(b"first"))
-        .send()
-        .await?;
-    assert!(first.version_id().is_none());
-    let second = client
-        .put_object()
-        .bucket("plain-source")
-        .key("doc")
-        .body(ByteStream::from_static(b"second"))
-        .send()
-        .await?;
-    assert!(second.version_id().is_none());
-    let get = client.get_object().bucket("plain-source").key("doc").send().await?;
-    assert!(get.version_id().is_none(), "GET must not return x-amz-version-id");
-    assert_eq!(get.body.collect().await?.into_bytes().as_ref(), b"second");
-    let head = client.head_object().bucket("plain-source").key("doc").send().await?;
-    assert!(head.version_id().is_none(), "HEAD must not return x-amz-version-id");
-    assert_eq!(source.stored_versions("plain-source", "doc").len(), 1, "overwrite must replace in place");
-
-    let deleted = client.delete_object().bucket("plain-source").key("doc").send().await?;
-    assert!(deleted.delete_marker().is_none() && deleted.version_id().is_none());
-    let missing = client
-        .get_object()
-        .bucket("plain-source")
-        .key("doc")
-        .send()
-        .await
-        .expect_err("deleted object must be gone");
-    assert_eq!(missing.raw_response().map(|response| response.status().as_u16()), Some(404));
-    assert_eq!(missing.code(), Some("NoSuchKey"));
-    let missing_head = client
-        .head_object()
-        .bucket("plain-source")
-        .key("doc")
-        .send()
-        .await
-        .expect_err("deleted object must fail HEAD");
-    assert_eq!(missing_head.raw_response().map(|response| response.status().as_u16()), Some(404));
-    assert!(source.stored_versions("plain-source", "doc").is_empty(), "DELETE must not leave a marker");
-
-    // The versioned bucket on the same target keeps its version ids.
-    let versioned = client
-        .put_object()
-        .bucket(SOURCE_BUCKET)
-        .key("doc")
-        .body(ByteStream::from_static(b"versioned"))
-        .send()
-        .await?;
-    assert!(versioned.version_id().is_some());
-    source.shutdown().await;
-    Ok(())
-}
-
-#[tokio::test]
-async fn fake_source_replays_standard_and_user_metadata() -> TestResult {
-    let (source, client) = fake_source().await?;
-    let body = payload(4096);
-    let expected_etag = format!("\"{}\"", {
-        use md5::Digest as _;
-        hex_simd::encode_to_string(md5::Md5::digest(&body), hex_simd::AsciiCase::Lower)
-    });
-    // 2026-01-01T00:00:00Z rendered as an HTTP date by the SDK.
-    let expires = DateTime::from_secs(1_767_225_600);
-    client
-        .put_object()
-        .bucket(SOURCE_BUCKET)
-        .key("meta")
-        .body(ByteStream::from(body.clone()))
-        .content_type("application/x-odm")
-        .content_encoding("gzip")
-        .content_disposition("attachment; filename=\"meta.bin\"")
-        .content_language("en-US")
-        .cache_control("max-age=60")
-        .expires(expires)
-        .metadata("Foo-Bar", "mixed case name")
-        .metadata("UPPER", "upper name")
-        .metadata("already-lower", "lower name")
-        .send()
-        .await?;
-
-    let head = client.head_object().bucket(SOURCE_BUCKET).key("meta").send().await?;
-    let get = client.get_object().bucket(SOURCE_BUCKET).key("meta").send().await?;
-    for (label, content_type, content_encoding, content_disposition, content_language, cache_control, expires_string, e_tag) in [
-        (
-            "HEAD",
-            head.content_type(),
-            head.content_encoding(),
-            head.content_disposition(),
-            head.content_language(),
-            head.cache_control(),
-            head.expires_string(),
-            head.e_tag(),
-        ),
-        (
-            "GET",
-            get.content_type(),
-            get.content_encoding(),
-            get.content_disposition(),
-            get.content_language(),
-            get.cache_control(),
-            get.expires_string(),
-            get.e_tag(),
-        ),
-    ] {
-        assert_eq!(content_type, Some("application/x-odm"), "{label}");
-        assert_eq!(content_encoding, Some("gzip"), "{label}");
-        assert_eq!(content_disposition, Some("attachment; filename=\"meta.bin\""), "{label}");
-        assert_eq!(content_language, Some("en-US"), "{label}");
-        assert_eq!(cache_control, Some("max-age=60"), "{label}");
-        assert_eq!(expires_string, Some("Thu, 01 Jan 2026 00:00:00 GMT"), "{label}");
-        assert_eq!(e_tag, Some(expected_etag.as_str()), "{label}");
-    }
-    for metadata in [head.metadata(), get.metadata()] {
-        let metadata = metadata.expect("user metadata must be replayed");
-        assert_eq!(metadata.get("foo-bar").map(String::as_str), Some("mixed case name"));
-        assert_eq!(metadata.get("upper").map(String::as_str), Some("upper name"));
-        assert_eq!(metadata.get("already-lower").map(String::as_str), Some("lower name"));
-        assert!(!metadata.contains_key("Foo-Bar") && !metadata.contains_key("UPPER"));
-    }
-    assert!(head.last_modified().is_some());
-    assert_eq!(head.last_modified(), get.last_modified());
-    assert_eq!(head.content_length(), Some(4096));
-    assert_eq!(get.body.collect().await?.into_bytes(), body);
-
-    // Seeded objects replay the same way.
-    let seeded_etag = source.put_seed_object(
-        SOURCE_BUCKET,
-        "seeded",
-        Bytes::from_static(b"seeded"),
-        &SeedMetadata::new()
-            .content_type("text/plain")
-            .content_encoding("identity")
-            .cache_control("no-store")
-            .user_metadata("Origin", "seed"),
-    );
-    let seeded = client.head_object().bucket(SOURCE_BUCKET).key("seeded").send().await?;
-    assert_eq!(seeded.e_tag(), Some(format!("\"{seeded_etag}\"").as_str()));
-    assert_eq!(seeded.content_type(), Some("text/plain"));
-    assert_eq!(seeded.content_encoding(), Some("identity"));
-    assert_eq!(seeded.cache_control(), Some("no-store"));
-    assert_eq!(
-        seeded
-            .metadata()
-            .and_then(|metadata| metadata.get("origin"))
-            .map(String::as_str),
-        Some("seed")
-    );
-    source.shutdown().await;
-    Ok(())
-}
-
-#[tokio::test]
-async fn fake_source_fault_actions_truncate_stall_and_status() -> TestResult {
-    let (source, client) = fake_source().await?;
-    let body = payload(4096);
-    source.put_seed_object(SOURCE_BUCKET, "faulty", body.clone(), &SeedMetadata::new());
-
-    // TruncateBodyAt: headers promise 4096 bytes, the body ends after 100.
-    source.inject_for_key(Operation::GetObject, "faulty", FaultAction::TruncateBodyAt(100), 1);
-    let truncated = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
-    assert_eq!(truncated.content_length(), Some(4096));
-    let short_read = truncated
-        .body
-        .collect()
-        .await
-        .expect_err("a truncated body must fail to collect");
-    let short_read = short_read.to_string();
-    assert!(!short_read.is_empty());
-
-    // ResponseStatus: arbitrary status with the matching S3 error code.
-    for (code, expected_code) in [
-        (429u16, "SlowDown"),
-        (404, "NoSuchKey"),
-        (500, "InternalError"),
-        (503, "ServiceUnavailable"),
-    ] {
-        source.inject(Operation::GetObject, FaultAction::ResponseStatus(code), 1);
-        let error = client
-            .get_object()
-            .bucket(SOURCE_BUCKET)
-            .key("faulty")
-            .send()
-            .await
-            .expect_err("scripted status must fail");
-        assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(code));
-        assert_eq!(error.code(), Some(expected_code));
-    }
-
-    // Stall: the fully computed response is held before its first byte.
-    source.inject(Operation::HeadObject, FaultAction::Stall(Duration::from_millis(400)), 1);
-    let started = Instant::now();
-    let stalled = client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
-    assert!(started.elapsed() >= Duration::from_millis(350), "stall must delay the first byte");
-    assert_eq!(stalled.content_length(), Some(4096));
-    let post_stall_started = Instant::now();
-    client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
-    assert!(post_stall_started.elapsed() < Duration::from_millis(350), "stall is consumed once");
-
-    // The object is intact once the script is drained.
-    let intact = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
-    assert_eq!(intact.body.collect().await?.into_bytes(), body);
-
-    assert_eq!(source.count_requests(Operation::GetObject, "faulty"), 6);
-    assert_eq!(source.count_requests(Operation::HeadObject, "faulty"), 2);
-    assert_eq!(source.count_requests(Operation::GetObject, "other"), 0);
-    let records = source.requests();
-    assert!(
-        records.iter().all(|record| record
-            .user_agent
-            .as_deref()
-            .is_some_and(|agent| agent.contains("aws-sdk-rust"))),
-        "the SDK user agent must be journaled"
-    );
-    assert!(
-        records
-            .iter()
-            .any(|record| record.fault == Some(FaultAction::TruncateBodyAt(100)))
-    );
-    assert!(
-        records
-            .iter()
-            .any(|record| record.fault == Some(FaultAction::Stall(Duration::from_millis(400))))
-    );
-    source.shutdown().await;
-    Ok(())
-}
-
-#[tokio::test]
-async fn fake_source_raised_object_cap_accepts_large_put() -> TestResult {
-    let source = FakeS3Target::start_with_options(FakeS3TargetOptions {
-        max_object_bytes: 96 * 1024 * 1024,
-    })
-    .await?;
-    source.create_bucket(SOURCE_BUCKET);
-    let client = fake_source_client(&source);
-    let len = 64 * 1024 * 1024 + 1;
-    client
-        .put_object()
-        .bucket(SOURCE_BUCKET)
-        .key("large")
-        .body(ByteStream::from(vec![7u8; len]))
-        .send()
-        .await?;
-    let head = client.head_object().bucket(SOURCE_BUCKET).key("large").send().await?;
-    assert_eq!(head.content_length(), Some(len as i64));
-    let tail = client
-        .get_object()
-        .bucket(SOURCE_BUCKET)
-        .key("large")
-        .range("bytes=-1")
-        .send()
-        .await?;
-    assert_eq!(tail.content_range(), Some(format!("bytes {}-{}/{len}", len - 1, len - 1).as_str()));
-    source.shutdown().await;
-    Ok(())
-}
-
-#[tokio::test]
-async fn odm_env_starts_rustfs_and_fake_source() -> TestResult {
-    let env = OdmTestEnv::start().await?;
-    env.source.create_bucket(SOURCE_BUCKET);
-    let local_bucket = "odm-local";
-    env.rustfs.create_test_bucket(local_bucket).await?;
-
-    let etags = env.seed_source(
-        SOURCE_BUCKET,
-        &[
-            SeedObject::new("seed/a", Bytes::from_static(b"alpha")),
-            SeedObject::new("seed/b", Bytes::from_static(b"beta"))
-                .with_metadata(SeedMetadata::new().content_type("text/plain").user_metadata("Kind", "seed")),
-        ],
-    );
-    assert_eq!(etags.len(), 2);
-    assert!(env.source.requests().is_empty(), "seeding must not touch the journal");
-    let source_client = env.source_client();
-    let seeded = source_client.head_object().bucket(SOURCE_BUCKET).key("seed/b").send().await?;
-    assert_eq!(seeded.content_type(), Some("text/plain"));
-    assert_eq!(seeded.e_tag(), Some(format!("\"{}\"", etags[1]).as_str()));
-    assert_eq!(env.source.count_requests(Operation::HeadObject, "seed/b"), 1);
-
-    env.assert_local_absent(local_bucket, "seed/a").await;
-    env.client
-        .put_object()
-        .bucket(local_bucket)
-        .key("seed/a")
-        .body(ByteStream::from_static(b"alpha"))
-        .send()
-        .await?;
-    env.assert_local_present(local_bucket, "seed/a", b"alpha").await;
-    env.assert_local_absent(local_bucket, "seed/b").await;
-
-    let spec = env.fake_source_spec(SOURCE_BUCKET).to_json();
-    assert_eq!(spec["version"], 1);
-    assert_eq!(spec["enabled"], true);
-    assert_eq!(spec["source"]["provider"], "s3");
-    assert_eq!(spec["source"]["endpoint"], env.source.endpoint());
-    assert_eq!(spec["source"]["bucket"], SOURCE_BUCKET);
-    assert_eq!(spec["source"]["credentials"]["secret_key"], "fake-secret");
-    assert_eq!(spec["policy"]["source_timeout"]["first_byte_ms"], 15_000);
-    assert!(spec["policy"]["bandwidth_limit_bytes_per_sec"].is_null());
-    let debug = format!("{:?}", env.fake_source_spec(SOURCE_BUCKET));
-    assert!(!debug.contains("fake-secret"), "Debug output must redact the secret");
-    Ok(())
-}
-
-#[tokio::test]
-async fn start_source_rustfs_round_trips_put_get() -> TestResult {
-    let env = OdmTestEnv::start().await?;
-    let source = start_source_rustfs().await?;
-    assert_ne!(source.url, env.rustfs.url, "the source must be a separate instance");
-
-    source.create_test_bucket(SOURCE_BUCKET).await?;
-    let source_client = source.create_s3_client();
-    let body = payload(70_000);
-    let put = source_client
-        .put_object()
-        .bucket(SOURCE_BUCKET)
-        .key("real/object")
-        .body(ByteStream::from(body.clone()))
-        .content_type("application/octet-stream")
-        .send()
-        .await?;
-    assert!(put.e_tag().is_some());
-    let get = source_client
-        .get_object()
-        .bucket(SOURCE_BUCKET)
-        .key("real/object")
-        .send()
-        .await?;
-    assert_eq!(get.content_type(), Some("application/octet-stream"));
-    assert_eq!(get.body.collect().await?.into_bytes(), body);
-
-    let visible_to_primary = env
-        .client
-        .list_buckets()
-        .send()
-        .await?
-        .buckets()
-        .iter()
-        .any(|bucket| bucket.name() == Some(SOURCE_BUCKET));
-    assert!(!visible_to_primary, "the two servers must not share state");
-    let spec = super::common::OdmSourceSpec::for_rustfs_source(&source, SOURCE_BUCKET).to_json();
-    assert_eq!(spec["source"]["provider"], "rustfs");
-    assert_eq!(spec["source"]["endpoint"], source.url);
-    Ok(())
-}

crates/e2e_test/src/on_demand_migration/mod.rs

@@ -1,24 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! On-demand migration (ODM) end-to-end suite (rustfs/backlog#2147).
-//!
-//! `common` is the shared environment: one RustFS under test, one programmable
-//! fake S3 source, admin-API wrappers, seeding and local-state assertions.
-//! `harness_self_test` proves the harness itself; ODM behavior scenarios are
-//! separate modules wired by later tasks.
-
-pub mod common;
-
-mod harness_self_test;

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