test(e2e): cover on-demand migration faults, concurrency and interactions (#7085)

* test(odm): add the fault, concurrency, interaction and real-source e2e

Twenty-one scenarios for on-demand migration: source failures and the
circuit breaker, single-flight and the pull-concurrency limits, how a
pulled object composes with encryption, Object Lock, quota, events,
replication, delete markers and the disable switch, and two cases against
a real second RustFS as the source.

Every assertion carries a source-request-count dimension so a case cannot
pass by serving the client while quietly re-reading the source.

* ci(odm): route the slow on-demand-migration e2e to the nightly lane

The fault, concurrency and real-source modules wait out the 30 s breaker
window, drive 100-deep bursts, or start extra RustFS processes, so they
join e2e-nightly and are subtracted from the e2e-full merge budget. The
e2e-smoke selection is unchanged.

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=9dccb0cd537cf79ae70c1c20e8281d36d03f2f09f81142a5341e26e3dc18709d
+sha256-darwin=09bc29b3e0ed7c5779331740e18e1d5ca05cb240c1363d1ecc95c1ef1e337038
 sha256-linux=86e69337ad1440252a2ee20a12063c989ed12442d3b1ddf9e9233acf0f2ec089

.config/e2e-nightly-selection.txt

@@ -1 +1 @@
-sha256=d06524b44de97ed8f62b0fd8cf9fa504e3cd520ffcaacc32691d6f890ebe7f20
+sha256=e58d2993aef56373d52ea40dfc6733fef42865195a71ccbd3883b0fafd84da75

.config/nextest.toml

@@ -475,10 +475,20 @@ path = "junit.xml"
 # loss, quorum, cleanup, notification fan-in, or admin-timeout behavior. The
 # consolidated nightly workflow runs them serially to avoid resource
 # starvation; failures are never retried.
+#
+# On-demand migration (backlog#2158 ODM-11) joins by the second clause: the
+# fault matrix waits out the 30 s circuit-breaker window, the concurrency
+# matrix drives 100-deep bursts, and the real-source cases start a second
+# (loop guard: a third) RustFS process. They are too slow or too heavy for
+# the merge budget; `on_demand_migration::{get_basic,interaction}_test` stay
+# in e2e-full, which excludes exactly these three modules.
 [profile.e2e-nightly]
 default-filter = """
     package(e2e_test)
-    & test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|degraded_listing_availability_test|heal_erasure_disk_rebuild_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/)
+    & (
+        test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|degraded_listing_availability_test|heal_erasure_disk_rebuild_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/)
+        | test(/^on_demand_migration::(concurrency_test|fault_test|real_source_test)::/)
+    )
 """
 fail-fast = false
 
@@ -536,6 +546,7 @@ default-filter = """
     & !test(/^protocols::/)
     & !test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|degraded_listing_availability_test|heal_erasure_disk_rebuild_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/)
     & !test(/^replication_extension_test::/)
+    & !test(/^on_demand_migration::(concurrency_test|fault_test|real_source_test)::/)
 """
 fail-fast = false
 

crates/e2e_test/src/on_demand_migration/common.rs

@@ -22,7 +22,9 @@
 //! 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, Operation, SeedMetadata};
+use crate::fake_s3_target::{
+    BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FakeS3TargetOptions, Operation, SeedMetadata,
+};
 use aws_config::retry::RetryConfig;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::config::{Credentials, Region};
@@ -277,6 +279,21 @@ impl SeedObject {
     }
 }
 
