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

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

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

.config/e2e-full-selection.txt

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

Cargo.lock

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

crates/e2e_test/src/fake_s3_target/README.md

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

crates/e2e_test/src/lib.rs

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

crates/e2e_test/src/on_demand_migration/common.rs

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

crates/e2e_test/src/on_demand_migration/harness_self_test.rs

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

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

crates/ecstore/Cargo.toml

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

crates/ecstore/src/api/mod.rs

@@ -128,7 +128,6 @@ pub mod bucket {
     }
 
     pub mod metadata {
-        pub use crate::bucket::metadata::BUCKET_DURABILITY_CONFIG;
         pub use crate::bucket::metadata::{
             BUCKET_ACCELERATE_CONFIG, BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_LOGGING_CONFIG,
             BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_QUOTA_CONFIG_FILE,
@@ -137,6 +136,7 @@ pub mod bucket {
             BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BUCKET_WEBSITE_CONFIG, BucketMetadata, OBJECT_LOCK_CONFIG,
             load_bucket_metadata, table_catalog_path_hash,
         };
+        pub use crate::bucket::metadata::{BUCKET_DURABILITY_CONFIG, BUCKET_ON_DEMAND_MIGRATION_CONFIG};
     }
 
     pub mod durability {
@@ -145,6 +145,29 @@ pub mod bucket {
         };
     }
 
+    pub mod on_demand_migration {
+        pub use crate::bucket::on_demand_migration::{
+            ApplyOutcome, BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION,
+            Breaker, BreakerState, BreakerTransition, BreakerVerdict, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, GaugeGuard,
+            LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup,
+            OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason,
+            PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS,
+            SourceLatencySnapshot, source_client_spec,
+        };
+        pub use crate::bucket::on_demand_migration::{
+            ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
+            OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy,
+            SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+        };
+        pub mod source_client {
+            pub use crate::bucket::on_demand_migration::source_client::{
+                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,
+                SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
+                resolve_path_style,
+            };
+        }
+    }
+
     pub mod metadata_sys {
         #[cfg(feature = "test-util")]
         pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
@@ -154,11 +177,11 @@ pub mod bucket {
             delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy,
             get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
             get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
-            get_object_lock_config, get_object_lock_config_state, get_public_access_block_config, get_quota_config,
-            get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config,
-            get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata,
-            set_bucket_metadata, update, update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation,
-            update_quota_if_incarnation, update_under_transaction_lock,
+            get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config, get_public_access_block_config,
+            get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config,
+            get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata,
+            remove_bucket_metadata, set_bucket_metadata, update, update_bucket_targets_under_transaction_lock,
+            update_config_with, update_if_incarnation, update_quota_if_incarnation, update_under_transaction_lock,
         };
     }
 
@@ -166,16 +189,6 @@ pub mod bucket {
         pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config};
     }
 
