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

* feat(ecstore): add on-demand migration bucket config model

Introduce OnDemandMigrationConfig (deny_unknown_fields, version 1) with typed validation, credential redaction, a secret-free Debug impl, and the OnceLock publish hook the runtime registers into. Exported through the api facade.

* feat(ecstore): persist on-demand migration config in bucket metadata

Store the config as a RustFS extension entry (on-demand-migration.json) with its update time in .metadata.bin, add the typed BucketMetadataSys accessor, and publish the config through the hook on every cache-install path alongside the durability sync.

* refactor(ecstore): extract shared remote S3 client builder

Move the aws_sdk_s3 client construction out of bucket_target_sys into
bucket/remote_s3_client.rs: endpoint assembly, credential provider,
path-style selection, custom CA / skip-TLS transports and the outbound
SSRF gate now build from a neutral RemoteS3EndpointSpec so replication
targets and the upcoming on-demand migration source client share one
policy. Replication builds its client through From<&BucketTarget>; the
gate keeps its relaxed semantics (private allowed, loopback only behind
RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET) verbatim. The builder also
gains optional connect/read timeouts and a User-Agent suffix
interceptor, both unset for replication.

Refs rustfs/backlog#2149

* feat(ecstore): add on-demand migration SourceClient

Add bucket/on_demand_migration/source_client.rs on top of the shared
remote S3 builder: HEAD, ranged streaming GET, ListObjectsV2 with
source-prefix mapping, GetObjectTagging and an admin probe. Every request
carries the x-rustfs-/x-minio-source-proxy-request anti-loop markers and
a RustFS-OnDemandMigration/<version> User-Agent suffix; SSE-C source
objects are rejected as unsupported. SourceError classifies SDK failures
(not found, access denied, throttled, timeout, connect, server error)
with retryability and a stable metrics label. Debug output redacts
credentials.

Refs rustfs/backlog#2149

* docs(operations): point outbound policy at shared remote S3 client builder

* chore: integrate ODM-01 and ODM-02 as B1 base (fix facade merge)

* 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.

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,