+/// Process arguments and environment a scenario needs on top of the ODM
+/// defaults, plus the fake source's own limits.
+#[derive(Debug, Default)]
+pub struct OdmEnvOptions<'a> {
+    pub source: FakeS3TargetOptions,
+    pub args: Vec<&'a str>,
+    pub env: Vec<(&'a str, &'a str)>,
+    /// Start the server with the file-backed KMS and a default key, so
+    /// bucket default encryption (SSE-S3) can be configured.
+    pub local_kms: bool,
+}
+
+/// Default key id of the [`OdmEnvOptions::local_kms`] backend.
+pub const LOCAL_KMS_DEFAULT_KEY_ID: &str = "rustfs-odm-e2e-default-key";
+
 /// RustFS under test plus its fake S3 source.
 pub struct OdmTestEnv {
     pub rustfs: RustFSTestEnvironment,
@@ -293,11 +310,44 @@ impl OdmTestEnv {
     }
 
     pub async fn start_with_options(options: FakeS3TargetOptions) -> Result<Self, BoxError> {
-        let source = FakeS3Target::start_with_options(options).await?;
+        Self::start_with(OdmEnvOptions {
+            source: options,
+            ..OdmEnvOptions::default()
+        })
+        .await
+    }
+
+    /// Start the pair with extra process arguments and environment for the
+    /// server under test (KMS, the usage scanner, replication timing). The
+    /// ODM module switch and the loopback-source opt-in are always set; a
+    /// caller-supplied entry with the same name wins.
+    pub async fn start_with(options: OdmEnvOptions<'_>) -> Result<Self, BoxError> {
+        let source = FakeS3Target::start_with_options(options.source).await?;
         let mut rustfs = RustFSTestEnvironment::new().await?;
-        rustfs
-            .start_rustfs_server_with_env(vec![], &[(ODM_MODULE_SWITCH_ENV, "true"), (ALLOW_LOOPBACK_SOURCE_ENV, "true")])
-            .await?;
+        let mut env: Vec<(&str, &str)> = vec![(ODM_MODULE_SWITCH_ENV, "true"), (ALLOW_LOOPBACK_SOURCE_ENV, "true")];
+        for (name, value) in &options.env {
+            match env.iter_mut().find(|(existing, _)| existing == name) {
+                Some(entry) => entry.1 = value,
+                None => env.push((name, value)),
+            }
+        }
+        let mut args: Vec<&str> = options.args;
+        let key_dir = format!("{}/kms-keys", rustfs.temp_dir);
+        if options.local_kms {
+            tokio::fs::create_dir_all(&key_dir).await?;
+            crate::kms::common::create_key_with_specific_id(&key_dir, LOCAL_KMS_DEFAULT_KEY_ID).await?;
+            args.extend_from_slice(&[
+                "--kms-enable",
+                "--kms-backend",
+                "local",
+                "--kms-key-dir",
+                &key_dir,
+                "--kms-default-key-id",
+                LOCAL_KMS_DEFAULT_KEY_ID,
+            ]);
+            env.push(("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true"));
+        }
+        rustfs.start_rustfs_server_with_env(args, &env).await?;
         let client = rustfs.create_s3_client();
         Ok(Self { rustfs, source, client })
     }
@@ -461,13 +511,15 @@ impl OdmTestEnv {
     /// Waits until the runtime consults the source for `bucket`: a config
     /// install is applied asynchronously after the admin call returns. The
     /// probe is a HEAD on a key that exists nowhere, so nothing is pulled and
-    /// only that key enters the negative cache.
+    /// only that key enters the negative cache. The probe key is per bucket,
+    /// so a second bucket, or a reinstalled configuration, waits for its own
+    /// state instead of observing an earlier journal entry.
     pub async fn wait_until_source_consulted(&self, bucket: &str) -> Result<(), BoxError> {
-        const PROBE_KEY: &str = "_odm-readiness-probe";
+        let probe_key = format!("_odm-readiness-probe-{bucket}-{}", uuid::Uuid::new_v4());
         let deadline = Instant::now() + Duration::from_secs(30);
         loop {
-            let _ = self.client.head_object().bucket(bucket).key(PROBE_KEY).send().await;
-            if self.source.count_requests(Operation::HeadObject, PROBE_KEY) > 0 {
+            let _ = self.client.head_object().bucket(bucket).key(&probe_key).send().await;
+            if self.source.count_requests(Operation::HeadObject, &probe_key) > 0 {
                 return Ok(());
             }
             if Instant::now() >= deadline {
@@ -477,6 +529,124 @@ impl OdmTestEnv {
         }
     }
 
+    /// Creates `bucket` unless it already exists, installs `spec` on it and
+    /// returns once the runtime consults the source. Scenarios with a second
+    /// bucket, a bucket created with non-default options, or a reinstalled
+    /// configuration all go through this.
+    pub async fn configure_and_wait(&self, bucket: &str, spec: &OdmSourceSpec) -> Result<(), BoxError> {
+        if self.client.head_bucket().bucket(bucket).send().await.is_err() {
+            self.rustfs.create_test_bucket(bucket).await?;
+        }
+        let response = self.configure_source(bucket, spec).await?;
+        if response.status != 200 {
+            return Err(format!("configure on-demand migration for {bucket}: {} {}", response.status, response.body).into());
+        }
+        self.wait_until_source_consulted(bucket).await
+    }
+
+    /// Raw signed request against the RustFS under test with additional
+    /// request headers. `Range`, `If-None-Match` and the anti-loop marker are
+    /// not part of the SigV4 signed-header set, so they are attached after
+    /// signing exactly as a real client's would be.
+    pub async fn raw_object_request(
+        &self,
+        method: http::Method,
+        bucket: &str,
+        key: &str,
+        headers: &[(&str, &str)],
+    ) -> Result<RawResponse, BoxError> {
+        let url = format!("{}/{bucket}/{key}", self.rustfs.url);
+        let uri = url.parse::<http::Uri>()?;
+        let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
+        let request = http::Request::builder()
+            .method(method.clone())
+            .uri(uri)
+            .header(http::header::HOST, authority)
+            .header("x-amz-content-sha256", rustfs_signer::constants::UNSIGNED_PAYLOAD)
+            .body(s3s::Body::empty())?;
+        let signed = rustfs_signer::sign_v4(request, 0, &self.rustfs.access_key, &self.rustfs.secret_key, "", "us-east-1");
+
+        let mut builder =
+            crate::common::local_http_client().request(reqwest::Method::from_bytes(method.as_str().as_bytes())?, &url);
+        for (name, value) in signed.headers() {
+            builder = builder.header(name, value);
+        }
+        for (name, value) in headers {
+            builder = builder.header(*name, *value);
+        }
+        let response = builder.send().await?;
+        Ok(RawResponse {
+            status: response.status().as_u16(),
+            headers: response.headers().clone(),
+            body: response.bytes().await?,
+        })
+    }
+
+    /// Parsed `GET .../{bucket}/status` body. Fails when the route does not
+    /// answer 200 so a scenario never asserts against an error document.
+    pub async fn status_json(&self, bucket: &str) -> Result<serde_json::Value, BoxError> {
+        let response = self.status(bucket).await?;
+        if response.status != 200 {
+            return Err(format!("status for {bucket}: {} {}", response.status, response.body).into());
+        }
+        response.json()
+    }
+
+    /// Reads one counter out of the status document by JSON pointer, e.g.
+    /// `/counters/pull_failures_total/queue_full`. Missing runtime state
+    /// reads as 0, which is what an operator sees too.
+    pub async fn status_counter(&self, bucket: &str, pointer: &str) -> Result<u64, BoxError> {
+        Ok(self
+            .status_json(bucket)
+            .await?
+            .pointer(pointer)
+            .and_then(serde_json::Value::as_u64)
+            .unwrap_or(0))
+    }
+
+    /// Polls [`Self::status_counter`] until it reaches `at_least`. Background
+    /// pulls and their failure counters land after the response that started
+    /// them, so every counter assertion about them has to wait.
+    pub async fn wait_for_status_counter(
+        &self,
+        bucket: &str,
+        pointer: &str,
+        at_least: u64,
+        timeout: Duration,
+    ) -> Result<u64, BoxError> {
+        let deadline = Instant::now() + timeout;
+        loop {
+            let value = self.status_counter(bucket, pointer).await?;
+            if value >= at_least {
+                return Ok(value);
+            }
+            if Instant::now() >= deadline {
+                return Err(format!("{pointer} for {bucket} stalled at {value}, expected at least {at_least}").into());
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    }
+
+    /// Highest `inflight_pulls` the status route reported while `work` ran,
+    /// sampled every 5 ms. The concurrency ceiling is only observable from
+    /// outside while pulls are in flight.
+    pub async fn peak_inflight_pulls<F, T>(&self, bucket: &str, work: F) -> Result<(T, u64), BoxError>
+    where
+        F: std::future::Future<Output = T>,
+    {
+        let mut peak = 0;
+        tokio::pin!(work);
+        let output = loop {
+            tokio::select! {
+                output = &mut work => break output,
+                _ = tokio::time::sleep(Duration::from_millis(5)) => {
+                    peak = peak.max(self.status_counter(bucket, "/inflight_pulls").await?);
+                }
+            }
+        };
+        Ok((output, peak))
+    }
+
     /// Panics if `key` is listed locally.
     pub async fn assert_local_absent(&self, bucket: &str, key: &str) {
         assert!(
@@ -515,3 +685,41 @@ pub async fn start_source_rustfs() -> Result<RustFSTestEnvironment, BoxError> {
     source.start_rustfs_server_without_cleanup(vec![]).await?;
     Ok(source)
 }
+
+/// Like [`start_source_rustfs`], but with on-demand migration enabled on the
+/// second server too, so it can be given a source of its own (the anti-loop
+/// scenario chains two migrating servers).
+pub async fn start_source_rustfs_with_odm() -> Result<RustFSTestEnvironment, BoxError> {
+    let mut source = RustFSTestEnvironment::new().await?;
+    source
+        .start_rustfs_server_without_cleanup_with_env(&[(ODM_MODULE_SWITCH_ENV, "true"), (ALLOW_LOOPBACK_SOURCE_ENV, "true")])
+        .await?;
+    Ok(source)
+}
+
+/// The full scenario fixture every behavior test starts from: a RustFS with
+/// `bucket`, an unversioned `source_bucket` on the fake source (a plain
+/// migration source), the configuration installed after `adjust` tweaked it,
+/// and the runtime proven to consult the source.
+pub async fn start_configured_env(
+    bucket: &str,
+    source_bucket: &str,
+    adjust: impl FnOnce(&mut OdmSourceSpec),
+) -> Result<OdmTestEnv, BoxError> {
+    start_configured_env_with(OdmEnvOptions::default(), bucket, source_bucket, adjust).await
+}
+
+/// [`start_configured_env`] with extra process arguments and environment.
+pub async fn start_configured_env_with(
+    options: OdmEnvOptions<'_>,
+    bucket: &str,
+    source_bucket: &str,
+    adjust: impl FnOnce(&mut OdmSourceSpec),
+) -> Result<OdmTestEnv, BoxError> {
+    let env = OdmTestEnv::start_with(options).await?;
+    env.source.create_bucket_with_mode(source_bucket, BucketMode::Unversioned);
+    let mut spec = env.fake_source_spec(source_bucket);
+    adjust(&mut spec);
+    env.configure_and_wait(bucket, &spec).await?;
+    Ok(env)
+}

crates/e2e_test/src/on_demand_migration/concurrency_test.rs

@@ -0,0 +1,183 @@
+// 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.
+
+//! Concurrency limits of on-demand migration (rustfs/backlog#2158):
+//! single-flight on one key, the `max_concurrent_pulls` ceiling, and a full
+//! background pull queue.
+//!
+//! The point of each case is what the source is spared, so the source
+//! journal (`count_requests`) carries the assertion in every one of them.
+
+use super::common::{BoxError, OdmTestEnv, RawResponse, SeedObject, start_configured_env};
+use crate::fake_s3_target::Operation;
+use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+use bytes::Bytes;
+use std::time::Duration;
+
+type TestResult = Result<(), BoxError>;
+
+const SOURCE_BUCKET: &str = "odm-concurrency-source";
+/// Background pulls land after the response that queued them.
+const SETTLE: Duration = Duration::from_secs(120);
+
+fn payload(len: usize) -> Bytes {
+    (0..len).map(|index| (index % 251) as u8).collect::<Vec<u8>>().into()
+}
+
+fn source_get_count(env: &OdmTestEnv, key: &str) -> usize {
+    env.source.count_requests(Operation::GetObject, key)
+}
+
+/// Case 9: 32 concurrent misses on one key coalesce into a single-flight
+/// pull. At most two source GETs are allowed: the leader plus one follower
+/// that gave up waiting and streamed through.
+#[tokio::test]
+async fn test_odm_concurrent_misses_on_one_key_coalesce() -> TestResult {
+    let bucket = "odm-concurrency-singleflight";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |_| {}).await?;
+    env.client
+        .put_bucket_versioning()
+        .bucket(bucket)
+        .versioning_configuration(
+            VersioningConfiguration::builder()
+                .status(BucketVersioningStatus::Enabled)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    let key = "singleflight/asset.bin";
+    let body = payload(512 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    let responses: Vec<RawResponse> = futures::future::try_join_all((0..32).map(|_| env.raw_get(bucket, key))).await?;
+    for (index, response) in responses.iter().enumerate() {
+        assert_eq!(response.status, 200, "reader {index}: {}", String::from_utf8_lossy(&response.body));
+        assert_eq!(response.body, body, "reader {index} received different bytes");
+    }
+
+    let source_gets = source_get_count(&env, key);
+    assert!(
+        (1..=2).contains(&source_gets),
+        "32 concurrent misses must not become {source_gets} source GETs"
+    );
+
+    assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "the leader stores the object");
+    env.assert_local_present(bucket, key, &body).await;
+    let versions = env.client.list_object_versions().bucket(bucket).prefix(key).send().await?;
+    assert_eq!(
+        versions.versions().len(),
+        1,
+        "the coalesced pull commits exactly one version: {:?}",
+        versions.versions()
+    );
+    assert_eq!(
+        source_get_count(&env, key),
+        source_gets,
+        "nothing pulls the object again once it is local"
+    );
+    Ok(())
+}
+
+/// Case 10: 64 misses on distinct keys never exceed `max_concurrent_pulls`
+/// in flight, and all of them eventually land.
+#[tokio::test]
+async fn test_odm_concurrent_pulls_respect_the_configured_ceiling() -> TestResult {
+    let bucket = "odm-concurrency-ceiling";
+    const MAX_CONCURRENT_PULLS: u32 = 4;
+    const KEYS: usize = 64;
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |spec| {
+        spec.policy.max_concurrent_pulls = MAX_CONCURRENT_PULLS;
+    })
+    .await?;
+
+    let body = payload(256 * 1024);
+    let keys: Vec<String> = (0..KEYS).map(|index| format!("ceiling/object-{index:03}.bin")).collect();
+    let seeds: Vec<SeedObject> = keys.iter().map(|key| SeedObject::new(key.clone(), body.clone())).collect();
+    env.seed_source(SOURCE_BUCKET, &seeds);
+
+    let reads = futures::future::try_join_all(keys.iter().map(|key| env.raw_get(bucket, key)));
+    let (responses, peak_inflight) = env.peak_inflight_pulls(bucket, reads).await?;
+    let responses = responses?;
+    for (key, response) in keys.iter().zip(&responses) {
+        assert_eq!(response.status, 200, "{key}: {}", String::from_utf8_lossy(&response.body));
+        assert_eq!(response.body, body, "{key} received different bytes");
+    }
+    assert!(
+        peak_inflight <= u64::from(MAX_CONCURRENT_PULLS),
+        "in-flight pulls peaked at {peak_inflight}, above the configured {MAX_CONCURRENT_PULLS}"
+    );
+    assert!(
+        peak_inflight >= 1,
+        "the poll never observed a pull in flight, so the ceiling assertion proves nothing"
+    );
+
+    for key in &keys {
+        assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "{key} must be stored locally");
+        assert_eq!(source_get_count(&env, key), 1, "{key} is pulled exactly once");
+    }
+    assert_eq!(env.status_counter(bucket, "/inflight_pulls").await?, 0, "every pull slot is released");
+    Ok(())
+}
+
+/// Case 11: with a small background queue, a burst of Range reads overflows
+/// it. The overflow is counted and dropped, never turned into a client
+/// failure: every reader still gets its 206 from the source.
+#[tokio::test]
+async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients() -> TestResult {
+    let bucket = "odm-concurrency-queue-full";
+    const REQUESTS: usize = 100;
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |spec| {
+        spec.policy.pull_queue_capacity = 8;
+        spec.policy.max_concurrent_pulls = 1;
+    })
+    .await?;
+
+    let body = payload(128 * 1024);
+    let keys: Vec<String> = (0..REQUESTS).map(|index| format!("queue/object-{index:03}.bin")).collect();
+    let seeds: Vec<SeedObject> = keys.iter().map(|key| SeedObject::new(key.clone(), body.clone())).collect();
+    env.seed_source(SOURCE_BUCKET, &seeds);
+
+    let responses: Vec<RawResponse> = futures::future::try_join_all(
+        keys.iter()
+            .map(|key| env.raw_object_request(http::Method::GET, bucket, key, &[("range", "bytes=0-1023")])),
+    )
+    .await?;
+    for (key, response) in keys.iter().zip(&responses) {
+        assert_eq!(response.status, 206, "{key}: {}", String::from_utf8_lossy(&response.body));
+        assert_eq!(response.body, body.slice(0..1024), "{key} served the wrong range");
+        assert_eq!(
+            response.header("content-range"),
+            Some(format!("bytes 0-1023/{}", body.len()).as_str()),
+            "{key}"
+        );
+    }
+
+    let queue_full = env
+        .wait_for_status_counter(bucket, "/counters/pull_failures_total/queue_full", 1, SETTLE)
+        .await?;
+    assert!(queue_full > 0, "a 100-deep burst must overflow an 8-slot queue");
+
+    let ranged_reads: usize = keys.iter().map(|key| source_get_count(&env, key)).sum();
+    assert!(
+        ranged_reads >= REQUESTS,
+        "every reader is served from the source: {ranged_reads} GETs for {REQUESTS} readers"
+    );
+    let dropped = keys.iter().filter(|key| source_get_count(&env, key) == 1).count();
+    assert!(
+        dropped > 0,
+        "the overflowed keys are the ones with no backfill GET, but every key got one"
+    );
+    Ok(())
+}

crates/e2e_test/src/on_demand_migration/fault_test.rs

@@ -0,0 +1,484 @@
+// 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.
+
+//! Source-failure scenarios for on-demand migration (rustfs/backlog#2158):
+//! access denied, the circuit breaker, first-byte and mid-body stream
+//! failures, ETag integrity, the negative cache, and an unsupported
+//! (SSE-C) source object.
+//!
+//! Every case asserts what the source was asked for, not only what the
+//! client received: a fault that silently turned into a second source
+//! request would otherwise pass.
+
+use super::common::{BoxError, OdmTestEnv, SeedObject, start_configured_env};
+use crate::fake_s3_target::{FaultAction, Operation};
+use bytes::Bytes;
+use std::time::{Duration, Instant};
+
+type TestResult = Result<(), BoxError>;
+
+const SOURCE_BUCKET: &str = "odm-fault-source";
+/// Header the GET/HEAD paths add when the answer came from the source.
+const ODM_RESPONSE_HEADER: &str = "x-rustfs-on-demand-migration";
+/// Status of the `SourceUnavailable` error the `propagate` policy returns.
+const SOURCE_UNAVAILABLE_STATUS: u16 = 424;
+/// Background pulls and their counters land after the response.
+const SETTLE: Duration = Duration::from_secs(60);
+/// Consecutive counted source failures that open the breaker
+/// (`BREAKER_FAILURE_THRESHOLD` in ecstore).
+const BREAKER_FAILURE_THRESHOLD: usize = 5;
+
+/// Position-dependent payload so a misaligned or truncated copy is caught.
+fn payload(len: usize) -> Bytes {
+    (0..len).map(|index| (index % 251) as u8).collect::<Vec<u8>>().into()
+}
+
+/// A source object with a well-formed but deliberately wrong single-part
+/// ETag: the fake source retains `x-rustfs-source-etag` verbatim, so HEAD
+/// and GET advertise an MD5 the body does not have.
+async fn seed_with_etag(env: &OdmTestEnv, key: &str, body: Bytes, etag: &str) -> TestResult {
+    let response = env
+        .source_client()
+        .put_object()
+        .bucket(SOURCE_BUCKET)
+        .key(key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(body))
+        .customize()
+        .mutate_request({
+            let etag = etag.to_string();
+            move |request| {
+                request.headers_mut().insert("x-rustfs-source-etag", etag.clone());
+            }
+        })
+        .send()
+        .await?;
+    assert_eq!(
+        response.e_tag(),
+        Some(format!("\"{etag}\"").as_str()),
+        "the fake source stores the announced ETag"
+    );
+    Ok(())
+}
+
+/// A source object that reports SSE-C: the fake source echoes the customer
+/// algorithm it captured from the replication passthrough transport header.
+async fn seed_with_ssec(env: &OdmTestEnv, key: &str, body: Bytes) -> TestResult {
+    env.source_client()
+        .put_object()
+        .bucket(SOURCE_BUCKET)
+        .key(key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(body))
+        .customize()
+        .mutate_request(|request| {
+            request.headers_mut().insert("x-rustfs-replication-ssec-algorithm", "AES256");
+        })
+        .send()
+        .await?;
+    Ok(())
+}
+
+/// Case 1: a 403 from the source is a configuration error, not a health
+/// signal. `propagate` answers 424 and records the class; `not_found` hides
+/// it as a 404. Neither counts toward the breaker.
+#[tokio::test]
+async fn test_odm_source_access_denied_propagates_without_opening_the_breaker() -> TestResult {
+    let propagating = "odm-fault-denied-propagate";
+    let hiding = "odm-fault-denied-notfound";
+    let env = start_configured_env(propagating, SOURCE_BUCKET, |_| {}).await?;
+    let mut hiding_spec = env.fake_source_spec(SOURCE_BUCKET);
+    hiding_spec.policy.source_error = "not_found".to_string();
+    env.configure_and_wait(hiding, &hiding_spec).await?;
+
+    let propagate_key = "denied/propagate.bin";
+    let hidden_key = "denied/hidden.bin";
+    env.seed_source(
+        SOURCE_BUCKET,
+        &[
+            SeedObject::new(propagate_key, payload(4096)),
+            SeedObject::new(hidden_key, payload(4096)),
+        ],
+    );
+
+    env.source
+        .inject_for_key(Operation::HeadObject, propagate_key, FaultAction::ResponseStatus(403), 1);
+    let denied = env.raw_get(propagating, propagate_key).await?;
+    assert_eq!(denied.status, SOURCE_UNAVAILABLE_STATUS, "{}", String::from_utf8_lossy(&denied.body));
+    assert!(
+        String::from_utf8_lossy(&denied.body).contains("SourceUnavailable"),
+        "the propagated error names the ODM source code: {}",
+        String::from_utf8_lossy(&denied.body)
+    );
+    assert_eq!(env.source.count_requests(Operation::HeadObject, propagate_key), 1);
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, propagate_key),
+        0,
+        "a denied HEAD never reaches the body"
+    );
+
+    let status = env.status_json(propagating).await?;
+    assert_eq!(
+        status.pointer("/last_source_error/class").and_then(|v| v.as_str()),
+        Some("access_denied"),
+        "{status}"
+    );
+    assert_eq!(
+        status.pointer("/breaker/state").and_then(|v| v.as_str()),
+        Some("closed"),
+        "a configuration error must not open the breaker: {status}"
+    );
+    assert_eq!(
+        status
+            .pointer("/counters/requests_total/get/source_error")
+            .and_then(|v| v.as_u64()),
+        Some(1),
+        "{status}"
+    );
+
+    env.source
+        .inject_for_key(Operation::HeadObject, hidden_key, FaultAction::ResponseStatus(403), 1);
+    let hidden = env.raw_get(hiding, hidden_key).await?;
+    assert_eq!(hidden.status, 404, "{}", String::from_utf8_lossy(&hidden.body));
+    assert_eq!(env.source.count_requests(Operation::HeadObject, hidden_key), 1);
+    assert_eq!(env.source.count_requests(Operation::GetObject, hidden_key), 0);
+
+    env.assert_local_absent(propagating, propagate_key).await;
+    env.assert_local_absent(hiding, hidden_key).await;
+    Ok(())
+}
+
+/// Case 2: repeated transport failures open the breaker; while it is open
+/// the source is not touched at all, and the half-open probe after the open
+/// window closes it again. The open window is a compiled-in 30 s constant
+/// (`BREAKER_OPEN_DURATION`), so this case waits in real time.
+///
+/// One ODM source call is several wire requests: the SDK retries a 503 on
+/// its own, and only the exhausted call counts as one breaker failure. The
+/// script is therefore deep enough to cover every retry, and the open state
+/// is waited for instead of being predicted from a request count.
+#[tokio::test]
+async fn test_odm_repeated_source_errors_open_the_breaker_and_recover() -> TestResult {
+    let bucket = "odm-fault-breaker";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |_| {}).await?;
+    let key = "breaker/doc.bin";
+    let body = payload(8192);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    env.source
+        .inject_for_key(Operation::HeadObject, key, FaultAction::ResponseStatus(503), 200);
+    let mut opened = false;
+    for attempt in 1..=BREAKER_FAILURE_THRESHOLD * 2 {
+        let response = env.raw_get(bucket, key).await?;
+        assert_eq!(
+            response.status,
+            SOURCE_UNAVAILABLE_STATUS,
+            "attempt {attempt}: {}",
+            String::from_utf8_lossy(&response.body)
+        );
+        if env
+            .status_json(bucket)
+            .await?
+            .pointer("/breaker/state")
+            .and_then(|v| v.as_str())
+            == Some("open")
+        {
+            opened = true;
+            break;
+        }
+    }
+    assert!(opened, "consecutive source failures must open the breaker");
+    assert!(
+        env.source.count_requests(Operation::HeadObject, key) >= BREAKER_FAILURE_THRESHOLD,
+        "each counted failure is at least one source request"
+    );
+
+    // With the script cleared, the only thing that can still fail a read is
+    // the open breaker itself.
+    env.source.clear_faults();
+    let source_requests = env.source.count_requests(Operation::HeadObject, key);
+    let rejected = env.raw_get(bucket, key).await?;
+    assert_eq!(rejected.status, SOURCE_UNAVAILABLE_STATUS, "{}", String::from_utf8_lossy(&rejected.body));
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, key),
+        source_requests,
+        "an open breaker never touches the source"
+    );
+    assert!(
+        env.status_counter(bucket, "/counters/requests_total/get/breaker_open")
+            .await?
+            >= 1,
+        "the rejected request is counted as breaker_open"
+    );
+
+    // Half-open admits exactly one probe once the open window elapses.
+    let deadline = Instant::now() + Duration::from_secs(120);
+    let recovered = loop {
+        let response = env.raw_get(bucket, key).await?;
+        if response.status == 200 {
+            break response;
+        }
+        assert_eq!(response.status, SOURCE_UNAVAILABLE_STATUS);
+        assert!(Instant::now() < deadline, "the breaker never left the open state");
+        tokio::time::sleep(Duration::from_secs(1)).await;
+    };
+    assert_eq!(recovered.body, body, "the recovered read serves the source bytes");
+    assert_eq!(recovered.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, key),
+        source_requests + 1,
+        "only the half-open probe reached the source"
+    );
+    assert_eq!(env.source.count_requests(Operation::GetObject, key), 1);
+    assert_eq!(
+        env.status_json(bucket)
+            .await?
+            .pointer("/breaker/state")
+            .and_then(|v| v.as_str()),
+        Some("closed"),
+        "a successful probe closes the breaker"
+    );
+    Ok(())
+}
+
+/// Case 3: a source that holds the response past `first_byte_ms` is a
+/// timeout, and the client never sees a 200 head. Every attempt the SDK
+/// makes on its own is stalled too, so the ODM call really does give up.
+#[tokio::test]
+async fn test_odm_source_stall_times_out_before_the_first_byte() -> TestResult {
+    let bucket = "odm-fault-stall";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |spec| {
+        spec.policy.source_timeout.first_byte_ms = 500;
+    })
+    .await?;
+    let key = "stall/doc.bin";
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, payload(4096))]);
+
+    env.source
+        .inject_for_key(Operation::HeadObject, key, FaultAction::Stall(Duration::from_secs(5)), 8);
+    let started = Instant::now();
+    let response = env.raw_get(bucket, key).await?;
+    let elapsed = started.elapsed();
+    assert_eq!(response.status, SOURCE_UNAVAILABLE_STATUS, "{}", String::from_utf8_lossy(&response.body));
+    assert!(
+        elapsed < Duration::from_secs(30),
+        "the read timeout must cut every attempt short, took {elapsed:?}"
+    );
+    let attempts = env.source.count_requests(Operation::HeadObject, key);
+    assert!(attempts >= 1, "the stalled HEAD is the only source request");
+    assert!(
+        elapsed < Duration::from_secs(5) * u32::try_from(attempts).unwrap_or(1),
+        "no attempt waited the stall out ({attempts} attempts in {elapsed:?})"
+    );
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        0,
+        "a timed-out HEAD never starts a body read"
+    );
+    assert_eq!(
+        env.status_json(bucket)
+            .await?
+            .pointer("/last_source_error/class")
+            .and_then(|v| v.as_str()),
+        Some("timeout"),
+    );
+    env.assert_local_absent(bucket, key).await;
+    Ok(())
+}
+
+/// Case 4: the source cuts the body of an inline pull. The client sees a
+/// short read, nothing is stored, and no multipart upload is left behind.
+#[tokio::test]
+async fn test_odm_inline_pull_aborts_when_the_source_body_is_cut() -> TestResult {
+    let bucket = "odm-fault-inline-cut";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |_| {}).await?;
+    let key = "cut/inline.bin";
+    let body = payload(256 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    env.source
+        .inject_for_key(Operation::GetObject, key, FaultAction::TruncateBodyAt(1024), 1);
+    // The client sees a transport failure while reading the body: the
+    // announced Content-Length is never delivered.
+    env.raw_get(bucket, key)
+        .await
+        .expect_err("a cut source body must not read back as a complete object");
+
+    assert_eq!(env.source.count_requests(Operation::HeadObject, key), 1);
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        1,
+        "an aborted inline pull is not retried on the same request"
+    );
+    // Give a stray background pull time to appear before asserting absence.
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    env.assert_local_absent(bucket, key).await;
+    let uploads = env.client.list_multipart_uploads().bucket(bucket).send().await?;
+    assert!(
+        uploads.uploads().is_empty(),
+        "an aborted pull leaves no multipart upload: {:?}",
+        uploads.uploads()
+    );
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        1,
+        "nothing re-reads the source afterwards"
+    );
+    Ok(())
+}
+
+/// Case 5: the background pull of a large object hits a cut body, counts the
+/// failure, and the retry stores the object.
+#[tokio::test]
+async fn test_odm_background_pull_retries_a_truncated_source_body() -> TestResult {
+    let bucket = "odm-fault-background-cut";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |spec| spec.policy.inline_max_bytes = 4096).await?;
+    let key = "cut/background.bin";
+    let body = payload(512 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    // The faults are consumed in order by the two GETs the large-object path
+    // makes: the passthrough that answers the client (unaffected), then the
+    // background pull (cut).
+    env.source
+        .inject_for_key(Operation::GetObject, key, FaultAction::Delay(Duration::ZERO), 1);
+    env.source
+        .inject_for_key(Operation::GetObject, key, FaultAction::TruncateBodyAt(2048), 1);
+
+    let response = env.raw_get(bucket, key).await?;
+    assert_eq!(response.status, 200, "{}", String::from_utf8_lossy(&response.body));
+    assert_eq!(response.body, body, "the passthrough is unaffected by the pull's fault");
+
+    // The cut body ends the pull attempt as a retryable source transport
+    // failure; the retry stores the object, so the pull as a whole succeeds
+    // and no failure is counted (only a pull that gives up is).
+    assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "the retry must store the object");
+    env.assert_local_present(bucket, key, &body).await;
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        3,
+        "one passthrough, one cut pull, one successful retry"
+    );
+    let status = env.status_json(bucket).await?;
+    assert_eq!(
+        status
+            .pointer("/counters/pulled_objects_total/background")
+            .and_then(|v| v.as_u64()),
+        Some(1),
+        "{status}"
+    );
+    assert_eq!(
+        status
+            .pointer("/counters/pull_failures_total")
+            .and_then(|failures| failures.as_object())
+            .map(|failures| failures.values().filter_map(serde_json::Value::as_u64).sum::<u64>()),
+        Some(0),
+        "a retried attempt is not a failed pull: {status}"
+    );
+    Ok(())
+}
+
+/// Case 6: the source advertises an ETag its bytes do not match. The client
+/// still gets every byte; the write-back is discarded as an integrity
+/// failure and nothing is stored.
+#[tokio::test]
+async fn test_odm_wrong_source_etag_discards_the_write_back() -> TestResult {
+    let bucket = "odm-fault-etag";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |_| {}).await?;
+    let key = "etag/mismatch.bin";
+    let body = payload(64 * 1024);
+    seed_with_etag(&env, key, body.clone(), "0123456789abcdef0123456789abcdef").await?;
+
+    let response = env.raw_get(bucket, key).await?;
+    assert_eq!(response.status, 200, "{}", String::from_utf8_lossy(&response.body));
+    assert_eq!(response.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(response.body, body, "the client receives the complete source bytes");
+
+    env.wait_for_status_counter(bucket, "/counters/pull_failures_total/etag_mismatch", 1, SETTLE)
+        .await?;
+    env.assert_local_absent(bucket, key).await;
+    assert_eq!(env.source.count_requests(Operation::HeadObject, key), 1);
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        1,
+        "a discarded write-back is not re-read"
+    );
+    Ok(())
+}
+
+/// Case 7: a source miss is remembered for `negative_cache_ttl_secs`, and
+/// re-checked once the entry expires.
+#[tokio::test]
+async fn test_odm_source_not_found_is_negative_cached_for_the_ttl() -> TestResult {
+    let bucket = "odm-fault-negative-cache";
+    let ttl = Duration::from_secs(3);
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |spec| {
+        spec.policy.negative_cache_ttl_secs = ttl.as_secs();
+    })
+    .await?;
+    let key = "negative/nowhere.bin";
+
+    for attempt in 1..=10 {
+        let response = env.raw_get(bucket, key).await?;
+        assert_eq!(response.status, 404, "attempt {attempt}: {}", String::from_utf8_lossy(&response.body));
+    }
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, key),
+        1,
+        "nine of the ten misses stop at the negative cache"
+    );
+    assert!(
+        env.status_counter(bucket, "/counters/requests_total/get/negative_cached")
+            .await?
+            >= 9,
+        "the cached misses are counted"
+    );
+
+    tokio::time::sleep(ttl + Duration::from_secs(2)).await;
+    let response = env.raw_get(bucket, key).await?;
+    assert_eq!(response.status, 404);
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, key),
+        2,
+        "an expired entry re-checks the source once"
+    );
+    assert_eq!(env.source.count_requests(Operation::GetObject, key), 0);
+    Ok(())
+}
+
+/// Case 8: an SSE-C source object cannot be migrated (the key belongs to the
+/// source's client), so the read fails as unsupported without a body read.
+#[tokio::test]
+async fn test_odm_ssec_source_object_is_unsupported() -> TestResult {
+    let bucket = "odm-fault-ssec";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |_| {}).await?;
+    let key = "ssec/secret.bin";
+    seed_with_ssec(&env, key, payload(4096)).await?;
+
+    let response = env.raw_get(bucket, key).await?;
+    assert_eq!(response.status, SOURCE_UNAVAILABLE_STATUS, "{}", String::from_utf8_lossy(&response.body));
+    assert_eq!(env.source.count_requests(Operation::HeadObject, key), 1);
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        0,
+        "an unsupported object is rejected on the HEAD"
+    );
+    assert_eq!(
+        env.status_json(bucket)
+            .await?
+            .pointer("/counters/requests_total/get/unsupported")
+            .and_then(|v| v.as_u64()),
+        Some(1),
+    );
+    env.assert_local_absent(bucket, key).await;
+    Ok(())
+}