-    pub mod on_demand_migration {
-        pub mod source_client {
-            pub use crate::bucket::on_demand_migration::source_client::{
-                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,
-                SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
-                resolve_path_style,
-            };
-        }
-    }
-
     pub mod object_lock {
         pub use crate::bucket::object_lock::{ObjectLockApi, ObjectLockStatusExt};
 

crates/ecstore/src/bucket/metadata.rs

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

crates/ecstore/src/bucket/metadata_sys.rs

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

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

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

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

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

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

@@ -0,0 +1,130 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Per-bucket cache of keys the source answered 404 for
+//! (rustfs/backlog#2152). A hit short-circuits the source lookup for
+//! `policy.negative_cache_ttl_secs`; a TTL of zero disables the cache.
+//!
+//! Entries are never invalidated on a local PUT: once the object exists
+//! locally the handler never consults ODM for it, so a stale negative entry
+//! is harmless.
+
+use std::time::Duration;
+
+/// Upper bound on remembered keys per bucket; LRU eviction beyond it.
+pub const NEGATIVE_CACHE_MAX_ENTRIES: u64 = 100_000;
+
+#[derive(Debug)]
+pub struct NegativeCache {
+    cache: Option<moka::sync::Cache<String, ()>>,
+    ttl: Duration,
+}
+
+impl NegativeCache {
+    /// `ttl == 0` builds a disabled cache that never records anything.
+    pub fn new(ttl: Duration) -> Self {
+        Self::with_capacity(ttl, NEGATIVE_CACHE_MAX_ENTRIES)
+    }
+
+    pub fn with_capacity(ttl: Duration, max_entries: u64) -> Self {
+        let cache = (!ttl.is_zero()).then(|| {
+            moka::sync::Cache::builder()
+                .max_capacity(max_entries)
+                .time_to_live(ttl)
+                .build()
+        });
+        Self { cache, ttl }
+    }
+
+    pub fn is_enabled(&self) -> bool {
+        self.cache.is_some()
+    }
+
+    pub fn ttl(&self) -> Duration {
+        self.ttl
+    }
+
+    /// Whether `key` is currently remembered as absent on the source.
+    pub fn contains(&self, key: &str) -> bool {
+        self.cache.as_ref().is_some_and(|cache| cache.get(key).is_some())
+    }
+
+    /// Remembers `key` as absent; no-op when disabled.
+    pub fn insert(&self, key: &str) {
+        if let Some(cache) = &self.cache {
+            cache.insert(key.to_string(), ());
+        }
+    }
+
+    /// Forgets `key` (e.g. after an admin-triggered backfill found it).
+    pub fn remove(&self, key: &str) {
+        if let Some(cache) = &self.cache {
+            cache.invalidate(key);
+        }
+    }
+
+    /// Approximate live entry count, for status snapshots only.
+    pub fn len(&self) -> u64 {
+        self.cache.as_ref().map_or(0, |cache| {
+            cache.run_pending_tasks();
+            cache.entry_count()
+        })
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn entry_expires_after_ttl() {
+        let cache = NegativeCache::new(Duration::from_millis(80));
+        assert!(cache.is_enabled());
+        cache.insert("a/x");
+        assert!(cache.contains("a/x"));
+        assert!(!cache.contains("a/y"));
+        std::thread::sleep(Duration::from_millis(160));
+        assert!(!cache.contains("a/x"), "entry must expire after the TTL");
+    }
+
+    #[test]
+    fn zero_ttl_disables_the_cache() {
+        let cache = NegativeCache::new(Duration::ZERO);
+        assert!(!cache.is_enabled());
+        cache.insert("a/x");
+        assert!(!cache.contains("a/x"));
+        assert!(cache.is_empty());
+    }
+
+    #[test]
+    fn remove_forgets_a_key() {
+        let cache = NegativeCache::new(Duration::from_secs(30));
+        cache.insert("a/x");
+        cache.remove("a/x");
+        assert!(!cache.contains("a/x"));
+    }
+
+    #[test]
+    fn capacity_bounds_entries() {
+        let cache = NegativeCache::with_capacity(Duration::from_secs(30), 4);
+        for i in 0..64 {
+            cache.insert(&format!("k{i}"));
+        }
+        assert!(cache.len() <= 4, "len {} exceeds capacity", cache.len());
+    }
+}

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

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

crates/rio/Cargo.toml

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

crates/rio/benches/tee_reader.rs

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

crates/rio/src/lib.rs

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

crates/scanner/src/scanner.rs

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

crates/scanner/src/scanner/cycle_state.rs

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

crates/scanner/src/scanner/leadership.rs

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

crates/scanner/src/scanner/tests.rs

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

docs/architecture/compat-cleanup-register.md

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

rustfs/src/module_switches.rs

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

rustfs/src/startup_bucket_metadata.rs

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

rustfs/src/storage/storage_api.rs

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

rustfs/src/storage_api.rs

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