fix(ci): refresh Linux full E2E selection

fix(scanner): accept bootstrap marker on timeout fence

.config/e2e-full-selection.txt

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

Cargo.lock

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

crates/e2e_test/src/fake_s3_target/README.md

@@ -1,11 +1,15 @@
 # Programmable fake S3 target
 
-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.
+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.
 
 `FakeS3Target::start()` creates the listener. Add target buckets with `create_bucket`, point a RustFS remote target at `address()`, use `FAKE_ACCESS_KEY` / `FAKE_SECRET_KEY`, then enqueue per-operation faults with `inject`. Faults for one operation are consumed in FIFO order and do not consume faults queued for another operation. A fault is consumed only after `s3s` verifies the full request signature, so anonymous, other-access-key, and bad-signature traffic cannot disturb a script.
 
-Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
+Supported data operations are HeadBucket, GetBucketVersioning, 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.
 
-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.
+`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.
 
-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.
+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.

crates/e2e_test/src/lib.rs

@@ -23,10 +23,17 @@ pub mod common;
 #[cfg(test)]
 pub mod chaos;
 
-// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8).
+// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8)
+// and on-demand-migration source scenarios (backlog#2151).
 #[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

@@ -0,0 +1,452 @@
+// 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

@@ -0,0 +1,606 @@
+// 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 unstalled_started = Instant::now();
+    client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
+    assert!(unstalled_started.elapsed() < Duration::from_millis(350), "stall is consumed once");
+
+    // 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

@@ -0,0 +1,24 @@
+// 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/src/api/mod.rs

@@ -166,6 +166,16 @@ pub mod bucket {
         pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config};
     }
 
+    pub mod on_demand_migration {
+        pub mod source_client {
+            pub use crate::bucket::on_demand_migration::source_client::{
+                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,
+                SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
+                resolve_path_style,
+            };
+        }
+    }
+
     pub mod object_lock {
         pub use crate::bucket::object_lock::{ObjectLockApi, ObjectLockStatusExt};
 
@@ -199,6 +209,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/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/mod.rs

@@ -0,0 +1,18 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! On-demand migration (ODM): serve and back-fill objects from an external
+//! S3-compatible source bucket.
+
+pub mod source_client;

crates/ecstore/src/bucket/remote_s3_client.rs

@@ -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"
+        );
+    }
+}

crates/rio/Cargo.toml

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

crates/rio/benches/tee_reader.rs

@@ -0,0 +1,103 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Throughput of `tee_reader` versus reading the same source directly:
+//! 64 MiB of data served in 1 MiB chunks, consumed with 1 MiB reads.
+
+use bytes::Bytes;
+use criterion::{Criterion, Throughput, criterion_group, criterion_main};
+use rustfs_rio::tee_reader;
+use std::hint::black_box;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
+
+const CHUNK_BYTES: usize = 1024 * 1024;
+const TOTAL_BYTES: usize = 64 * 1024 * 1024;
+const TEE_BUFFER_BYTES: usize = 4 * CHUNK_BYTES;
+
+/// In-memory source that serves at most `CHUNK_BYTES` per poll.
+struct ChunkedSource {
+    data: Bytes,
+    pos: usize,
+}
+
+impl AsyncRead for ChunkedSource {
+    fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+        let remaining = self.data.len() - self.pos;
+        let n = CHUNK_BYTES.min(remaining).min(buf.remaining());
+        buf.put_slice(&self.data[self.pos..self.pos + n]);
+        self.pos += n;
+        Poll::Ready(Ok(()))
+    }
+}
+
+async fn consume<R: AsyncRead + Unpin>(mut reader: R) -> usize {
+    let mut buf = vec![0u8; CHUNK_BYTES];
+    let mut total = 0;
+    loop {
+        let n = reader.read(&mut buf).await.expect("read");
+        if n == 0 {
+            return total;
+        }
+        total += n;
+    }
+}
+
+fn bench_tee_reader(c: &mut Criterion) {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()
+        .expect("build tokio runtime for tee_reader benchmark");
+    let data = Bytes::from(vec![0xA5u8; TOTAL_BYTES]);
+
+    let mut group = c.benchmark_group("tee_reader_64mib_1mib_chunks");
+    group.throughput(Throughput::Bytes(TOTAL_BYTES as u64));
+    group.sample_size(10);
+
+    group.bench_function("direct_read", |b| {
+        b.iter(|| {
+            let source = ChunkedSource {
+                data: data.clone(),
+                pos: 0,
+            };
+            let total = runtime.block_on(consume(source));
+            black_box(total)
+        })
+    });
+
+    group.bench_function("tee_primary_plus_secondary", |b| {
+        b.iter(|| {
+            let source = ChunkedSource {
+                data: data.clone(),
+                pos: 0,
+            };
+            let (primary, secondary) = tee_reader(source, TEE_BUFFER_BYTES);
+            let totals = runtime.block_on(async {
+                let secondary_task = tokio::spawn(consume(secondary));
+                let primary_total = consume(primary).await;
+                let secondary_total = secondary_task.await.expect("secondary task");
+                (primary_total, secondary_total)
+            });
+            black_box(totals)
+        })
+    });
+
+    group.finish();
+}
+
+criterion_group!(benches, bench_tee_reader);
+criterion_main!(benches);

crates/rio/src/lib.rs

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

crates/scanner/src/scanner.rs

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

crates/scanner/src/scanner/cycle_state.rs

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

crates/scanner/src/scanner/leadership.rs

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

crates/scanner/src/scanner/tests.rs

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

docs/architecture/compat-cleanup-register.md

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

docs/operations/outbound-connection-policy.md

@@ -18,7 +18,7 @@ RustFS validates every operator-configured outbound destination to close a serve
 | Target configuration validation (startup and admin API) | Full policy | `crates/targets/src/config/common.rs` `validate_outbound_http_url`; `rustfs/src/admin/handlers/target_descriptor.rs` |
 | OIDC discovery, JWKS, and token requests | Full policy | A blocked provider logs `OIDC provider discovery blocked by outbound policy` naming the origin to allowlist (`crates/iam/src/oidc.rs`) |
 | Object Lambda targets | Full policy | `rustfs/src/admin/router.rs` `outbound_policy` |
-| Bucket replication targets | Literal check, relaxed | Private addresses are always allowed; loopback only with `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET=true` (`crates/ecstore/src/bucket/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` |

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