crates/e2e_test/src/on_demand_migration/interaction_test.rs

@@ -0,0 +1,825 @@
+// 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.
+
+//! How on-demand migration composes with the rest of the bucket surface
+//! (rustfs/backlog#2158): default encryption, Object Lock, quota,
+//! notifications, replication, versioning and delete markers, the disable
+//! switch, and the admin view.
+//!
+//! A pulled object goes through the internal put path, so it must be
+//! indistinguishable from a client PUT. Each case pins both the resulting
+//! local object and what the source was asked for.
+
+use super::common::{
+    AdminResponse, BoxError, OdmEnvOptions, OdmSourceSpec, OdmTestEnv, SeedObject, start_configured_env,
+    start_configured_env_with,
+};
+use crate::common::{RustFSTestEnvironment, replication_fast_env, signed_request};
+use crate::fake_s3_target::{BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, Operation};
+use crate::object_lock::common::put_object_lock_configuration;
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use aws_sdk_s3::types::{
+    BucketVersioningStatus, Event, FilterRule, FilterRuleName, NotificationConfiguration, NotificationConfigurationFilter,
+    ObjectLockRetentionMode, QueueConfiguration, S3KeyFilter, ServerSideEncryption, ServerSideEncryptionByDefault,
+    ServerSideEncryptionConfiguration, ServerSideEncryptionRule, VersioningConfiguration,
+};
+use bytes::Bytes;
+use local_ip_address::local_ip;
+use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
+use serde_json::Value;
+use std::time::{Duration, Instant};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+use tokio::sync::mpsc;
+
+type TestResult = Result<(), BoxError>;
+
+const SOURCE_BUCKET: &str = "odm-interaction-source";
+const ODM_RESPONSE_HEADER: &str = "x-rustfs-on-demand-migration";
+/// `userIdentity.principalId` every write-back event carries.
+const ODM_PRINCIPAL_ID: &str = "rustfs-on-demand-migration";
+const SETTLE: Duration = Duration::from_secs(120);
+
+fn payload(len: usize) -> Bytes {
+    (0..len).map(|index| (index % 251) as u8).collect::<Vec<u8>>().into()
+}
+
+async fn admin(
+    env: &RustFSTestEnvironment,
+    method: http::Method,
+    path: &str,
+    body: Option<Value>,
+) -> Result<AdminResponse, BoxError> {
+    let url = format!("{}{path}", env.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, &env.access_key, &env.secret_key, body, content_type).await?;
+    Ok(AdminResponse {
+        status: response.status().as_u16(),
+        body: response.text().await?,
+    })
+}
+
+async fn enable_versioning(env: &OdmTestEnv, bucket: &str) -> TestResult {
+    env.client
+        .put_bucket_versioning()
+        .bucket(bucket)
+        .versioning_configuration(
+            VersioningConfiguration::builder()
+                .status(BucketVersioningStatus::Enabled)
+                .build(),
+        )
+        .send()
+        .await?;
+    Ok(())
+}
+
+/// Case 12: a bucket that encrypts by default stores the pulled object
+/// encrypted, and it reads back as plaintext afterwards without touching the
+/// source again.
+#[tokio::test]
+async fn test_odm_pulled_object_uses_bucket_default_encryption() -> TestResult {
+    let bucket = "odm-interaction-sse";
+    let env = start_configured_env_with(
+        OdmEnvOptions {
+            local_kms: true,
+            ..OdmEnvOptions::default()
+        },
+        bucket,
+        SOURCE_BUCKET,
+        |_| {},
+    )
+    .await?;
+    env.client
+        .put_bucket_encryption()
+        .bucket(bucket)
+        .server_side_encryption_configuration(
+            ServerSideEncryptionConfiguration::builder()
+                .rules(
+                    ServerSideEncryptionRule::builder()
+                        .apply_server_side_encryption_by_default(
+                            ServerSideEncryptionByDefault::builder()
+                                .sse_algorithm(ServerSideEncryption::Aes256)
+                                .build()?,
+                        )
+                        .build(),
+                )
+                .build()?,
+        )
+        .send()
+        .await?;
+
+    let key = "sse/report.bin";
+    let body = payload(128 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    let first = env.raw_get(bucket, key).await?;
+    assert_eq!(first.status, 200, "{}", String::from_utf8_lossy(&first.body));
+    assert_eq!(first.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(first.body, body);
+    assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "the pull must store the object");
+
+    let second = env.raw_get(bucket, key).await?;
+    assert_eq!(second.status, 200, "{}", String::from_utf8_lossy(&second.body));
+    assert_eq!(second.header(ODM_RESPONSE_HEADER), None, "the second read is local");
+    assert_eq!(
+        second.header("x-amz-server-side-encryption"),
+        Some("AES256"),
+        "the write-back honours the bucket default encryption"
+    );
+    assert_eq!(second.body, body, "the encrypted copy reads back as the source bytes");
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        1,
+        "the encrypted local copy serves the second read"
+    );
+    Ok(())
+}
+
+/// Case 13: a pulled object inherits the bucket's default Object Lock
+/// retention, so it cannot be deleted while the retention holds.
+#[tokio::test]
+async fn test_odm_pulled_object_inherits_object_lock_retention() -> TestResult {
+    let bucket = "odm-interaction-object-lock";
+    let env = OdmTestEnv::start().await?;
+    env.source.create_bucket_with_mode(SOURCE_BUCKET, BucketMode::Unversioned);
+    env.client
+        .create_bucket()
+        .bucket(bucket)
+        .object_lock_enabled_for_bucket(true)
+        .send()
+        .await?;
+    put_object_lock_configuration(&env.client, bucket, ObjectLockRetentionMode::Compliance, Some(1), None).await?;
+    let spec = env.fake_source_spec(SOURCE_BUCKET);
+    env.configure_and_wait(bucket, &spec).await?;
+
+    let key = "locked/record.bin";
+    let body = payload(32 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    let pulled = env.raw_get(bucket, key).await?;
+    assert_eq!(pulled.status, 200, "{}", String::from_utf8_lossy(&pulled.body));
+    assert_eq!(pulled.body, body);
+    assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "the pull must store the object");
+
+    let head = env.client.head_object().bucket(bucket).key(key).send().await?;
+    assert_eq!(
+        head.object_lock_mode().map(|mode| mode.as_str()),
+        Some("COMPLIANCE"),
+        "the default retention mode is applied to the pulled object"
+    );
+    assert!(head.object_lock_retain_until_date().is_some(), "a retain-until date is set");
+
+    let version_id = head.version_id().ok_or("an Object Lock bucket is versioned")?.to_string();
+    let error = env
+        .client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(&version_id)
+        .send()
+        .await
+        .expect_err("a COMPLIANCE-retained version cannot be deleted");
+    assert_eq!(error.code(), Some("AccessDenied"), "{error:?}");
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        1,
+        "the rejected delete never consults the source"
+    );
+    Ok(())
+}
+
+/// Case 14: the write-back obeys the bucket quota. The client is still
+/// served from the source, but nothing is stored and the failure is counted.
+#[tokio::test]
+async fn test_odm_write_back_respects_the_bucket_quota() -> TestResult {
+    let bucket = "odm-interaction-quota";
+    let env = start_configured_env_with(
+        OdmEnvOptions {
+            env: vec![("RUSTFS_SCANNER_CYCLE", "1"), ("RUSTFS_SCANNER_START_DELAY_SECS", "0")],
+            ..OdmEnvOptions::default()
+        },
+        bucket,
+        SOURCE_BUCKET,
+        |_| {},
+    )
+    .await?;
+
+    // Fill the bucket past the quota it is about to get, so the write-back's
+    // admission check has to reject it.
+    let filler = payload(2 * 1024 * 1024);
+    env.client
+        .put_object()
+        .bucket(bucket)
+        .key("quota/filler.bin")
+        .body(aws_sdk_s3::primitives::ByteStream::from(filler.clone()))
+        .send()
+        .await?;
+    wait_for_bucket_usage(&env, bucket, filler.len() as u64).await?;
+    set_bucket_quota(&env, bucket, 1024 * 1024).await?;
+
+    let key = "quota/oversized.bin";
+    let body = payload(2 * 1024 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    let response = env.raw_get(bucket, key).await?;
+    assert_eq!(response.status, 200, "{}", String::from_utf8_lossy(&response.body));
+    assert_eq!(response.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(response.body, body, "a full bucket still serves the client from the source");
+
+    // ODM-05 fixed the failure-reason label set without a `quota` value, so a
+    // rejected admission is reported as a local write failure.
+    env.wait_for_status_counter(bucket, "/counters/pull_failures_total/local_write", 1, SETTLE)
+        .await?;
+    env.assert_local_absent(bucket, key).await;
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        1,
+        "the rejected write-back is not retried against the source"
+    );
+    Ok(())
+}
+
+/// The quota route answers 503 until the durable-quota capability is
+/// confirmed on the fresh single-node deployment, so the write is retried.
+async fn set_bucket_quota(env: &OdmTestEnv, bucket: &str, quota_bytes: u64) -> TestResult {
+    let deadline = Instant::now() + Duration::from_secs(60);
+    loop {
+        let response = admin(
+            &env.rustfs,
+            http::Method::PUT,
+            &format!("/rustfs/admin/v3/quota/{bucket}"),
+            Some(serde_json::json!({ "quota": quota_bytes, "quota_type": "HARD" })),
+        )
+        .await?;
+        if response.status < 300 {
+            return Ok(());
+        }
+        if response.status != 503 || Instant::now() >= deadline {
+            return Err(format!("set quota for {bucket}: {} {}", response.status, response.body).into());
+        }
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+}
+
+async fn wait_for_bucket_usage(env: &OdmTestEnv, bucket: &str, at_least: u64) -> TestResult {
+    let deadline = Instant::now() + Duration::from_secs(60);
+    loop {
+        let response = admin(&env.rustfs, http::Method::GET, &format!("/rustfs/admin/v3/quota-stats/{bucket}"), None).await?;
+        if response.status == 200 {
+            let usage = serde_json::from_str::<Value>(&response.body)?
+                .get("current_usage")
+                .and_then(Value::as_u64)
+                .unwrap_or(0);
+            if usage >= at_least {
+                return Ok(());
+            }
+        }
+        if Instant::now() >= deadline {
+            return Err(format!("bucket usage for {bucket} did not reach {at_least} bytes: {}", response.body).into());
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+}
+
+/// Case 15: a pull emits an ordinary creation event attributed to the
+/// migration principal, and `emit_events=false` silences it.
+#[tokio::test]
+async fn test_odm_pull_emits_object_created_events_unless_disabled() -> TestResult {
+    let emitting = "odm-interaction-events";
+    let silent = "odm-interaction-events-off";
+    // The collector binds first: the outbound guard rejects a webhook
+    // endpoint on a private address unless its origin is allowed at startup.
+    let (endpoint, mut events) = spawn_event_collector().await?;
+    let allowed_origin = reqwest::Url::parse(&endpoint)?.origin().ascii_serialization();
+    let env = start_configured_env_with(
+        OdmEnvOptions {
+            env: vec![(ENV_OUTBOUND_ALLOW_ORIGINS, allowed_origin.as_str())],
+            ..OdmEnvOptions::default()
+        },
+        emitting,
+        SOURCE_BUCKET,
+        |_| {},
+    )
+    .await?;
+    let mut silent_spec = env.fake_source_spec(SOURCE_BUCKET);
+    silent_spec.policy.emit_events = false;
+    env.configure_and_wait(silent, &silent_spec).await?;
+
+    let target = "odm-events";
+    let switches = admin(
+        &env.rustfs,
+        http::Method::PUT,
+        "/rustfs/admin/v3/module-switches",
+        Some(serde_json::json!({ "notify_enabled": true, "audit_enabled": false })),
+    )
+    .await?;
+    assert_eq!(switches.status, 200, "{}", switches.body);
+    let queue_dir = format!("{}/notify-queue-{target}", env.rustfs.temp_dir);
+    tokio::fs::create_dir_all(&queue_dir).await?;
+    let configured = admin(
+        &env.rustfs,
+        http::Method::PUT,
+        &format!("/rustfs/admin/v3/target/notify_webhook/{target}"),
+        Some(serde_json::json!({
+            "key_values": [
+                { "key": "endpoint", "value": endpoint },
+                { "key": "queue_dir", "value": queue_dir },
+            ]
+        })),
+    )
+    .await?;
+    assert_eq!(configured.status, 200, "{}", configured.body);
+    wait_for_target_online(&env.rustfs, target).await?;
+    for bucket in [emitting, silent] {
+        put_notification_config(&env, bucket, target).await?;
+    }
+
+    // Control: an ordinary client PUT must produce an event, so a missing
+    // one below is about the write-back and not about the pipeline.
+    let control_key = "events/control.bin";
+    env.client
+        .put_object()
+        .bucket(emitting)
+        .key(control_key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(payload(1024)))
+        .send()
+        .await?;
+    let control = wait_for_event(&mut events, emitting, control_key, Duration::from_secs(60))
+        .await
+        .ok_or("the notification pipeline delivered no event for a plain PUT")?;
+    assert_eq!(
+        control.pointer("/eventName").and_then(Value::as_str),
+        Some("s3:ObjectCreated:Put"),
+        "{control}"
+    );
+
+    let emitting_key = "events/pulled.bin";
+    let silent_key = "events/quiet.bin";
+    let body = payload(16 * 1024);
+    env.seed_source(
+        SOURCE_BUCKET,
+        &[
+            SeedObject::new(emitting_key, body.clone()),
+            SeedObject::new(silent_key, body.clone()),
+        ],
+    );
+
+    for (bucket, key) in [(emitting, emitting_key), (silent, silent_key)] {
+        let response = env.raw_get(bucket, key).await?;
+        assert_eq!(response.status, 200, "{bucket}: {}", String::from_utf8_lossy(&response.body));
+        assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "{bucket}/{key} must be stored");
+        assert_eq!(env.source.count_requests(Operation::GetObject, key), 1, "{bucket}/{key}");
+    }
+
+    let record = wait_for_event(&mut events, emitting, emitting_key, Duration::from_secs(60))
+        .await
+        .ok_or("no creation event for the pulled object")?;
+    assert_eq!(
+        record.pointer("/eventName").and_then(Value::as_str),
+        Some("s3:ObjectCreated:Put"),
+        "{record}"
+    );
+    assert_eq!(
+        record.pointer("/userIdentity/principalId").and_then(Value::as_str),
+        Some(ODM_PRINCIPAL_ID),
+        "{record}"
+    );
+
+    // The silent bucket's object landed before the event above was observed,
+    // so a missing event here is a decision, not a race.
+    assert!(
+        wait_for_event(&mut events, silent, silent_key, Duration::from_secs(5))
+            .await
+            .is_none(),
+        "emit_events=false must not publish a creation event"
+    );
+    Ok(())
+}
+
+async fn put_notification_config(env: &OdmTestEnv, bucket: &str, target: &str) -> TestResult {
+    let queue = QueueConfiguration::builder()
+        .id(format!("{bucket}-rule"))
+        .queue_arn(format!("arn:rustfs:sqs:us-east-1:{target}:webhook"))
+        .events(Event::from("s3:ObjectCreated:*"))
+        .filter(
+            NotificationConfigurationFilter::builder()
+                .key(
+                    S3KeyFilter::builder()
+                        .filter_rules(FilterRule::builder().name(FilterRuleName::Prefix).value("events/").build())
+                        .build(),
+                )
+                .build(),
+        )
+        .build()?;
+    env.client
+        .put_bucket_notification_configuration()
+        .bucket(bucket)
+        .notification_configuration(NotificationConfiguration::builder().queue_configurations(queue).build())
+        .send()
+        .await?;
+    Ok(())
+}
+
+async fn wait_for_target_online(env: &RustFSTestEnvironment, target: &str) -> TestResult {
+    let deadline = Instant::now() + Duration::from_secs(30);
+    loop {
+        let response = admin(env, http::Method::GET, "/rustfs/admin/v3/target/list", None).await?;
+        if response.status == 200 {
+            let body: Value = serde_json::from_str(&response.body)?;
+            let online = body["notification_endpoints"].as_array().is_some_and(|endpoints| {
+                endpoints.iter().any(|endpoint| {
+                    endpoint["account_id"].as_str() == Some(target) && endpoint["status"].as_str() == Some("online")
+                })
+            });
+            if online {
+                return Ok(());
+            }
+        }
+        if Instant::now() >= deadline {
+            return Err(format!("webhook target {target} did not come online: {}", response.body).into());
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+}
+
+/// Minimal HTTP receiver: answers everything 200 (so the target's
+/// reachability probe reports online) and forwards parsed POST bodies.
+async fn spawn_event_collector() -> Result<(String, mpsc::UnboundedReceiver<Value>), BoxError> {
+    let listener = TcpListener::bind("0.0.0.0:0").await?;
+    let port = listener.local_addr()?.port();
+    let endpoint = format!("http://{}/events", std::net::SocketAddr::new(local_ip()?, port));
+    let (tx, rx) = mpsc::unbounded_channel();
+    tokio::spawn(async move {
+        while let Ok((mut stream, _)) = listener.accept().await {
+            let tx = tx.clone();
+            tokio::spawn(async move {
+                let mut buffer = Vec::new();
+                let mut chunk = [0_u8; 4096];
+                let mut content_length = 0usize;
+                let mut header_end = None;
+                while header_end.is_none() {
+                    match stream.read(&mut chunk).await {
+                        Ok(0) | Err(_) => return,
+                        Ok(read) => buffer.extend_from_slice(&chunk[..read]),
+                    }
+                    header_end = buffer.windows(4).position(|window| window == b"\r\n\r\n");
+                }
+                let header_end = header_end.expect("loop exits only with a header end");
+                let headers = String::from_utf8_lossy(&buffer[..header_end]).to_string();
+                for line in headers.split("\r\n").skip(1) {
+                    if let Some((name, value)) = line.split_once(':')
+                        && name.trim().eq_ignore_ascii_case("content-length")
+                    {
+                        content_length = value.trim().parse().unwrap_or(0);
+                    }
+                }
+                let body_offset = header_end + 4;
+                while buffer.len() - body_offset < content_length {
+                    match stream.read(&mut chunk).await {
+                        Ok(0) | Err(_) => return,
+                        Ok(read) => buffer.extend_from_slice(&chunk[..read]),
+                    }
+                }
+                let _ = stream
+                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
+                    .await;
+                let _ = stream.shutdown().await;
+                if let Ok(value) = serde_json::from_slice::<Value>(&buffer[body_offset..body_offset + content_length]) {
+                    let _ = tx.send(value);
+                }
+            });
+        }
+    });
+    Ok((endpoint, rx))
+}
+
+/// The first delivered record for `bucket`/`key`, or `None` on timeout.
+async fn wait_for_event(
+    events: &mut mpsc::UnboundedReceiver<Value>,
+    bucket: &str,
+    key: &str,
+    timeout: Duration,
+) -> Option<Value> {
+    let deadline = Instant::now() + timeout;
+    loop {
+        let remaining = deadline.checked_duration_since(Instant::now())?;
+        let envelope = tokio::time::timeout(remaining, events.recv()).await.ok()??;
+        for record in envelope["Records"].as_array().into_iter().flatten() {
+            // S3 event notifications URL-encode the object key.
+            let record_key = record.pointer("/s3/object/key").and_then(Value::as_str).map(|raw| {
+                urlencoding::decode(raw)
+                    .map(|decoded| decoded.into_owned())
+                    .unwrap_or_else(|_| raw.to_string())
+            });
+            if record.pointer("/s3/bucket/name").and_then(Value::as_str) == Some(bucket) && record_key.as_deref() == Some(key) {
+                return Some(record.clone());
+            }
+        }
+    }
+}
+
+/// Case 16: a pulled object enters the replication pipeline like any other
+/// write, and a configuration whose source is one of the bucket's own
+/// replication targets is rejected.
+#[tokio::test]
+async fn test_odm_pulled_object_replicates_and_target_as_source_is_rejected() -> TestResult {
+    let bucket = "odm-interaction-replication";
+    let replica_bucket = "odm-replica";
+    let fast_env = replication_fast_env();
+    let env = start_configured_env_with(
+        OdmEnvOptions {
+            env: fast_env.clone(),
+            ..OdmEnvOptions::default()
+        },
+        bucket,
+        SOURCE_BUCKET,
+        |_| {},
+    )
+    .await?;
+    let replica = FakeS3Target::start().await?;
+    replica.create_bucket(replica_bucket);
+
+    enable_versioning(&env, bucket).await?;
+    let arn = set_remote_target(&env.rustfs, bucket, &replica.address(), replica_bucket).await?;
+    put_bucket_replication(&env.rustfs, bucket, &arn).await?;
+
+    let key = "replicated/asset.bin";
+    let body = payload(64 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]);
+
+    let response = env.raw_get(bucket, key).await?;
+    assert_eq!(response.status, 200, "{}", String::from_utf8_lossy(&response.body));
+    assert_eq!(response.body, body);
+    assert!(env.wait_local_listed(bucket, key, SETTLE).await?, "the pull must store the object");
+
+    let deadline = Instant::now() + SETTLE;
+    while !replica.has_object(replica_bucket, key) {
+        assert!(Instant::now() < deadline, "the pulled object was never replicated to the target");
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, key),
+        1,
+        "replication reads the local copy, never the migration source"
+    );
+
+    let looping = OdmSourceSpec::for_fake_source(&replica, replica_bucket);
+    let rejected = env.configure_source(bucket, &looping).await?;
+    assert_eq!(
+        rejected.status, 400,
+        "a bucket may not migrate from its own replication target: {}",
+        rejected.body
+    );
+    Ok(())
+}
+
+async fn set_remote_target(
+    env: &RustFSTestEnvironment,
+    bucket: &str,
+    endpoint: &str,
+    target_bucket: &str,
+) -> Result<String, BoxError> {
+    let response = admin(
+        env,
+        http::Method::PUT,
+        &format!("/rustfs/admin/v3/set-remote-target?bucket={}", urlencoding::encode(bucket)),
+        Some(serde_json::json!({
+            "endpoint": endpoint,
+            "credentials": { "accessKey": FAKE_ACCESS_KEY, "secretKey": FAKE_SECRET_KEY },
+            "targetbucket": target_bucket,
+            "secure": false,
+            "skipTlsVerify": false,
+            "type": "replication"
+        })),
+    )
+    .await?;
+    if response.status != 200 {
+        return Err(format!("set remote target: {} {}", response.status, response.body).into());
+    }
+    Ok(serde_json::from_str(&response.body)?)
+}
+
+async fn put_bucket_replication(env: &RustFSTestEnvironment, bucket: &str, arn: &str) -> TestResult {
+    let body = format!(
+        r#"<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Role></Role>
+  <Rule>
+    <ID>odm-rule</ID>
+    <Priority>1</Priority>
+    <Status>Enabled</Status>
+    <DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication>
+    <ExistingObjectReplication><Status>Enabled</Status></ExistingObjectReplication>
+    <Destination><Bucket>{arn}</Bucket></Destination>
+  </Rule>
+</ReplicationConfiguration>"#
+    );
+    let url = format!("{}/{bucket}?replication", env.url);
+    let response = signed_request(
+        http::Method::PUT,
+        &url,
+        &env.access_key,
+        &env.secret_key,
+        Some(body.into_bytes()),
+        Some("application/xml"),
+    )
+    .await?;
+    if response.status() != 200 {
+        let status = response.status();
+        return Err(format!("put bucket replication: {status} {}", response.text().await.unwrap_or_default()).into());
+    }
+    Ok(())
+}
+
+/// Case 17: a local delete marker is the authoritative answer in a versioned
+/// bucket, while an unversioned delete leaves nothing behind and the key is
+/// migrated again.
+#[tokio::test]
+async fn test_odm_delete_marker_shadows_the_source_but_a_plain_delete_does_not() -> TestResult {
+    let versioned = "odm-interaction-delete-marker";
+    let unversioned = "odm-interaction-plain-delete";
+    let env = start_configured_env(versioned, SOURCE_BUCKET, |_| {}).await?;
+    let spec = env.fake_source_spec(SOURCE_BUCKET);
+    env.configure_and_wait(unversioned, &spec).await?;
+    enable_versioning(&env, versioned).await?;
+
+    let key = "deleted/doc.bin";
+    let source_body = payload(8 * 1024);
+    let local_body = payload(4 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, source_body.clone())]);
+
+    for bucket in [versioned, unversioned] {
+        env.client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(aws_sdk_s3::primitives::ByteStream::from(local_body.clone()))
+            .send()
+            .await?;
+        env.client.delete_object().bucket(bucket).key(key).send().await?;
+    }
+
+    let shadowed = env.raw_get(versioned, key).await?;
+    assert_eq!(shadowed.status, 404, "{}", String::from_utf8_lossy(&shadowed.body));
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, key),
+        0,
+        "a local delete marker answers without the source"
+    );
+
+    let migrated = env.raw_get(unversioned, key).await?;
+    assert_eq!(migrated.status, 200, "{}", String::from_utf8_lossy(&migrated.body));
+    assert_eq!(migrated.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(migrated.body, source_body, "an unversioned delete leaves the source authoritative");
+    assert_eq!(env.source.count_requests(Operation::HeadObject, key), 1);
+    assert_eq!(env.source.count_requests(Operation::GetObject, key), 1);
+    Ok(())
+}
+
+/// Case 18: deleting the configuration stops all source traffic without
+/// touching what was already migrated, and reinstalling it resumes.
+#[tokio::test]
+async fn test_odm_disable_keeps_pulled_objects_and_stops_source_traffic() -> TestResult {
+    let bucket = "odm-interaction-disable";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |_| {}).await?;
+    let pulled_key = "disable/pulled.bin";
+    let untouched_key = "disable/untouched.bin";
+    let body = payload(32 * 1024);
+    env.seed_source(
+        SOURCE_BUCKET,
+        &[
+            SeedObject::new(pulled_key, body.clone()),
+            SeedObject::new(untouched_key, body.clone()),
+        ],
+    );
+
+    let pulled = env.raw_get(bucket, pulled_key).await?;
+    assert_eq!(pulled.status, 200, "{}", String::from_utf8_lossy(&pulled.body));
+    assert!(env.wait_local_listed(bucket, pulled_key, SETTLE).await?);
+
+    let disabled = env.disable(bucket).await?;
+    assert_eq!(disabled.status, 204, "{}", disabled.body);
+
+    let still_readable = env.raw_get(bucket, pulled_key).await?;
+    assert_eq!(still_readable.status, 200, "{}", String::from_utf8_lossy(&still_readable.body));
+    assert_eq!(still_readable.body, body, "a migrated object survives the disable");
+    assert_eq!(still_readable.header(ODM_RESPONSE_HEADER), None);
+    assert_eq!(env.source.count_requests(Operation::GetObject, pulled_key), 1);
+
+    let missing = env.raw_get(bucket, untouched_key).await?;
+    assert_eq!(missing.status, 404, "{}", String::from_utf8_lossy(&missing.body));
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, untouched_key),
+        0,
+        "a disabled bucket never reaches the source"
+    );
+
+    let spec = env.fake_source_spec(SOURCE_BUCKET);
+    env.configure_and_wait(bucket, &spec).await?;
+    let resumed = env.raw_get(bucket, untouched_key).await?;
+    assert_eq!(resumed.status, 200, "{}", String::from_utf8_lossy(&resumed.body));
+    assert_eq!(resumed.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(resumed.body, body);
+    assert_eq!(env.source.count_requests(Operation::GetObject, untouched_key), 1);
+    Ok(())
+}
+
+/// Case 19: the admin surface an operator sees — the configuration read back
+/// without its secret, and a status document whose counters match the source
+/// journal exactly.
+#[tokio::test]
+async fn test_odm_admin_config_is_redacted_and_status_counts_match_the_source() -> TestResult {
+    let bucket = "odm-interaction-admin";
+    let env = start_configured_env(bucket, SOURCE_BUCKET, |_| {}).await?;
+    let hit_key = "admin/present.bin";
+    let miss_key = "admin/absent.bin";
+    let body = payload(16 * 1024);
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(hit_key, body.clone())]);
+
+    let config = env.get_config(bucket).await?;
+    assert_eq!(config.status, 200, "{}", config.body);
+    let config = config.json()?;
+    assert_eq!(
+        config
+            .pointer("/config/source/credentials/secret_key")
+            .and_then(Value::as_str),
+        Some("REDACTED"),
+        "{config}"
+    );
+    assert_eq!(
+        config
+            .pointer("/config/source/credentials/access_key")
+            .and_then(Value::as_str),
+        Some(FAKE_ACCESS_KEY),
+        "the access key stays readable: {config}"
+    );
+    assert!(
+        !config.to_string().contains(FAKE_SECRET_KEY),
+        "the secret must not appear anywhere in the response"
+    );
+
+    let hit = env.raw_get(bucket, hit_key).await?;
+    assert_eq!(hit.status, 200, "{}", String::from_utf8_lossy(&hit.body));
+    for _ in 0..2 {
+        let miss = env.raw_get(bucket, miss_key).await?;
+        assert_eq!(miss.status, 404, "{}", String::from_utf8_lossy(&miss.body));
+    }
+    assert!(env.wait_local_listed(bucket, hit_key, SETTLE).await?);
+
+    let status = env.status_json(bucket).await?;
+    assert_eq!(status.pointer("/configured").and_then(Value::as_bool), Some(true), "{status}");
+    assert_eq!(status.pointer("/enabled").and_then(Value::as_bool), Some(true), "{status}");
+    assert_eq!(status.pointer("/module_enabled").and_then(Value::as_bool), Some(true), "{status}");
+    assert_eq!(status.pointer("/provider").and_then(Value::as_str), Some("s3"), "{status}");
+    assert_eq!(
+        status
+            .pointer("/counters/requests_total/get/source_hit")
+            .and_then(Value::as_u64),
+        Some(1),
+        "one source hit, matching the one source GET: {status}"
+    );
+    assert_eq!(
+        status
+            .pointer("/counters/requests_total/get/source_miss")
+            .and_then(Value::as_u64),
+        Some(1),
+        "only the first miss reached the source: {status}"
+    );
+    assert_eq!(
+        status
+            .pointer("/counters/requests_total/get/negative_cached")
+            .and_then(Value::as_u64),
+        Some(1),
+        "the second miss stopped at the negative cache: {status}"
+    );
+    assert_eq!(
+        status
+            .pointer("/counters/pulled_objects_total/inline")
+            .and_then(Value::as_u64),
+        Some(1),
+        "{status}"
+    );
+    assert_eq!(
+        status.pointer("/counters/pulled_bytes_total").and_then(Value::as_u64),
+        Some(body.len() as u64),
+        "{status}"
+    );
+    assert_eq!(env.source.count_requests(Operation::GetObject, hit_key), 1);
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, miss_key),
+        1,
+        "the status counters and the source journal agree"
+    );
+    Ok(())
+}

