diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index cccf24577..d8895815d 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=9dccb0cd537cf79ae70c1c20e8281d36d03f2f09f81142a5341e26e3dc18709d +sha256-darwin=09bc29b3e0ed7c5779331740e18e1d5ca05cb240c1363d1ecc95c1ef1e337038 sha256-linux=86e69337ad1440252a2ee20a12063c989ed12442d3b1ddf9e9233acf0f2ec089 diff --git a/.config/e2e-nightly-selection.txt b/.config/e2e-nightly-selection.txt index 002f5c318..c54d1cf59 100644 --- a/.config/e2e-nightly-selection.txt +++ b/.config/e2e-nightly-selection.txt @@ -1 +1 @@ -sha256=d06524b44de97ed8f62b0fd8cf9fa504e3cd520ffcaacc32691d6f890ebe7f20 +sha256=e58d2993aef56373d52ea40dfc6733fef42865195a71ccbd3883b0fafd84da75 diff --git a/.config/nextest.toml b/.config/nextest.toml index 32adfa6bb..d436d3f24 100644 --- a/.config/nextest.toml +++ b/.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 diff --git a/crates/e2e_test/src/on_demand_migration/common.rs b/crates/e2e_test/src/on_demand_migration/common.rs index 4058e2a59..8886e64a5 100644 --- a/crates/e2e_test/src/on_demand_migration/common.rs +++ b/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 { - 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 { + 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 { + let url = format!("{}/{bucket}/{key}", self.rustfs.url); + let uri = url.parse::()?; + 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 { + 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 { + 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 { + 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(&self, bucket: &str, work: F) -> Result<(T, u64), BoxError> + where + F: std::future::Future, + { + 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 { 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 { + 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 { + 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 { + 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) +} diff --git a/crates/e2e_test/src/on_demand_migration/concurrency_test.rs b/crates/e2e_test/src/on_demand_migration/concurrency_test.rs new file mode 100644 index 000000000..60d75a099 --- /dev/null +++ b/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::>().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 = 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 = (0..KEYS).map(|index| format!("ceiling/object-{index:03}.bin")).collect(); + let seeds: Vec = 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 = (0..REQUESTS).map(|index| format!("queue/object-{index:03}.bin")).collect(); + let seeds: Vec = keys.iter().map(|key| SeedObject::new(key.clone(), body.clone())).collect(); + env.seed_source(SOURCE_BUCKET, &seeds); + + let responses: Vec = 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(()) +} diff --git a/crates/e2e_test/src/on_demand_migration/fault_test.rs b/crates/e2e_test/src/on_demand_migration/fault_test.rs new file mode 100644 index 000000000..b51ea370b --- /dev/null +++ b/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::>().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::()), + 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(()) +} diff --git a/crates/e2e_test/src/on_demand_migration/interaction_test.rs b/crates/e2e_test/src/on_demand_migration/interaction_test.rs new file mode 100644 index 000000000..c810c4f4b --- /dev/null +++ b/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::>().into() +} + +async fn admin( + env: &RustFSTestEnvironment, + method: http::Method, + path: &str, + body: Option, +) -> Result { + 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::(&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), 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::(&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, + bucket: &str, + key: &str, + timeout: Duration, +) -> Option { + 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 { + 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#" + + + odm-rule + 1 + Enabled + Enabled + Enabled + {arn} + +"# + ); + 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(()) +} diff --git a/crates/e2e_test/src/on_demand_migration/mod.rs b/crates/e2e_test/src/on_demand_migration/mod.rs index 4dd04dced..54686c386 100644 --- a/crates/e2e_test/src/on_demand_migration/mod.rs +++ b/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; diff --git a/crates/e2e_test/src/on_demand_migration/real_source_test.rs b/crates/e2e_test/src/on_demand_migration/real_source_test.rs new file mode 100644 index 000000000..4e8d78792 --- /dev/null +++ b/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::>().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 { + 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 { + 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(()) +}