crates/e2e_test/src/on_demand_migration/mod.rs

@@ -17,9 +17,16 @@
 //! `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; `get_basic_test` covers the
-//! GET read-through (rustfs/backlog#2156).
+//! GET read-through (rustfs/backlog#2156). The fault, concurrency,
+//! interaction and real-source matrix is rustfs/backlog#2158; its lane split
+//! lives in `.config/nextest.toml` (fault / concurrency / real source run
+//! nightly, the rest in the merge lane).
 
 pub mod common;
 
+mod concurrency_test;
+mod fault_test;
 mod get_basic_test;
 mod harness_self_test;
+mod interaction_test;
+mod real_source_test;

crates/e2e_test/src/on_demand_migration/real_source_test.rs

@@ -0,0 +1,266 @@
+// 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 against a real RustFS source (rustfs/backlog#2158).
+//!
+//! These cases start a second (and, for the loop guard, a third) RustFS
+//! process, so they carry the `_real_single_node` marker and run in the
+//! nightly lane. A real source keeps no request journal, so "the source was
+//! not consulted" is proven by removing the object from the source and
+//! showing the read still succeeds, or by pointing the *second* server at a
+//! fake source whose journal must stay empty.
+
+use super::common::{
+    AdminResponse, BoxError, ODM_ADMIN_ROUTE, OdmSourceSpec, OdmTestEnv, RawResponse, SeedObject, start_source_rustfs,
+    start_source_rustfs_with_odm,
+};
+use crate::common::{RustFSTestEnvironment, signed_request};
+use crate::fake_s3_target::{BucketMode, Operation};
+use aws_sdk_s3::Client;
+use bytes::Bytes;
+use std::time::{Duration, Instant};
+
+type TestResult = Result<(), BoxError>;
+
+const ODM_RESPONSE_HEADER: &str = "x-rustfs-on-demand-migration";
+/// Reinstalled configurations are applied asynchronously; every phase polls
+/// for the new behavior instead of sleeping.
+const APPLY_TIMEOUT: Duration = Duration::from_secs(30);
+const SETTLE: Duration = Duration::from_secs(60);
+
+fn payload(len: usize) -> Bytes {
+    (0..len).map(|index| (index % 251) as u8).collect::<Vec<u8>>().into()
+}
+
+/// `PUT /rustfs/admin/v3/on-demand-migration/{bucket}` against any server,
+/// not just the one under test.
+async fn configure_odm(env: &RustFSTestEnvironment, bucket: &str, spec: &OdmSourceSpec) -> Result<AdminResponse, BoxError> {
+    let url = format!("{}{ODM_ADMIN_ROUTE}/{bucket}", env.url);
+    let body = serde_json::to_vec(&spec.to_json())?;
+    let response = signed_request(
+        http::Method::PUT,
+        &url,
+        &env.access_key,
+        &env.secret_key,
+        Some(body),
+        Some("application/json"),
+    )
+    .await?;
+    Ok(AdminResponse {
+        status: response.status().as_u16(),
+        body: response.text().await?,
+    })
+}
+
+async fn put_object(client: &Client, bucket: &str, key: &str, body: Bytes) -> TestResult {
+    client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(body))
+        .send()
+        .await?;
+    Ok(())
+}
+
+/// Polls a read against the server under test until it answers `expected`.
+/// This is how a reinstalled configuration is waited for when the source
+/// keeps no journal to probe.
+async fn wait_for_get_status(env: &OdmTestEnv, bucket: &str, key: &str, expected: u16) -> Result<RawResponse, BoxError> {
+    let deadline = Instant::now() + APPLY_TIMEOUT;
+    loop {
+        let response = env.raw_get(bucket, key).await?;
+        if response.status == expected {
+            return Ok(response);
+        }
+        if Instant::now() >= deadline {
+            return Err(format!(
+                "GET {bucket}/{key} stayed at {} instead of {expected}: {}",
+                response.status,
+                String::from_utf8_lossy(&response.body)
+            )
+            .into());
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+}
+
+/// Case 20: a second RustFS as the migration source — the pull, a HEAD
+/// passthrough, a Range read, and both prefix knobs.
+#[tokio::test]
+async fn test_odm_rustfs_source_serves_pull_head_range_and_prefixes_real_single_node() -> TestResult {
+    let bucket = "odm-real-source";
+    let source_bucket = "odm-real-origin";
+    let source = start_source_rustfs().await?;
+    let source_client = source.create_s3_client();
+    source.create_test_bucket(source_bucket).await?;
+
+    let env = OdmTestEnv::start().await?;
+    env.rustfs.create_test_bucket(bucket).await?;
+    let spec = OdmSourceSpec::for_rustfs_source(&source, source_bucket);
+    let configured = configure_odm(&env.rustfs, bucket, &spec).await?;
+    assert_eq!(configured.status, 200, "{}", configured.body);
+
+    // Phase 1: a miss is pulled and stored; removing it from the source
+    // afterwards proves the second read never goes back to the source.
+    let pulled_key = "real/pulled.bin";
+    let pulled_body = payload(256 * 1024);
+    put_object(&source_client, source_bucket, pulled_key, pulled_body.clone()).await?;
+    let pulled = wait_for_get_status(&env, bucket, pulled_key, 200).await?;
+    assert_eq!(pulled.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(pulled.body, pulled_body, "the client receives the source bytes");
+    assert!(env.wait_local_listed(bucket, pulled_key, SETTLE).await?, "the pull must store the object");
+    source_client
+        .delete_object()
+        .bucket(source_bucket)
+        .key(pulled_key)
+        .send()
+        .await?;
+    let local = env.raw_get(bucket, pulled_key).await?;
+    assert_eq!(local.status, 200, "{}", String::from_utf8_lossy(&local.body));
+    assert_eq!(local.header(ODM_RESPONSE_HEADER), None, "a local hit is not marked");
+    assert_eq!(local.body, pulled_body, "the object is served from the local copy");
+
+    // Phase 2: HEAD proxies metadata without storing anything.
+    let head_key = "real/head-only.bin";
+    let head_body = payload(9_000);
+    put_object(&source_client, source_bucket, head_key, head_body.clone()).await?;
+    let head = env.raw_object_request(http::Method::HEAD, bucket, head_key, &[]).await?;
+    assert_eq!(head.status, 200, "HEAD must be answered from the source");
+    assert_eq!(head.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(head.header("content-length"), Some(head_body.len().to_string().as_str()));
+    env.assert_local_absent(bucket, head_key).await;
+
+    // Phase 3: a Range read is passed through as a 206.
+    let range_key = "real/range.bin";
+    let range_body = payload(100_000);
+    put_object(&source_client, source_bucket, range_key, range_body.clone()).await?;
+    let ranged = env
+        .raw_object_request(http::Method::GET, bucket, range_key, &[("range", "bytes=100-199")])
+        .await?;
+    assert_eq!(ranged.status, 206, "{}", String::from_utf8_lossy(&ranged.body));
+    assert_eq!(ranged.header("content-range"), Some("bytes 100-199/100000"));
+    assert_eq!(ranged.body, range_body.slice(100..200));
+
+    // Phase 4: `filter.prefix` decides which local keys may consult the
+    // source at all.
+    let allowed_key = "allowed/doc.bin";
+    let denied_key = "denied/doc.bin";
+    let filtered_body = payload(4_096);
+    put_object(&source_client, source_bucket, allowed_key, filtered_body.clone()).await?;
+    put_object(&source_client, source_bucket, denied_key, filtered_body.clone()).await?;
+    let mut filtered = OdmSourceSpec::for_rustfs_source(&source, source_bucket);
+    filtered.filter.prefix = Some("allowed/".to_string());
+    let response = configure_odm(&env.rustfs, bucket, &filtered).await?;
+    assert_eq!(response.status, 200, "{}", response.body);
+    let denied = wait_for_get_status(&env, bucket, denied_key, 404).await?;
+    assert_eq!(denied.header(ODM_RESPONSE_HEADER), None);
+    env.assert_local_absent(bucket, denied_key).await;
+    let allowed = wait_for_get_status(&env, bucket, allowed_key, 200).await?;
+    assert_eq!(allowed.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(allowed.body, filtered_body);
+
+    // Phase 5: `filter.source_prefix` rewrites the key on the way out, so a
+    // local key resolves to a different key in the source bucket.
+    let rewritten_key = "rewritten/doc.bin";
+    let rewritten_body = payload(2_048);
+    put_object(&source_client, source_bucket, &format!("archive/{rewritten_key}"), rewritten_body.clone()).await?;
+    let mut rewriting = OdmSourceSpec::for_rustfs_source(&source, source_bucket);
+    rewriting.filter.source_prefix = Some("archive/".to_string());
+    let response = configure_odm(&env.rustfs, bucket, &rewriting).await?;
+    assert_eq!(response.status, 200, "{}", response.body);
+    let rewritten = wait_for_get_status(&env, bucket, rewritten_key, 200).await?;
+    assert_eq!(rewritten.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(rewritten.body, rewritten_body, "the source prefix is prepended to the local key");
+    Ok(())
+}
+
+/// Case 21: two migrating servers pointed at each other must not build a
+/// request loop. The middle server also has a fake source of its own, whose
+/// journal is the evidence: a key it would happily fetch for a direct client
+/// is never fetched for a request that arrived with the anti-loop marker.
+#[tokio::test]
+async fn test_odm_chained_sources_stop_at_the_loop_guard_real_single_node() -> TestResult {
+    let bucket = "odm-loop-guard";
+    let fake_bucket = "odm-loop-fake";
+    let env = OdmTestEnv::start().await?;
+    env.source.create_bucket_with_mode(fake_bucket, BucketMode::Unversioned);
+    env.rustfs.create_test_bucket(bucket).await?;
+
+    let middle = start_source_rustfs_with_odm().await?;
+    let middle_client = middle.create_s3_client();
+    middle.create_test_bucket(bucket).await?;
+    let middle_spec = OdmSourceSpec::for_fake_source(&env.source, fake_bucket);
+    let configured = configure_odm(&middle, bucket, &middle_spec).await?;
+    assert_eq!(configured.status, 200, "{}", configured.body);
+
+    let chained = OdmSourceSpec::for_rustfs_source(&middle, bucket);
+    let configured = configure_odm(&env.rustfs, bucket, &chained).await?;
+    assert_eq!(configured.status, 200, "{}", configured.body);
+
+    // The first hop works: an object that only the middle server holds is
+    // migrated to the server under test.
+    let present_key = "loop/present.bin";
+    let present_body = payload(16 * 1024);
+    put_object(&middle_client, bucket, present_key, present_body.clone()).await?;
+    let served = wait_for_get_status(&env, bucket, present_key, 200).await?;
+    assert_eq!(served.header(ODM_RESPONSE_HEADER), Some("source"));
+    assert_eq!(served.body, present_body, "the first hop serves the middle server's object");
+
+    // The second hop does not: this key exists only on the middle server's
+    // own source, and the anti-loop marker stops the chain there.
+    let guarded_key = "loop/chain-guard.bin";
+    let guarded_body = payload(8 * 1024);
+    env.seed_source(fake_bucket, &[SeedObject::new(guarded_key, guarded_body.clone())]);
+    let guarded = env.raw_get(bucket, guarded_key).await?;
+    assert_eq!(guarded.status, 404, "{}", String::from_utf8_lossy(&guarded.body));
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, guarded_key),
+        0,
+        "a chained request must not reach a third source"
+    );
+    assert_eq!(env.source.count_requests(Operation::GetObject, guarded_key), 0);
+
+    // Proof that the guard, and not a broken configuration, is what stopped
+    // it: the same key served directly by the middle server does reach the
+    // fake source.
+    let direct = middle_client.get_object().bucket(bucket).key(guarded_key).send().await?;
+    assert_eq!(direct.body.collect().await?.into_bytes(), guarded_body);
+    assert_eq!(
+        env.source.count_requests(Operation::GetObject, guarded_key),
+        1,
+        "an unmarked request does consult the middle server's source"
+    );
+
+    // Now make the pair mutual and prove the read still terminates.
+    let mutual = OdmSourceSpec::for_rustfs_source(&env.rustfs, bucket);
+    let configured = configure_odm(&middle, bucket, &mutual).await?;
+    assert_eq!(configured.status, 200, "{}", configured.body);
+
+    let mutual_key = "loop/mutual.bin";
+    let started = Instant::now();
+    let response = wait_for_get_status(&env, bucket, mutual_key, 404).await?;
+    assert_eq!(response.header(ODM_RESPONSE_HEADER), None);
+    assert!(
+        started.elapsed() < Duration::from_secs(10),
+        "a mutual configuration must not loop, took {:?}",
+        started.elapsed()
+    );
+    assert_eq!(
+        env.source.count_requests(Operation::HeadObject, mutual_key),
+        0,
+        "the fake source is out of the chain once the pair is mutual"
+    );
+    Ok(())
+}