diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index bfcdfad20..74b157fe7 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=ef914ec0b8daa9c2c5e52f501d339914662f42d6f6ed9d33877d56b97adf16f9 +sha256-darwin=9dccb0cd537cf79ae70c1c20e8281d36d03f2f09f81142a5341e26e3dc18709d sha256-linux=a8a816d7bb0e7cb5632b1863b33794bcb9fc7e765f150aa5e1bf16518e28dfb4 diff --git a/crates/e2e_test/src/fake_s3_target/README.md b/crates/e2e_test/src/fake_s3_target/README.md index 7f1feae0b..4546d9a66 100644 --- a/crates/e2e_test/src/fake_s3_target/README.md +++ b/crates/e2e_test/src/fake_s3_target/README.md @@ -1,11 +1,15 @@ # Programmable fake S3 target -This module is the shared failure-injection boundary for replication end-to-end tests. It runs an in-process, path-style S3 endpoint backed by `s3s`; no production crate depends on it. +This module is the shared failure-injection boundary for replication end-to-end tests and the programmable external source for on-demand-migration (ODM) tests. It runs an in-process, path-style S3 endpoint backed by `s3s`; no production crate depends on it. `FakeS3Target::start()` creates the listener. Add target buckets with `create_bucket`, point a RustFS remote target at `address()`, use `FAKE_ACCESS_KEY` / `FAKE_SECRET_KEY`, then enqueue per-operation faults with `inject`. Faults for one operation are consumed in FIFO order and do not consume faults queued for another operation. A fault is consumed only after `s3s` verifies the full request signature, so anonymous, other-access-key, and bad-signature traffic cannot disturb a script. -Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB. +Supported data operations are HeadBucket, GetBucketVersioning, ListObjectsV2, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets created with `create_bucket` are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB. -Fault actions cover HTTP 401/403/503 responses, pre-dispatch delay, connection abort when a logical request-body threshold is reached, streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions. Each record also journals a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract. +`create_bucket_with_mode(name, BucketMode::Unversioned)` models a plain migration source: PUT overwrites in place, DELETE removes the key without a delete marker, GetBucketVersioning reports no status, and no `x-amz-version-id` is returned by PUT, GET, HEAD, tagging, or multipart completion. The only `versionId` such a bucket accepts is `null`; any other value is rejected with `InvalidArgument`. The mode is fixed at creation. -The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type at 1 KiB. A PUT or uploaded part is capped at 64 MiB; a completed multipart object and all stored object/part data are capped at 128 MiB. Body drain, body-permit waits, delay, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound. +ListObjectsV2 lists current versions only (a key whose newest version is a delete marker is hidden) in byte order and supports `prefix`, `delimiter`, `max-keys` (clamped to 1000), `start-after`, and `continuation-token`; common prefixes count toward `max-keys`, `IsTruncated` / `NextContinuationToken` / `KeyCount` follow S3, and continuation tokens are opaque. `encoding-type` and `fetch-owner` are accepted but ignored, and ListObjects (v1) is not implemented. GET and HEAD honor `Range` in the `bytes=first-last`, `bytes=first-`, and `bytes=-suffix` forms with a 206 status, exact `Content-Range`, and `Accept-Ranges: bytes`; unsatisfiable ranges answer 416 `InvalidRange` with `Content-Range: bytes */`. PUT and CreateMultipartUpload accept `Content-Type`, `Content-Encoding`, `Content-Disposition`, `Content-Language`, `Cache-Control`, `Expires`, and `x-amz-meta-*` (names stored lowercased), and HEAD/GET replay them verbatim together with `Last-Modified` and the ETag (hex MD5 for single PUTs, `-` for multipart objects). `put_seed_object` stores an object directly, bypassing the wire, the fault script, and the journal, so a source can be seeded without polluting the assertions a scenario later makes. + +Fault actions cover HTTP 401/403/503 responses (`Status`), any 4xx/5xx status paired with the matching S3 error code (`ResponseStatus`), pre-dispatch delay, holding a fully computed successful response before its first byte (`Stall`), connection abort when a logical request-body threshold is reached, GetObject bodies cut off after N bytes while `Content-Length` announces the full size (`TruncateBodyAt`), streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions and `count_requests(operation, key)` counts entries for one exact key. Each record journals the `Range` and `User-Agent` request headers, the ListObjectsV2 `prefix` and `continuation-token` query values, and a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract. + +The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type and each standard object header at 1 KiB. By default a PUT or uploaded part is capped at 64 MiB and a completed multipart object and all stored object/part data are capped at 128 MiB; `FakeS3Target::start_with_options(FakeS3TargetOptions { max_object_bytes })` raises the object cap up to 256 MiB, and the total budget then becomes twice the object cap (never below 128 MiB). Body drain, body-permit waits, delay, stall, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound. diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index c5d7edffb..d9be301d4 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -12,14 +12,15 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Programmable in-process S3 target for replication failure-path tests. +//! Programmable in-process S3 target for replication failure-path tests and +//! on-demand-migration source scenarios. //! //! See [`README.md`](README.md) for the supported protocol and fault surface. use async_trait::async_trait; use bytes::{Bytes, BytesMut}; use futures::StreamExt; -use http::header::{CONTENT_LENGTH, ETAG, LAST_MODIFIED}; +use http::header::{CONTENT_LENGTH, ETAG, LAST_MODIFIED, RANGE, USER_AGENT}; use http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode, Uri}; use hyper::body::Incoming; use hyper::server::conn::http1; @@ -29,13 +30,13 @@ use md5::{Digest as Md5Digest, Md5}; use s3s::access::{S3Access, S3AccessContext}; use s3s::auth::SimpleAuth; use s3s::dto::{ - AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput, - CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput, - DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput, GetBucketVersioningOutput, - GetObjectInput, GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput, - HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ObjectVersionId, PutObjectInput, - PutObjectOutput, PutObjectTaggingInput, PutObjectTaggingOutput, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat, - UploadPartInput, UploadPartOutput, + AbortMultipartUploadInput, AbortMultipartUploadOutput, CommonPrefix, CompleteMultipartUploadInput, + CompleteMultipartUploadOutput, CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, + DeleteObjectOutput, DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput, + GetBucketVersioningOutput, GetObjectInput, GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, + HeadBucketOutput, HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsV2Input, + ListObjectsV2Output, Object, ObjectStorageClass, ObjectVersionId, PutObjectInput, PutObjectOutput, PutObjectTaggingInput, + PutObjectTaggingOutput, Range, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat, UploadPartInput, UploadPartOutput, }; use s3s::service::{S3Service, S3ServiceBuilder}; use s3s::validation::{AwsNameValidation, NameValidation}; @@ -55,6 +56,8 @@ use uuid::Uuid; type BoxError = Box; const MAX_BUFFERED_BODY_BYTES: usize = 64 * 1024 * 1024; +/// Hard ceiling for [`FakeS3TargetOptions::max_object_bytes`]. +const MAX_CONFIGURABLE_OBJECT_BYTES: usize = 256 * 1024 * 1024; const MAX_TOTAL_STORED_BYTES: usize = 128 * 1024 * 1024; const MAX_OBJECT_VERSIONS: usize = 4096; const MAX_MULTIPART_UPLOADS: usize = 256; @@ -72,7 +75,12 @@ const MAX_FAULT_DURATION: Duration = Duration::from_secs(30); const MAX_REQUEST_DURATION: Duration = Duration::from_secs(65); const MAX_CONNECTION_DURATION: Duration = Duration::from_secs(100); const DISCONNECT_HEADER: &str = "x-rustfs-fake-target-disconnect"; +const STALL_HEADER: &str = "x-rustfs-fake-target-stall-ms"; const WRONG_ETAG: &str = "\"fake-target-wrong-etag\""; +/// Version id stored for objects in unversioned buckets. It is the only id a +/// versioned request may address there, and it is never echoed on the wire. +const NULL_VERSION_ID: &str = "null"; +const MAX_LIST_KEYS: i32 = 1000; const SOURCE_VERSION_ID_HEADERS: [&str; 2] = ["x-rustfs-source-version-id", "x-minio-source-version-id"]; const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-source-mtime"]; const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] = @@ -104,7 +112,7 @@ pub const FAKE_ACCESS_KEY: &str = "fake-access"; pub const FAKE_SECRET_KEY: &str = "fake-secret"; /// S3 operations understood by the target and its request journal. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] pub enum Operation { HeadBucket, GetBucketVersioning, @@ -116,18 +124,148 @@ pub enum Operation { PutObjectTagging, DeleteObjectTagging, ListObjectVersions, + ListObjectsV2, CreateMultipartUpload, UploadPart, CompleteMultipartUpload, AbortMultipartUpload, + #[default] Unknown, } +/// Versioning mode chosen at bucket creation; it never changes afterwards. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BucketMode { + /// PUT appends a version, DELETE writes a delete marker (the replication + /// target model and the `create_bucket` default). + Versioned, + /// PUT overwrites in place, DELETE removes the key, and no + /// `x-amz-version-id` is ever returned (a plain migration source). + Unversioned, +} + +/// Per-instance knobs for [`FakeS3Target::start_with_options`]. The default +/// keeps every limit documented in the README. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FakeS3TargetOptions { + /// Largest single PUT body, uploaded part, and stored object. Raising it + /// above the 64 MiB default also raises the total storage budget to twice + /// this value so one object of that size can still be stored; 256 MiB is + /// the hard ceiling. + pub max_object_bytes: usize, +} + +impl Default for FakeS3TargetOptions { + fn default() -> Self { + Self { + max_object_bytes: MAX_BUFFERED_BODY_BYTES, + } + } +} + +/// Metadata attached to an object stored through +/// [`FakeS3Target::put_seed_object`]. User metadata names are stored +/// lowercased, exactly as they would arrive as `x-amz-meta-*` headers. +#[derive(Clone, Default)] +pub struct SeedMetadata { + content_type: Option, + metadata: Vec<(String, String)>, + standard_headers: StandardHeaders, + last_modified: Option, +} + +impl SeedMetadata { + pub fn new() -> Self { + Self::default() + } + + pub fn content_type(mut self, value: impl Into) -> Self { + self.content_type = Some(value.into()); + self + } + + /// Add one `x-amz-meta-` entry (name given without the prefix). + pub fn user_metadata(mut self, name: impl Into, value: impl Into) -> Self { + self.metadata.push((name.into(), value.into())); + self + } + + pub fn cache_control(mut self, value: impl Into) -> Self { + self.standard_headers.cache_control = Some(value.into()); + self + } + + pub fn content_disposition(mut self, value: impl Into) -> Self { + self.standard_headers.content_disposition = Some(value.into()); + self + } + + pub fn content_encoding(mut self, value: impl Into) -> Self { + self.standard_headers.content_encoding = Some(value.into()); + self + } + + pub fn content_language(mut self, value: impl Into) -> Self { + self.standard_headers.content_language = Some(value.into()); + self + } + + /// Verbatim `Expires` header value. + pub fn expires(mut self, value: impl Into) -> Self { + self.standard_headers.expires = Some(value.into()); + self + } + + /// Fixed `Last-Modified` instead of the seeding time. + pub fn last_modified(mut self, value: SystemTime) -> Self { + self.last_modified = Some(Timestamp::from(value)); + self + } +} + +/// Resolved storage limits derived from [`FakeS3TargetOptions`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct StoreLimits { + max_object_bytes: usize, + max_total_bytes: usize, +} + +impl Default for StoreLimits { + fn default() -> Self { + Self::from_options(&FakeS3TargetOptions::default()) + } +} + +impl StoreLimits { + fn from_options(options: &FakeS3TargetOptions) -> Self { + assert!( + (1..=MAX_CONFIGURABLE_OBJECT_BYTES).contains(&options.max_object_bytes), + "fake target object cap must be between 1 byte and 256 MiB" + ); + Self { + max_object_bytes: options.max_object_bytes, + max_total_bytes: MAX_TOTAL_STORED_BYTES.max(options.max_object_bytes.saturating_mul(2)), + } + } +} + /// One fault consumed by the next matching operation. #[derive(Debug, Clone, PartialEq, Eq)] pub enum FaultAction { /// Return the exact HTTP status without entering the S3 backend. Status(StatusCode), + /// Return an arbitrary 4xx/5xx status without entering the S3 backend, + /// with the S3 error code a real service would pair with it (404 + /// `NoSuchKey`, 429 `SlowDown`, 500 `InternalError`, ...). + ResponseStatus(u16), + /// GetObject only: announce the full `Content-Length`, send the first N + /// body bytes, then abort the connection so the client observes a short + /// read. Ignored by every other operation. + TruncateBodyAt(usize), + /// Apply the request normally, then hold the complete response (status + /// line included) for the duration before the first byte is written — + /// the first-byte-timeout scenario. Error responses are not held. + Stall(Duration), /// Wait before dispatching the request normally. Delay(Duration), /// Close once the body stream reaches this logical byte threshold. Hyper may @@ -212,9 +350,44 @@ pub struct RequestRecord { pub consumed_bytes: Option, pub replication_timestamps: ReplicationTimestampHeaders, pub proxy_headers: ProxyHeaderSnapshot, + /// Verbatim `Range` request header, so range-forwarding tests can pin the + /// exact wire syntax a migrating server sent to its source. + pub range: Option, + pub user_agent: Option, + /// ListObjectsV2 `prefix` query value. + pub prefix: Option, + /// ListObjectsV2 `continuation-token` query value, journaled so resumed + /// listings can be told apart from restarted ones. + pub continuation_token: Option, pub fault: Option, } +/// Request headers journaled with every record, captured before the fault +/// script is consulted. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +struct JournaledHeaders { + content_length: Option, + replication_timestamps: ReplicationTimestampHeaders, + proxy_headers: ProxyHeaderSnapshot, + range: Option, + user_agent: Option, +} + +impl JournaledHeaders { + fn from_headers(headers: &HeaderMap) -> Self { + Self { + content_length: headers + .get(CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse().ok()), + replication_timestamps: ReplicationTimestampHeaders::from_headers(headers), + proxy_headers: ProxyHeaderSnapshot::from_headers(headers), + range: header_value(headers, &[RANGE.as_str()]).map(bounded_journal_value), + user_agent: header_value(headers, &[USER_AGENT.as_str()]).map(bounded_journal_value), + } + } +} + #[derive(Default)] struct ControlState { scripts: HashMap>, @@ -231,6 +404,7 @@ struct StoreState { /// transport headers instead of storing them (see /// [`REPLICATION_SSE_TRANSPORT_PREFIX`]). drop_unlisted_replication_headers: bool, + limits: StoreLimits, buckets: HashMap, uploads: HashMap, total_bytes: usize, @@ -238,11 +412,44 @@ struct StoreState { total_parts: usize, } -#[derive(Default)] struct BucketState { + versioned: bool, + /// Versions per key, newest first (see `upsert_version`); an unversioned + /// bucket holds exactly one version per key. objects: HashMap>, } +/// Standard HTTP object metadata accepted on PUT / CreateMultipartUpload and +/// replayed verbatim on HEAD / GET. +#[derive(Clone, Default)] +struct StandardHeaders { + cache_control: Option, + content_disposition: Option, + content_encoding: Option, + content_language: Option, + expires: Option, +} + +impl StandardHeaders { + fn validate(&self) -> S3Result { + let too_long = [ + &self.cache_control, + &self.content_disposition, + &self.content_encoding, + &self.content_language, + &self.expires, + ] + .into_iter() + .flatten() + .any(|value| value.len() > MAX_CONTENT_TYPE_BYTES); + if too_long { + Err(s3s::s3_error!(InvalidArgument, "standard object headers exceed 1024 bytes")) + } else { + Ok(()) + } + } +} + #[derive(Clone)] struct ObjectVersion { version_id: String, @@ -252,6 +459,7 @@ struct ObjectVersion { delete_marker: bool, content_type: Option, metadata: Option>, + standard_headers: StandardHeaders, /// Object tags as ordered key/value pairs (PutObjectTagging replaces the /// whole set, DeleteObjectTagging clears it). tags: Vec<(String, String)>, @@ -267,6 +475,7 @@ struct MultipartState { version_id: String, content_type: Option, metadata: Option>, + standard_headers: StandardHeaders, replication_sse_headers: Vec<(String, String)>, parts: BTreeMap, } @@ -337,12 +546,18 @@ pub struct FakeS3Target { } impl FakeS3Target { - /// Bind a new target on a random loopback port. + /// Bind a new target on a random loopback port with the default limits. pub async fn start() -> Result { - Self::start_with_connection_gate(MAX_CONNECTION_DURATION, None).await + Self::start_with_options(FakeS3TargetOptions::default()).await + } + + /// Bind a new target with explicit limits (see [`FakeS3TargetOptions`]). + pub async fn start_with_options(options: FakeS3TargetOptions) -> Result { + Self::start_with_connection_gate(StoreLimits::from_options(&options), MAX_CONNECTION_DURATION, None).await } async fn start_with_connection_gate( + limits: StoreLimits, connection_duration: Duration, connection_gate: Option>, ) -> Result { @@ -352,7 +567,10 @@ impl FakeS3Target { let control = Arc::new(Mutex::new(ControlState::default())); let body_limit = Arc::new(Semaphore::new(MAX_CONCURRENT_BODY_REQUESTS)); let backend = FakeBackend { - store: Arc::new(Mutex::new(StoreState::default())), + store: Arc::new(Mutex::new(StoreState { + limits, + ..Default::default() + })), control: Arc::clone(&control), body_limit: Arc::clone(&body_limit), }; @@ -438,16 +656,92 @@ impl FakeS3Target { self.address.to_string() } - /// Pre-create a general-purpose bucket in the shared global namespace. - /// Account-regional namespace buckets are intentionally not modeled. + /// Pre-create a versioned general-purpose bucket in the shared global + /// namespace. Account-regional namespace buckets are intentionally not + /// modeled. pub fn create_bucket(&self, bucket: impl Into) { + self.create_bucket_with_mode(bucket, BucketMode::Versioned); + } + + /// Pre-create a bucket with an explicit versioning mode. Re-creating an + /// existing bucket is a no-op, but its mode must match: the mode decides + /// how retained versions are addressed and cannot change afterwards. + pub fn create_bucket_with_mode(&self, bucket: impl Into, mode: BucketMode) { let bucket = bucket.into(); assert!(valid_bucket_name(&bucket), "fake target bucket name must be S3-valid"); + let versioned = mode == BucketMode::Versioned; let mut state = lock(&self.backend.store); - if !state.buckets.contains_key(&bucket) { - assert!(state.buckets.len() < MAX_BUCKETS, "fake target retains at most 256 buckets"); + match state.buckets.get(&bucket) { + Some(existing) => { + assert_eq!(existing.versioned, versioned, "fake target bucket mode cannot change: {bucket}"); + } + None => { + assert!(state.buckets.len() < MAX_BUCKETS, "fake target retains at most 256 buckets"); + state.buckets.insert( + bucket, + BucketState { + versioned, + objects: HashMap::new(), + }, + ); + } } - state.buckets.entry(bucket).or_default(); + } + + /// Store an object directly, bypassing the wire, the fault script, and + /// the request journal. Seeds a migration source without polluting the + /// journal that the scenario under test later asserts on. Returns the + /// ETag (hex MD5) the object will report. Same caps as a PUT. + pub fn put_seed_object(&self, bucket: &str, key: impl Into, body: impl Into, seed: &SeedMetadata) -> String { + let body = body.into(); + let e_tag = md5_hex(&body); + let mut state = lock(&self.backend.store); + assert!( + body.len() <= state.limits.max_object_bytes, + "seed object exceeds the fake target object cap" + ); + let content_type = seed.content_type.clone(); + let metadata = (!seed.metadata.is_empty()).then(|| { + seed.metadata + .iter() + .map(|(name, value)| (name.to_ascii_lowercase(), value.clone())) + .collect() + }); + validate_stored_metadata(&content_type, &metadata).expect("seed metadata must fit the retained caps"); + seed.standard_headers + .validate() + .expect("seed standard headers must fit the retained caps"); + let versioned = state.buckets.get(bucket).expect("fake target bucket must exist").versioned; + let version = ObjectVersion { + version_id: if versioned { + Uuid::new_v4().to_string() + } else { + NULL_VERSION_ID.to_string() + }, + body, + e_tag: e_tag.clone(), + last_modified: seed + .last_modified + .clone() + .unwrap_or_else(|| Timestamp::from(SystemTime::now())), + delete_marker: false, + content_type, + metadata, + standard_headers: seed.standard_headers.clone(), + tags: Vec::new(), + replication_sse_headers: Vec::new(), + }; + upsert_version(&mut state, bucket, key.into(), version).expect("seed object must fit the storage budget"); + e_tag + } + + /// Number of journaled requests for one operation on one exact key. + pub fn count_requests(&self, operation: Operation, key: &str) -> usize { + lock(&self.control) + .requests + .iter() + .filter(|record| record.operation == operation && record.key.as_deref() == Some(key)) + .count() } /// Remove all retained object versions while preserving the bucket. @@ -609,13 +903,28 @@ fn validate_fault_action(action: &FaultAction) { FaultAction::Delay(duration) if *duration > MAX_FAULT_DURATION => { panic!("fault delay must not exceed 30 seconds"); } + FaultAction::Stall(duration) if *duration > MAX_FAULT_DURATION => { + panic!("fault stall must not exceed 30 seconds"); + } FaultAction::SlowDrain { delay, .. } if *delay >= MAX_FAULT_DURATION => { panic!("slow-drain slice delay must be below 30 seconds"); } + FaultAction::ResponseStatus(code) if !(400..=599).contains(code) => { + panic!("scripted response status must be a 4xx or 5xx code"); + } _ => {} } } +/// The HTTP status a pre-dispatch status fault answers with, if any. +fn scripted_status(action: &FaultAction) -> Option { + match action { + FaultAction::Status(status) => Some(*status), + FaultAction::ResponseStatus(code) => StatusCode::from_u16(*code).ok(), + _ => None, + } +} + fn queued_fault_count(state: &ControlState) -> usize { state.scripts.values().map(VecDeque::len).sum::() + state.keyed_scripts.values().map(VecDeque::len).sum::() } @@ -632,28 +941,10 @@ impl S3Access for FaultAccess { let parsed = parse_request(context.method(), context.uri()); let operation = operation_from_s3_name(context.s3_op().name()); - let content_length = context - .headers() - .get(CONTENT_LENGTH) - .and_then(|value| value.to_str().ok()) - .and_then(|value| value.parse().ok()); - let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers()); - let proxy_headers = ProxyHeaderSnapshot::from_headers(context.headers()); - let fault = record_request( - &self.control, - operation, - context.method().clone(), - parsed, - content_length, - replication_timestamps, - proxy_headers, - ); - if let Some(RequestFault { - action: FaultAction::Status(status), - .. - }) = fault.as_ref() - { - return Err(scripted_status_error(*status)); + let journaled = JournaledHeaders::from_headers(context.headers()); + let fault = record_request(&self.control, operation, context.method().clone(), parsed, journaled); + if let Some(status) = fault.as_ref().and_then(|fault| scripted_status(&fault.action)) { + return Err(scripted_status_error(status)); } let prebody_permit = if operation == Operation::CompleteMultipartUpload { Some( @@ -689,6 +980,7 @@ fn operation_from_s3_name(name: &str) -> Operation { "GetObjectTagging" => Operation::GetObjectTagging, "PutObjectTagging" => Operation::PutObjectTagging, "DeleteObjectTagging" => Operation::DeleteObjectTagging, + "ListObjectsV2" => Operation::ListObjectsV2, "CreateMultipartUpload" => Operation::CreateMultipartUpload, "UploadPart" => Operation::UploadPart, "CompleteMultipartUpload" => Operation::CompleteMultipartUpload, @@ -702,9 +994,7 @@ fn record_request( operation: Operation, method: Method, parsed: ParsedRequest, - content_length: Option, - replication_timestamps: ReplicationTimestampHeaders, - proxy_headers: ProxyHeaderSnapshot, + headers: JournaledHeaders, ) -> Option { let mut state = lock(control); let action = parsed @@ -727,10 +1017,14 @@ fn record_request( version_id: parsed.version_id.map(bounded_journal_value), upload_id: parsed.upload_id.map(bounded_journal_value), part_number: parsed.part_number, - content_length, + content_length: headers.content_length, consumed_bytes: None, - replication_timestamps, - proxy_headers, + replication_timestamps: headers.replication_timestamps, + proxy_headers: headers.proxy_headers, + range: headers.range, + user_agent: headers.user_agent, + prefix: parsed.prefix.map(bounded_journal_value), + continuation_token: parsed.continuation_token.map(bounded_journal_value), fault: action.clone(), }); action.map(|action| RequestFault { sequence, action }) @@ -752,6 +1046,15 @@ async fn handle_request(request: Request, service: S3Service) -> Resul if response.headers_mut().remove(DISCONNECT_HEADER).is_some() { return Err(io::Error::new(io::ErrorKind::ConnectionAborted, "fake target scripted disconnect").into()); } + if let Some(stall_ms) = response + .headers_mut() + .remove(STALL_HEADER) + .and_then(|value| value.to_str().ok()?.parse::().ok()) + { + // The response is fully computed; nothing reaches the socket until + // the stall elapses, so the client sees no first byte at all. + sleep(Duration::from_millis(stall_ms)).await; + } Ok(response) } @@ -759,6 +1062,7 @@ async fn call_s3(service: S3Service, request: Request) -> Result, upload_id: Option, part_number: Option, + prefix: Option, + continuation_token: Option, } fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest { @@ -778,6 +1084,8 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest { let query = query_values(uri.query()); let version_id = query.get("versionId").cloned().flatten(); let upload_id = query.get("uploadId").cloned().flatten(); + let prefix = query.get("prefix").cloned().flatten(); + let continuation_token = query.get("continuation-token").cloned().flatten(); let part_number = query .get("partNumber") .and_then(Clone::clone) @@ -788,10 +1096,12 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest { .filter(|name| name.as_str() != "x-id") .all(|name| allowed.contains(&name.as_str())) }; + let list_type_v2 = query.get("list-type").is_some_and(|value| value.as_deref() == Some("2")); let operation = match (method, key.is_some()) { (&Method::HEAD, false) => Operation::HeadBucket, (&Method::GET, false) if query.contains_key("versioning") => Operation::GetBucketVersioning, (&Method::GET, false) if query.contains_key("versions") => Operation::ListObjectVersions, + (&Method::GET, false) if list_type_v2 => Operation::ListObjectsV2, (&Method::PUT, true) if upload_id.is_some() && part_number.is_some() => Operation::UploadPart, (&Method::PUT, true) if upload_id.is_some() || query.contains_key("partNumber") => Operation::Unknown, (&Method::POST, true) if query.contains_key("uploads") => Operation::CreateMultipartUpload, @@ -820,6 +1130,8 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest { version_id, upload_id, part_number, + prefix, + continuation_token, } } @@ -939,10 +1251,19 @@ fn request_fault(request: &S3Request) -> Option { request.extensions.get::().cloned() } +/// Pair a scripted HTTP status with the S3 error code a real service would +/// send, so SDK-side error classification matches production sources. fn scripted_status_error(status: StatusCode) -> s3s::S3Error { let mut error = match status { + StatusCode::BAD_REQUEST => s3s::s3_error!(InvalidRequest, "scripted fake target fault"), StatusCode::UNAUTHORIZED => s3s::s3_error!(UnauthorizedAccess, "scripted fake target fault"), StatusCode::FORBIDDEN => s3s::s3_error!(AccessDenied, "scripted fake target fault"), + StatusCode::NOT_FOUND => s3s::s3_error!(NoSuchKey, "scripted fake target fault"), + StatusCode::METHOD_NOT_ALLOWED => s3s::s3_error!(MethodNotAllowed, "scripted fake target fault"), + StatusCode::REQUEST_TIMEOUT => s3s::s3_error!(RequestTimeout, "scripted fake target fault"), + StatusCode::RANGE_NOT_SATISFIABLE => s3s::s3_error!(InvalidRange, "scripted fake target fault"), + StatusCode::TOO_MANY_REQUESTS => s3s::s3_error!(SlowDown, "scripted fake target fault"), + StatusCode::NOT_IMPLEMENTED => s3s::s3_error!(NotImplemented, "scripted fake target fault"), StatusCode::SERVICE_UNAVAILABLE => s3s::s3_error!(ServiceUnavailable, "scripted fake target fault"), _ => s3s::s3_error!(InternalError, "scripted fake target fault"), }; @@ -966,7 +1287,9 @@ fn update_consumed(control: &Mutex, sequence: u64, consumed: usize async fn apply_non_body_fault(fault: Option<&RequestFault>, control: &Mutex) -> S3Result<()> { match fault.map(|fault| &fault.action) { - Some(FaultAction::Status(status)) => Err(scripted_status_error(*status)), + Some(action @ (FaultAction::Status(_) | FaultAction::ResponseStatus(_))) => { + Err(scripted_status_error(scripted_status(action).expect("status faults carry a valid code"))) + } Some(FaultAction::Delay(duration)) => { sleep(*duration).await; Ok(()) @@ -978,6 +1301,8 @@ async fn apply_non_body_fault(fault: Option<&RequestFault>, control: &Mutex Ok(()), } } @@ -987,13 +1312,14 @@ async fn collect_stream( content_length: Option, fault: Option<&RequestFault>, control: &Mutex, + max_body_bytes: usize, ) -> S3Result { let mut body = body.unwrap_or_else(|| StreamingBlob::new(Body::empty())); let capacity = content_length .and_then(|length| usize::try_from(length).ok()) .unwrap_or_default(); - if capacity > MAX_BUFFERED_BODY_BYTES { - return Err(s3s::s3_error!(EntityTooLarge, "fake target buffers at most 64 MiB")); + if capacity > max_body_bytes { + return Err(body_too_large(max_body_bytes)); } if let Some(RequestFault { action: FaultAction::Delay(duration), @@ -1004,7 +1330,9 @@ async fn collect_stream( } timeout(MAX_FAULT_DURATION, async { match fault.map(|fault| &fault.action) { - Some(FaultAction::Status(status)) => return Err(scripted_status_error(*status)), + Some(action @ (FaultAction::Status(_) | FaultAction::ResponseStatus(_))) => { + return Err(scripted_status_error(scripted_status(action).expect("status faults carry a valid code"))); + } Some(FaultAction::Delay(_)) => {} Some(FaultAction::DisconnectAfterBytes(limit)) => { let mut consumed = 0usize; @@ -1017,15 +1345,19 @@ async fn collect_stream( return Err(scripted_disconnect_error()); } Some(FaultAction::SlowDrain { chunk_bytes, delay }) => { - return collect_stream_slow(body, capacity, *chunk_bytes, *delay).await; + return collect_stream_slow(body, capacity, *chunk_bytes, *delay, max_body_bytes).await; } - Some(FaultAction::WrongEtag) | Some(FaultAction::DisconnectAfterResponse) | None => {} + Some(FaultAction::WrongEtag) + | Some(FaultAction::DisconnectAfterResponse) + | Some(FaultAction::TruncateBodyAt(_)) + | Some(FaultAction::Stall(_)) + | None => {} } let mut output = BytesMut::with_capacity(capacity); while let Some(chunk) = body.next().await { let chunk = chunk.map_err(|error| s3s::s3_error!(InternalError, "request body failed: {error}"))?; - ensure_body_growth(output.len(), chunk.len())?; + ensure_body_growth(output.len(), chunk.len(), max_body_bytes)?; output.extend_from_slice(&chunk); } Ok(output.freeze()) @@ -1034,12 +1366,18 @@ async fn collect_stream( .map_err(|_| s3s::s3_error!(RequestTimeout, "fake target body drain exceeded 30 seconds"))? } -async fn collect_stream_slow(mut body: StreamingBlob, capacity: usize, chunk_bytes: usize, delay: Duration) -> S3Result { +async fn collect_stream_slow( + mut body: StreamingBlob, + capacity: usize, + chunk_bytes: usize, + delay: Duration, + max_body_bytes: usize, +) -> S3Result { let mut output = BytesMut::with_capacity(capacity); while let Some(chunk) = body.next().await { let chunk = chunk.map_err(|error| s3s::s3_error!(InternalError, "request body failed: {error}"))?; for slice in chunk.chunks(chunk_bytes) { - ensure_body_growth(output.len(), slice.len())?; + ensure_body_growth(output.len(), slice.len(), max_body_bytes)?; output.extend_from_slice(slice); sleep(delay).await; } @@ -1047,6 +1385,38 @@ async fn collect_stream_slow(mut body: StreamingBlob, capacity: usize, chunk_byt Ok(output.freeze()) } +fn body_too_large(max_body_bytes: usize) -> s3s::S3Error { + s3s::s3_error!(EntityTooLarge, "fake target buffers at most {max_body_bytes} bytes") +} + +/// GetObject body that announces `content_length` bytes but delivers only the +/// first `truncate_at`, then fails the stream so hyper aborts the connection +/// mid-body: the client observes a short read against the declared length. +fn truncated_body(body: Bytes, truncate_at: usize) -> StreamingBlob { + enum Step { + Prefix(Bytes), + Abort, + Done, + } + let prefix = body.slice(..truncate_at.min(body.len())); + // The error is delayed one tick so hyper flushes the head and the prefix + // before it sees the failure; an immediately-ready error aborts the + // connection before any byte reaches the client. + StreamingBlob::wrap(futures::stream::unfold(Step::Prefix(prefix), |step| async move { + match step { + Step::Prefix(prefix) => Some((Ok(prefix), Step::Abort)), + Step::Abort => { + sleep(Duration::from_millis(20)).await; + Some(( + Err(io::Error::new(io::ErrorKind::ConnectionAborted, "fake target scripted body truncation")), + Step::Done, + )) + } + Step::Done => None, + } + })) +} + async fn assemble_multipart( parts: Vec<(Bytes, [u8; 16])>, total_len: usize, @@ -1084,12 +1454,17 @@ fn apply_response_fault(mut response: S3Response, fault: Option<&RequestFa if fault.is_some_and(|fault| fault.action == FaultAction::DisconnectAfterResponse) { response.headers.insert(DISCONNECT_HEADER, HeaderValue::from_static("true")); } + if let Some(FaultAction::Stall(duration)) = fault.map(|fault| &fault.action) + && let Ok(value) = HeaderValue::from_str(&duration.as_millis().to_string()) + { + response.headers.insert(STALL_HEADER, value); + } response } -fn ensure_body_growth(current: usize, added: usize) -> S3Result { - if current.checked_add(added).is_none_or(|total| total > MAX_BUFFERED_BODY_BYTES) { - Err(s3s::s3_error!(EntityTooLarge, "fake target buffers at most 64 MiB")) +fn ensure_body_growth(current: usize, added: usize, max_body_bytes: usize) -> S3Result { + if current.checked_add(added).is_none_or(|total| total > max_body_bytes) { + Err(body_too_large(max_body_bytes)) } else { Ok(()) } @@ -1122,7 +1497,7 @@ fn ensure_store_budget(state: &StoreState, removed_bytes: usize, added_bytes: us .checked_sub(removed_bytes) .and_then(|total| total.checked_add(added_bytes)) .ok_or_else(|| s3s::s3_error!(InternalError, "fake target storage accounting overflow"))?; - if total_bytes > MAX_TOTAL_STORED_BYTES { + if total_bytes > state.limits.max_total_bytes { return Err(s3s::s3_error!(ServiceUnavailable, "fake target storage budget exhausted")); } if adds_version && state.total_versions >= MAX_OBJECT_VERSIONS { @@ -1147,19 +1522,25 @@ fn ensure_upload_budget(state: &StoreState) -> S3Result { } } +/// Insert a version newest-first. In a versioned bucket a version id is +/// idempotent (a replayed replication PUT replaces its own copy); in an +/// unversioned bucket the new version replaces whatever the key held. fn upsert_version(state: &mut StoreState, bucket: &str, key: String, version: ObjectVersion) -> S3Result { - let versions = state + let bucket_state = state .buckets .get(bucket) - .ok_or_else(|| s3s::s3_error!(NoSuchBucket, "bucket does not exist"))? - .objects - .get(&key); + .ok_or_else(|| s3s::s3_error!(NoSuchBucket, "bucket does not exist"))?; + let versioned = bucket_state.versioned; + let versions = bucket_state.objects.get(&key); let existing = versions.and_then(|versions| { versions .iter() .position(|candidate| candidate.version_id == version.version_id) .map(|position| (position, versions[position].body.len())) }); + // An unversioned bucket holds one version per key, so a replacement + // always matches the stored `null` id and `existing` covers it. + debug_assert!(versioned || versions.is_none_or(|versions| versions.len() <= 1)); ensure_store_budget(state, existing.map_or(0, |(_, bytes)| bytes), version.body.len(), existing.is_none())?; state.total_bytes = state.total_bytes - existing.map_or(0, |(_, bytes)| bytes) + version.body.len(); if existing.is_none() { @@ -1270,6 +1651,170 @@ fn set_version_tags( Ok(resolved) } +/// Whether version ids are surfaced for this bucket. Unknown buckets report +/// `true`; the caller's lookup raises `NoSuchBucket` first. +fn bucket_versioned(state: &StoreState, bucket: &str) -> bool { + state.buckets.get(bucket).is_none_or(|bucket| bucket.versioned) +} + +/// Bytes of the stored body a GET/HEAD answers with, plus the `Content-Range` +/// header and 206 status when a `Range` was applied. +struct ServedRange { + range: std::ops::Range, + content_range: Option, + status: Option, +} + +impl ServedRange { + fn len(&self) -> usize { + self.range.end - self.range.start + } +} + +/// Resolve a `Range` request header against the object length following RFC +/// 9110 byte ranges as S3 applies them: `first-last` (last clamped to the +/// object), `first-`, and `-suffix`; unsatisfiable ranges (first beyond the +/// end, zero suffix, empty object) answer 416 with `Content-Range: bytes */len`. +fn resolve_range(range: Option<&Range>, len: usize) -> S3Result { + let Some(range) = range else { + return Ok(ServedRange { + range: 0..len, + content_range: None, + status: None, + }); + }; + let unsatisfiable = || { + let mut error = s3s::s3_error!(InvalidRange, "the requested range is not satisfiable"); + let mut headers = HeaderMap::new(); + if let Ok(value) = HeaderValue::from_str(&format!("bytes */{len}")) { + headers.insert("content-range", value); + } + error.set_headers(headers); + error + }; + if len == 0 { + return Err(unsatisfiable()); + } + let served = range.check(len as u64).map_err(|_| unsatisfiable())?; + let start = served.start as usize; + let end = served.end as usize; + Ok(ServedRange { + range: start..end, + content_range: Some(format!("bytes {start}-{}/{len}", end - 1)), + status: Some(StatusCode::PARTIAL_CONTENT), + }) +} + +/// One ListObjectsV2 result entry in listing order. +enum ListEntry { + Object { + key: String, + e_tag: String, + last_modified: Timestamp, + size: usize, + }, + CommonPrefix(String), +} + +impl ListEntry { + fn name(&self) -> &str { + match self { + ListEntry::Object { key, .. } => key, + ListEntry::CommonPrefix(prefix) => prefix, + } + } +} + +struct ListPage { + entries: Vec, + truncated: bool, +} + +/// Listing cursor decoded from `start-after` / `continuation-token`. +struct ListCursor<'a> { + /// Keys at or before this name are skipped. + after: &'a str, + /// The name was a folded common prefix: every key under it is skipped + /// too, so the next page resumes after the whole group. + skips_group: bool, +} + +fn list_page( + bucket: &BucketState, + prefix: &str, + delimiter: Option<&str>, + cursors: &[ListCursor<'_>], + max_keys: usize, +) -> ListPage { + if max_keys == 0 { + // An empty page cannot carry a resume position, so it is never + // reported as truncated. + return ListPage { + entries: Vec::new(), + truncated: false, + }; + } + let mut current: Vec<(&String, &ObjectVersion)> = bucket + .objects + .iter() + .filter_map(|(key, versions)| { + versions + .first() + .filter(|version| !version.delete_marker) + .map(|version| (key, version)) + }) + .filter(|(key, _)| key.starts_with(prefix)) + .filter(|(key, _)| { + cursors + .iter() + .all(|cursor| key.as_str() > cursor.after && !(cursor.skips_group && key.starts_with(cursor.after))) + }) + .collect(); + current.sort_by(|left, right| left.0.as_bytes().cmp(right.0.as_bytes())); + + let mut entries: Vec = Vec::new(); + let mut truncated = false; + for (key, version) in current { + let folded = delimiter.and_then(|delimiter| { + key[prefix.len()..] + .find(delimiter) + .map(|position| key[..prefix.len() + position + delimiter.len()].to_string()) + }); + if let Some(common) = &folded + && entries.last().is_some_and(|entry| entry.name() == common) + { + continue; + } + if entries.len() >= max_keys { + truncated = true; + break; + } + entries.push(match folded { + Some(common) => ListEntry::CommonPrefix(common), + None => ListEntry::Object { + key: key.clone(), + e_tag: version.e_tag.clone(), + last_modified: version.last_modified.clone(), + size: version.body.len(), + }, + }); + } + ListPage { entries, truncated } +} + +/// Continuation tokens are opaque on the wire (hex of the last listed name) +/// so a client that inspects or edits them gets no positional hint. +fn encode_continuation_token(name: &str) -> String { + hex_simd::encode_to_string(name.as_bytes(), hex_simd::AsciiCase::Lower) +} + +fn decode_continuation_token(token: &str) -> S3Result { + hex_simd::decode_to_vec(token.as_bytes()) + .ok() + .and_then(|bytes| String::from_utf8(bytes).ok()) + .ok_or_else(|| s3s::s3_error!(InvalidArgument, "the continuation token provided is incorrect")) +} + #[async_trait] impl S3 for FakeBackend { async fn head_bucket(&self, req: S3Request) -> S3Result> { @@ -1287,12 +1832,99 @@ impl S3 for FakeBackend { ) -> S3Result> { let fault = request_fault(&req); apply_non_body_fault(fault.as_ref(), &self.control).await?; - if !lock(&self.store).buckets.contains_key(&req.input.bucket) { - return Err(s3s::s3_error!(NoSuchBucket, "bucket does not exist")); - } + let versioned = lock(&self.store) + .buckets + .get(&req.input.bucket) + .map(|bucket| bucket.versioned) + .ok_or_else(|| s3s::s3_error!(NoSuchBucket, "bucket does not exist"))?; Ok(apply_response_fault( S3Response::new(GetBucketVersioningOutput { - status: Some("Enabled".to_string().into()), + status: versioned.then(|| "Enabled".to_string().into()), + ..Default::default() + }), + fault.as_ref(), + )) + } + + /// Current versions only (a key whose newest version is a delete marker + /// is hidden), keys in byte order, `delimiter` folding into common + /// prefixes that count toward `max-keys`, and `start-after` / + /// `continuation-token` cursors. `encoding-type` and `fetch-owner` are + /// accepted but ignored. + async fn list_objects_v2(&self, req: S3Request) -> S3Result> { + let fault = request_fault(&req); + apply_non_body_fault(fault.as_ref(), &self.control).await?; + let input = req.input; + let prefix = input.prefix.clone().unwrap_or_default(); + let delimiter = input.delimiter.clone().filter(|delimiter| !delimiter.is_empty()); + let max_keys = input.max_keys.unwrap_or(MAX_LIST_KEYS).clamp(0, MAX_LIST_KEYS); + let resume_after = input + .continuation_token + .as_deref() + .map(decode_continuation_token) + .transpose()?; + let mut cursors = Vec::with_capacity(2); + if let Some(start_after) = input.start_after.as_deref() { + cursors.push(ListCursor { + after: start_after, + skips_group: false, + }); + } + if let Some(resume_after) = resume_after.as_deref() { + cursors.push(ListCursor { + after: resume_after, + skips_group: delimiter + .as_deref() + .is_some_and(|delimiter| resume_after.ends_with(delimiter)), + }); + } + let page = { + let state = lock(&self.store); + let bucket_state = state + .buckets + .get(&input.bucket) + .ok_or_else(|| s3s::s3_error!(NoSuchBucket, "bucket does not exist"))?; + list_page(bucket_state, &prefix, delimiter.as_deref(), &cursors, max_keys as usize) + }; + let next_continuation_token = page + .truncated + .then(|| page.entries.last().map(|entry| encode_continuation_token(entry.name()))) + .flatten(); + let key_count = page.entries.len() as i32; + let mut contents = Vec::new(); + let mut common_prefixes = Vec::new(); + for entry in page.entries { + match entry { + ListEntry::Object { + key, + e_tag, + last_modified, + size, + } => contents.push(Object { + key: Some(key), + e_tag: Some(ETag::Strong(e_tag)), + last_modified: Some(last_modified), + size: Some(size as i64), + storage_class: Some(ObjectStorageClass::STANDARD.to_string().into()), + ..Default::default() + }), + ListEntry::CommonPrefix(prefix) => common_prefixes.push(CommonPrefix { prefix: Some(prefix) }), + } + } + Ok(apply_response_fault( + S3Response::new(ListObjectsV2Output { + name: Some(input.bucket), + prefix: input.prefix, + delimiter: input.delimiter, + max_keys: Some(max_keys), + key_count: Some(key_count), + continuation_token: input.continuation_token, + start_after: input.start_after, + encoding_type: input.encoding_type, + is_truncated: Some(page.truncated), + next_continuation_token, + contents: (!contents.is_empty()).then_some(contents), + common_prefixes: (!common_prefixes.is_empty()).then_some(common_prefixes), ..Default::default() }), fault.as_ref(), @@ -1364,13 +1996,31 @@ impl S3 for FakeBackend { .map_err(|_| s3s::s3_error!(ServiceUnavailable, "fake target body limiter closed"))?; let headers = req.headers; let input = req.input; - let body = collect_stream(input.body, input.content_length, fault.as_ref(), &self.control).await?; - validate_stored_metadata(&input.content_type, &input.metadata)?; - let (assign_own, drop_unlisted) = { + let (assign_own, drop_unlisted, limits, versioned) = { let state = lock(&self.store); - (state.assign_own_version_ids, state.drop_unlisted_replication_headers) + ( + state.assign_own_version_ids, + state.drop_unlisted_replication_headers, + state.limits, + bucket_versioned(&state, &input.bucket), + ) + }; + let body = + collect_stream(input.body, input.content_length, fault.as_ref(), &self.control, limits.max_object_bytes).await?; + validate_stored_metadata(&input.content_type, &input.metadata)?; + let standard_headers = StandardHeaders { + cache_control: input.cache_control, + content_disposition: input.content_disposition, + content_encoding: input.content_encoding, + content_language: input.content_language, + expires: input.expires, + }; + standard_headers.validate()?; + let version_id = if versioned { + new_version_id(&headers, assign_own)? + } else { + NULL_VERSION_ID.to_string() }; - let version_id = new_version_id(&headers, assign_own)?; let e_tag = match source_etag(&headers)? { Some(value) => value, None => { @@ -1386,6 +2036,7 @@ impl S3 for FakeBackend { delete_marker: false, content_type: input.content_type, metadata: input.metadata, + standard_headers, tags: Vec::new(), replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted), }; @@ -1393,7 +2044,7 @@ impl S3 for FakeBackend { Ok(apply_response_fault( S3Response::new(PutObjectOutput { e_tag: Some(ETag::Strong(maybe_wrong_etag(fault.as_ref(), e_tag))), - version_id: Some(version_id), + version_id: versioned.then_some(version_id), ..Default::default() }), fault.as_ref(), @@ -1404,58 +2055,86 @@ impl S3 for FakeBackend { let fault = request_fault(&req); apply_non_body_fault(fault.as_ref(), &self.control).await?; let input = req.input; - let version = { + let (version, versioned) = { let state = lock(&self.store); - find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())? + ( + find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())?, + bucket_versioned(&state, &input.bucket), + ) }; let sse_customer_algorithm = stored_sse_customer_algorithm(&version); - Ok(apply_response_fault( - S3Response::new(GetObjectOutput { - body: Some(StreamingBlob::new(Body::from(version.body.clone()))), - content_length: Some(version.body.len() as i64), - content_type: version.content_type, - metadata: version.metadata, - e_tag: Some(ETag::Strong(version.e_tag)), - last_modified: Some(version.last_modified.clone()), - version_id: Some(version.version_id), - sse_customer_algorithm, - ..Default::default() - }), - fault.as_ref(), - )) + let served = resolve_range(input.range.as_ref(), version.body.len())?; + let body = version.body.slice(served.range.clone()); + let body = match fault.as_ref().map(|fault| &fault.action) { + Some(FaultAction::TruncateBodyAt(truncate_at)) => truncated_body(body, *truncate_at), + _ => StreamingBlob::from(body), + }; + let mut response = S3Response::new(GetObjectOutput { + body: Some(body), + content_length: Some(served.len() as i64), + content_range: served.content_range, + accept_ranges: Some("bytes".to_string()), + content_type: version.content_type, + metadata: version.metadata, + cache_control: version.standard_headers.cache_control, + content_disposition: version.standard_headers.content_disposition, + content_encoding: version.standard_headers.content_encoding, + content_language: version.standard_headers.content_language, + expires: version.standard_headers.expires, + e_tag: Some(ETag::Strong(version.e_tag)), + last_modified: Some(version.last_modified.clone()), + version_id: versioned.then_some(version.version_id), + sse_customer_algorithm, + ..Default::default() + }); + response.status = served.status; + Ok(apply_response_fault(response, fault.as_ref())) } async fn head_object(&self, req: S3Request) -> S3Result> { let fault = request_fault(&req); apply_non_body_fault(fault.as_ref(), &self.control).await?; let input = req.input; - let version = { + let (version, versioned) = { let state = lock(&self.store); - find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())? + ( + find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())?, + bucket_versioned(&state, &input.bucket), + ) }; let sse_customer_algorithm = stored_sse_customer_algorithm(&version); - Ok(apply_response_fault( - S3Response::new(HeadObjectOutput { - content_length: Some(version.body.len() as i64), - content_type: version.content_type, - metadata: version.metadata, - e_tag: Some(ETag::Strong(version.e_tag)), - last_modified: Some(version.last_modified.clone()), - version_id: Some(version.version_id), - sse_customer_algorithm, - ..Default::default() - }), - fault.as_ref(), - )) + let served = resolve_range(input.range.as_ref(), version.body.len())?; + let mut response = S3Response::new(HeadObjectOutput { + content_length: Some(served.len() as i64), + content_range: served.content_range, + accept_ranges: Some("bytes".to_string()), + content_type: version.content_type, + metadata: version.metadata, + cache_control: version.standard_headers.cache_control, + content_disposition: version.standard_headers.content_disposition, + content_encoding: version.standard_headers.content_encoding, + content_language: version.standard_headers.content_language, + expires: version.standard_headers.expires, + e_tag: Some(ETag::Strong(version.e_tag)), + last_modified: Some(version.last_modified.clone()), + version_id: versioned.then_some(version.version_id), + sse_customer_algorithm, + ..Default::default() + }); + response.status = served.status; + Ok(apply_response_fault(response, fault.as_ref())) } async fn get_object_tagging(&self, req: S3Request) -> S3Result> { let fault = request_fault(&req); apply_non_body_fault(fault.as_ref(), &self.control).await?; let input = req.input; - let version = { + let (version, versioned) = { let state = lock(&self.store); - find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())? + ( + find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())?, + bucket_versioned(&state, &input.bucket), + ) }; let tag_set: TagSet = version .tags @@ -1468,7 +2147,7 @@ impl S3 for FakeBackend { Ok(apply_response_fault( S3Response::new(GetObjectTaggingOutput { tag_set, - version_id: Some(ObjectVersionId::from(version.version_id)), + version_id: versioned.then(|| ObjectVersionId::from(version.version_id)), }), fault.as_ref(), )) @@ -1484,13 +2163,16 @@ impl S3 for FakeBackend { .into_iter() .map(|tag| (tag.key.unwrap_or_default(), tag.value.unwrap_or_default())) .collect(); - let version_id = { + let (version_id, versioned) = { let mut state = lock(&self.store); - set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), tags)? + ( + set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), tags)?, + bucket_versioned(&state, &input.bucket), + ) }; Ok(apply_response_fault( S3Response::new(PutObjectTaggingOutput { - version_id: Some(ObjectVersionId::from(version_id)), + version_id: versioned.then(|| ObjectVersionId::from(version_id)), }), fault.as_ref(), )) @@ -1503,13 +2185,16 @@ impl S3 for FakeBackend { let fault = request_fault(&req); apply_non_body_fault(fault.as_ref(), &self.control).await?; let input = req.input; - let version_id = { + let (version_id, versioned) = { let mut state = lock(&self.store); - set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), Vec::new())? + ( + set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), Vec::new())?, + bucket_versioned(&state, &input.bucket), + ) }; Ok(apply_response_fault( S3Response::new(DeleteObjectTaggingOutput { - version_id: Some(ObjectVersionId::from(version_id)), + version_id: versioned.then(|| ObjectVersionId::from(version_id)), }), fault.as_ref(), )) @@ -1521,8 +2206,29 @@ impl S3 for FakeBackend { let headers = req.headers; let input = req.input; let mut state = lock(&self.store); - if !state.buckets.contains_key(&input.bucket) { + let Some(versioned) = state.buckets.get(&input.bucket).map(|bucket| bucket.versioned) else { return Err(s3s::s3_error!(NoSuchBucket, "bucket does not exist")); + }; + if !versioned { + // Unversioned buckets know only the `null` version: DELETE removes + // the key outright and never writes a delete marker. + if input + .version_id + .as_deref() + .is_some_and(|version_id| version_id != NULL_VERSION_ID) + { + return Err(s3s::s3_error!(InvalidArgument, "invalid version id specified")); + } + let removed = state + .buckets + .get_mut(&input.bucket) + .expect("bucket existence checked above") + .objects + .remove(&input.key) + .unwrap_or_default(); + state.total_bytes -= removed.iter().map(|version| version.body.len()).sum::(); + state.total_versions -= removed.len(); + return Ok(apply_response_fault(S3Response::new(DeleteObjectOutput::default()), fault.as_ref())); } if let Some(version_id) = input.version_id { let (removed_bytes, removed_versions, delete_marker, remove_key) = { @@ -1591,6 +2297,7 @@ impl S3 for FakeBackend { delete_marker: true, content_type: None, metadata: None, + standard_headers: StandardHeaders::default(), tags: Vec::new(), replication_sse_headers: Vec::new(), }, @@ -1619,12 +2326,24 @@ impl S3 for FakeBackend { } ensure_upload_budget(&state)?; validate_stored_metadata(&input.content_type, &input.metadata)?; + let standard_headers = StandardHeaders { + cache_control: input.cache_control, + content_disposition: input.content_disposition, + content_encoding: input.content_encoding, + content_language: input.content_language, + expires: input.expires, + }; + standard_headers.validate()?; let upload_id = Uuid::new_v4().to_string(); // Read the flags before the mutable borrow of `state.uploads` below // (and never re-lock the store: the mutex is not reentrant). let mint_own = state.assign_own_version_ids || state.assign_own_multipart_version_ids; let drop_unlisted = state.drop_unlisted_replication_headers; - let version_id = new_version_id(&headers, mint_own)?; + let version_id = if bucket_versioned(&state, &input.bucket) { + new_version_id(&headers, mint_own)? + } else { + NULL_VERSION_ID.to_string() + }; state.uploads.insert( upload_id.clone(), MultipartState { @@ -1633,6 +2352,7 @@ impl S3 for FakeBackend { version_id, content_type: input.content_type, metadata: input.metadata, + standard_headers, replication_sse_headers: captured_replication_sse_headers(&headers, drop_unlisted), parts: BTreeMap::new(), }, @@ -1658,7 +2378,8 @@ impl S3 for FakeBackend { if !(1..=10_000).contains(&input.part_number) { return Err(s3s::s3_error!(InvalidArgument, "part number must be in 1..=10000")); } - let body = collect_stream(input.body, input.content_length, fault.as_ref(), &self.control).await?; + let max_object_bytes = lock(&self.store).limits.max_object_bytes; + let body = collect_stream(input.body, input.content_length, fault.as_ref(), &self.control, max_object_bytes).await?; let (digest, _body_permit) = md5_digest(body.clone(), _body_permit).await?; let e_tag = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower); let mut state = lock(&self.store); @@ -1771,17 +2492,19 @@ impl S3 for FakeBackend { version_id: upload.version_id.clone(), content_type: upload.content_type.clone(), metadata: upload.metadata.clone(), + standard_headers: upload.standard_headers.clone(), replication_sse_headers: upload.replication_sse_headers.clone(), parts: BTreeMap::new(), }, selected, ) }; + let max_total_bytes = lock(&self.store).limits.max_total_bytes; let total_len = selected_parts.iter().try_fold(0usize, |total, (_, part)| { total .checked_add(part.body.len()) - .filter(|total| *total <= MAX_TOTAL_STORED_BYTES) - .ok_or_else(|| s3s::s3_error!(EntityTooLarge, "multipart object exceeds 128 MiB")) + .filter(|total| *total <= max_total_bytes) + .ok_or_else(|| s3s::s3_error!(EntityTooLarge, "multipart object exceeds {max_total_bytes} bytes")) })?; let assembly_parts = selected_parts .iter() @@ -1798,10 +2521,12 @@ impl S3 for FakeBackend { delete_marker: false, content_type: upload.content_type, metadata: upload.metadata, + standard_headers: upload.standard_headers, tags: Vec::new(), replication_sse_headers: upload.replication_sse_headers, }; let mut state = lock(&self.store); + let versioned = bucket_versioned(&state, &input.bucket); let current = state .uploads .get(&input.upload_id) @@ -1843,7 +2568,7 @@ impl S3 for FakeBackend { bucket: Some(input.bucket), key: Some(input.key), e_tag: Some(ETag::Strong(maybe_wrong_etag(fault.as_ref(), e_tag))), - version_id: Some(upload.version_id), + version_id: versioned.then_some(upload.version_id), ..Default::default() }), fault.as_ref(), @@ -3201,7 +3926,8 @@ mod tests { let connection_duration = Duration::from_millis(200); let (connection_gate, gate_rx) = watch::channel(false); - let bounded_target = FakeS3Target::start_with_connection_gate(connection_duration, Some(gate_rx)).await?; + let bounded_target = + FakeS3Target::start_with_connection_gate(StoreLimits::default(), connection_duration, Some(gate_rx)).await?; bounded_target.create_bucket("target-bucket"); let bounded_client = client_with_credentials(&bounded_target, FAKE_ACCESS_KEY, FAKE_SECRET_KEY); let mut stalled_connections = Vec::with_capacity(MAX_CONNECTIONS); @@ -3230,8 +3956,9 @@ mod tests { fn slow_drain_rejects_zero_chunk_size() { let runtime = tokio::runtime::Runtime::new().expect("runtime"); runtime.block_on(async { - assert!(ensure_body_growth(MAX_BUFFERED_BODY_BYTES, 0).is_ok()); - let too_large = ensure_body_growth(MAX_BUFFERED_BODY_BYTES, 1).expect_err("64 MiB + 1 must be rejected"); + assert!(ensure_body_growth(MAX_BUFFERED_BODY_BYTES, 0, MAX_BUFFERED_BODY_BYTES).is_ok()); + let too_large = + ensure_body_growth(MAX_BUFFERED_BODY_BYTES, 1, MAX_BUFFERED_BODY_BYTES).expect_err("64 MiB + 1 must be rejected"); assert_eq!(too_large.code().as_str(), "EntityTooLarge"); assert!(validate_retained_identifier("v".repeat(MAX_RETAINED_IDENTIFIER_BYTES), "version").is_ok()); let identifier_overflow = validate_retained_identifier("v".repeat(MAX_RETAINED_IDENTIFIER_BYTES + 1), "version") @@ -3266,20 +3993,32 @@ mod tests { Some(MAX_BUFFERED_BODY_BYTES as i64), None, &body_control, + MAX_BUFFERED_BODY_BYTES, ) .await .expect("exact body cap must be accepted"); assert_eq!(collected.len(), MAX_BUFFERED_BODY_BYTES); drop(collected); let over_body = Bytes::from(vec![0; MAX_BUFFERED_BODY_BYTES + 1]); - let normal_overflow = - collect_stream(Some(StreamingBlob::new(Body::from(over_body.clone()))), None, None, &body_control) - .await - .expect_err("normal drain must enforce the body cap"); + let normal_overflow = collect_stream( + Some(StreamingBlob::new(Body::from(over_body.clone()))), + None, + None, + &body_control, + MAX_BUFFERED_BODY_BYTES, + ) + .await + .expect_err("normal drain must enforce the body cap"); assert_eq!(normal_overflow.code().as_str(), "EntityTooLarge"); - let slow_overflow = collect_stream_slow(StreamingBlob::new(Body::from(over_body)), 0, 1024 * 1024, Duration::ZERO) - .await - .expect_err("slow drain must enforce the body cap"); + let slow_overflow = collect_stream_slow( + StreamingBlob::new(Body::from(over_body)), + 0, + 1024 * 1024, + Duration::ZERO, + MAX_BUFFERED_BODY_BYTES, + ) + .await + .expect_err("slow drain must enforce the body cap"); assert_eq!(slow_overflow.code().as_str(), "EntityTooLarge"); let full = StoreState { total_bytes: MAX_TOTAL_STORED_BYTES, @@ -3328,6 +4067,7 @@ mod tests { version_id: index.to_string(), content_type: None, metadata: None, + standard_headers: StandardHeaders::default(), replication_sse_headers: Vec::new(), parts: BTreeMap::new(), }, @@ -3345,13 +4085,12 @@ mod tests { operation: Operation::PutObject, bucket: "bucket".to_string(), key: Some(format!("key-{index}")), - version_id: None, - upload_id: None, - part_number: None, + ..Default::default() + }, + JournaledHeaders { + content_length: Some(0), + ..Default::default() }, - Some(0), - ReplicationTimestampHeaders::default(), - ProxyHeaderSnapshot::default(), ); } let records = lock(&control).requests.clone(); @@ -3370,11 +4109,9 @@ mod tests { key: Some(utf8_boundary), version_id: Some("v".repeat(MAX_RETAINED_IDENTIFIER_BYTES + 1)), upload_id: Some("u".repeat(MAX_RETAINED_IDENTIFIER_BYTES + 1)), - part_number: None, + ..Default::default() }, - None, - ReplicationTimestampHeaders::default(), - ProxyHeaderSnapshot::default(), + JournaledHeaders::default(), ); { let bounded_records = lock(&bounded_control); diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index e4720c060..b364c44a6 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -23,10 +23,17 @@ pub mod common; #[cfg(test)] pub mod chaos; -// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8). +// Programmable S3 target for replication failure-path tests (backlog#1147 repl-8) +// and on-demand-migration source scenarios (backlog#2151). #[cfg(test)] pub mod fake_s3_target; +// On-demand migration (backlog#2147): shared two-server environment, admin +// wrappers, and the harness self-test (backlog#2151). Behavior scenarios are +// added by later ODM tasks. +#[cfg(test)] +pub mod on_demand_migration; + // Socket-level network fault-injection proxy for black-box cluster tests // (backlog#1325 network fault-injection block): latency / blackhole / one-way // partition on the wire between nodes. Serves #1312/#1319 (lock-plane one-way diff --git a/crates/e2e_test/src/on_demand_migration/common.rs b/crates/e2e_test/src/on_demand_migration/common.rs new file mode 100644 index 000000000..7405a2650 --- /dev/null +++ b/crates/e2e_test/src/on_demand_migration/common.rs @@ -0,0 +1,452 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Shared environment for on-demand migration (ODM) end-to-end tests. +//! +//! [`OdmTestEnv`] pairs one RustFS server under test with one in-process +//! programmable S3 source ([`FakeS3Target`]). Admin calls target the route +//! convention fixed by the tracking plan +//! (`/rustfs/admin/v3/on-demand-migration/{bucket}`, JSON bodies); the +//! server side lands with ODM-07, so until then the wrappers compile but are +//! not exercised by the harness self-test. + +use crate::common::{RustFSTestEnvironment, signed_request}; +use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FakeS3TargetOptions, SeedMetadata}; +use aws_config::retry::RetryConfig; +use aws_sdk_s3::Client; +use aws_sdk_s3::config::{Credentials, Region}; +use aws_smithy_http_client::Builder as SmithyHttpClientBuilder; +use bytes::Bytes; +use serde::Serialize; +use std::fmt; + +pub type BoxError = Box; + +/// Module switch the server reads at startup (`false` before GA). The harness +/// turns it on so scenario tests exercise the feature without repeating it. +pub const ODM_MODULE_SWITCH_ENV: &str = "RUSTFS_ON_DEMAND_MIGRATION_ENABLED"; +/// Admin route prefix; the bucket name is appended as one path segment. +pub const ODM_ADMIN_ROUTE: &str = "/rustfs/admin/v3/on-demand-migration"; +/// Region the fake source is addressed with (it accepts any SigV4 region). +pub const FAKE_SOURCE_REGION: &str = "us-east-1"; + +/// Wire form of the bucket-level ODM configuration (ODM-01 model). Every +/// field is public so a scenario can tweak one knob and serialize the rest +/// with the documented defaults. +#[derive(Debug, Clone, Serialize)] +pub struct OdmSourceSpec { + pub version: u32, + pub enabled: bool, + pub source: OdmSource, + pub filter: OdmFilter, + pub policy: OdmPolicy, +} + +#[derive(Debug, Clone, Serialize)] +pub struct OdmSource { + pub provider: String, + pub endpoint: String, + pub region: String, + pub bucket: String, + pub path_style: String, + pub credentials: Option, + pub tls: OdmTls, +} + +#[derive(Clone, Serialize)] +pub struct OdmCredentials { + pub access_key: String, + pub secret_key: String, + pub session_token: Option, +} + +impl fmt::Debug for OdmCredentials { + /// Test logs are captured into CI artifacts; keep the secret out of them. + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OdmCredentials") + .field("access_key", &self.access_key) + .field("secret_key", &"REDACTED") + .field("session_token", &self.session_token.as_ref().map(|_| "REDACTED")) + .finish() + } +} + +#[derive(Debug, Clone, Default, Serialize)] +pub struct OdmTls { + pub skip_verify: bool, + pub ca_cert_pem: Option, +} + +#[derive(Debug, Clone, Default, Serialize)] +pub struct OdmFilter { + pub prefix: Option, + pub source_prefix: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct OdmPolicy { + pub head: String, + pub range_get: String, + pub source_error: String, + pub respect_local_delete_marker: bool, + pub preserve_etag: bool, + pub copy_tags: bool, + pub emit_events: bool, + pub negative_cache_ttl_secs: u64, + pub inline_max_bytes: u64, + pub multipart_part_size_bytes: u64, + pub max_concurrent_pulls: u32, + pub pull_queue_capacity: u32, + pub source_timeout: OdmSourceTimeout, + pub bandwidth_limit_bytes_per_sec: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct OdmSourceTimeout { + pub connect_ms: u64, + pub first_byte_ms: u64, + pub idle_ms: u64, +} + +impl Default for OdmPolicy { + /// The ODM-01 defaults verbatim. + fn default() -> Self { + Self { + head: "proxy".to_string(), + range_get: "serve_and_backfill".to_string(), + source_error: "propagate".to_string(), + respect_local_delete_marker: true, + preserve_etag: true, + copy_tags: false, + emit_events: true, + negative_cache_ttl_secs: 30, + inline_max_bytes: 16 * 1024 * 1024, + multipart_part_size_bytes: 64 * 1024 * 1024, + max_concurrent_pulls: 8, + pull_queue_capacity: 1024, + source_timeout: OdmSourceTimeout { + connect_ms: 5_000, + first_byte_ms: 15_000, + idle_ms: 30_000, + }, + bandwidth_limit_bytes_per_sec: None, + } + } +} + +impl OdmSourceSpec { + /// Enabled configuration pointing at a bucket on the fake source with the + /// fixture credentials, path-style addressing, and default policy. + pub fn for_fake_source(source: &FakeS3Target, source_bucket: impl Into) -> Self { + Self::new( + "s3", + source.endpoint(), + FAKE_SOURCE_REGION, + source_bucket, + FAKE_ACCESS_KEY, + FAKE_SECRET_KEY, + ) + } + + /// Enabled configuration pointing at a bucket on a second RustFS server + /// (see [`start_source_rustfs`]). + pub fn for_rustfs_source(source: &RustFSTestEnvironment, source_bucket: impl Into) -> Self { + Self::new( + "rustfs", + &source.url, + FAKE_SOURCE_REGION, + source_bucket, + &source.access_key, + &source.secret_key, + ) + } + + fn new( + provider: &str, + endpoint: &str, + region: &str, + source_bucket: impl Into, + access_key: &str, + secret_key: &str, + ) -> Self { + Self { + version: 1, + enabled: true, + source: OdmSource { + provider: provider.to_string(), + endpoint: endpoint.to_string(), + region: region.to_string(), + bucket: source_bucket.into(), + path_style: "path".to_string(), + credentials: Some(OdmCredentials { + access_key: access_key.to_string(), + secret_key: secret_key.to_string(), + session_token: None, + }), + tls: OdmTls::default(), + }, + filter: OdmFilter::default(), + policy: OdmPolicy::default(), + } + } + + pub fn to_json(&self) -> serde_json::Value { + serde_json::to_value(self).expect("ODM source spec serializes") + } +} + +/// Backfill job control (ODM-12 route shape). +#[derive(Debug, Clone)] +pub enum BackfillOp { + Start(BackfillRequest), + Cancel, + Status, +} + +#[derive(Debug, Clone, Default, Serialize)] +pub struct BackfillRequest { + pub prefix: Option, + pub skip_existing: Option, + pub dry_run: bool, +} + +/// Status plus raw body of an admin call, so a scenario can assert on the +/// HTTP status first and only then parse the JSON. +#[derive(Debug, Clone)] +pub struct AdminResponse { + pub status: u16, + pub body: String, +} + +impl AdminResponse { + pub fn json(&self) -> Result { + Ok(serde_json::from_str(&self.body)?) + } +} + +/// One object to seed into the source. +#[derive(Clone)] +pub struct SeedObject { + pub key: String, + pub body: Bytes, + pub metadata: SeedMetadata, +} + +impl SeedObject { + pub fn new(key: impl Into, body: impl Into) -> Self { + Self { + key: key.into(), + body: body.into(), + metadata: SeedMetadata::new(), + } + } + + pub fn with_metadata(mut self, metadata: SeedMetadata) -> Self { + self.metadata = metadata; + self + } +} + +/// RustFS under test plus its fake S3 source. +pub struct OdmTestEnv { + pub rustfs: RustFSTestEnvironment, + pub source: FakeS3Target, + /// S3 client for the RustFS under test. + pub client: Client, +} + +impl OdmTestEnv { + /// Start a fake source with default limits and a RustFS server with the + /// ODM module switch enabled. + pub async fn start() -> Result { + Self::start_with_options(FakeS3TargetOptions::default()).await + } + + pub async fn start_with_options(options: FakeS3TargetOptions) -> Result { + let source = FakeS3Target::start_with_options(options).await?; + let mut rustfs = RustFSTestEnvironment::new().await?; + rustfs + .start_rustfs_server_with_env(vec![], &[(ODM_MODULE_SWITCH_ENV, "true")]) + .await?; + let client = rustfs.create_s3_client(); + Ok(Self { rustfs, source, client }) + } + + /// S3 client addressing the fake source directly, for assertions on the + /// source's own state. Retries are off so a scripted fault is consumed by + /// exactly the request the test issued. + pub fn source_client(&self) -> Client { + fake_source_client(&self.source) + } + + /// Enabled ODM configuration for `source_bucket` on the fake source. + pub fn fake_source_spec(&self, source_bucket: impl Into) -> OdmSourceSpec { + OdmSourceSpec::for_fake_source(&self.source, source_bucket) + } + + /// `PUT /rustfs/admin/v3/on-demand-migration/{bucket}` with the JSON spec. + pub async fn configure_source(&self, bucket: &str, spec: &OdmSourceSpec) -> Result { + self.admin(http::Method::PUT, &format!("/{bucket}"), Some(spec.to_json())) + .await + } + + /// Same as [`Self::configure_source`] with `dry-run=true`: validate and + /// probe without persisting. + pub async fn validate_source(&self, bucket: &str, spec: &OdmSourceSpec) -> Result { + self.admin(http::Method::PUT, &format!("/{bucket}?dry-run=true"), Some(spec.to_json())) + .await + } + + /// `GET .../{bucket}`: redacted configuration, 404 when unconfigured. + pub async fn get_config(&self, bucket: &str) -> Result { + self.admin(http::Method::GET, &format!("/{bucket}"), None).await + } + + /// `DELETE .../{bucket}`: remove the configuration (idempotent). + pub async fn disable(&self, bucket: &str) -> Result { + self.admin(http::Method::DELETE, &format!("/{bucket}"), None).await + } + + /// `GET .../{bucket}/status`: runtime snapshot. + pub async fn status(&self, bucket: &str) -> Result { + self.admin(http::Method::GET, &format!("/{bucket}/status"), None).await + } + + /// Backfill control: `POST .../{bucket}/backfill?op=start|cancel` or + /// `GET .../{bucket}/backfill` for the checkpoint. + pub async fn backfill(&self, bucket: &str, op: BackfillOp) -> Result { + match op { + BackfillOp::Start(request) => { + self.admin( + http::Method::POST, + &format!("/{bucket}/backfill?op=start"), + Some(serde_json::to_value(request)?), + ) + .await + } + BackfillOp::Cancel => { + self.admin(http::Method::POST, &format!("/{bucket}/backfill?op=cancel"), None) + .await + } + BackfillOp::Status => self.admin(http::Method::GET, &format!("/{bucket}/backfill"), None).await, + } + } + + async fn admin( + &self, + method: http::Method, + path_and_query: &str, + body: Option, + ) -> Result { + let url = format!("{}{ODM_ADMIN_ROUTE}{path_and_query}", self.rustfs.url); + let body = body.map(|value| serde_json::to_vec(&value)).transpose()?; + let content_type = body.is_some().then_some("application/json"); + let response = signed_request(method, &url, &self.rustfs.access_key, &self.rustfs.secret_key, body, content_type).await?; + Ok(AdminResponse { + status: response.status().as_u16(), + body: response.text().await?, + }) + } + + /// Store objects directly in the fake source (no wire traffic, no journal + /// entries). Returns the ETags in input order. + pub fn seed_source(&self, source_bucket: &str, objects: &[SeedObject]) -> Vec { + objects + .iter() + .map(|object| { + self.source + .put_seed_object(source_bucket, object.key.clone(), object.body.clone(), &object.metadata) + }) + .collect() + } + + /// Whether `key` is listed by the RustFS under test. Listing is served from + /// local state only, so this does not trigger a migration the way GET or + /// HEAD would. + pub async fn local_key_listed(&self, bucket: &str, key: &str) -> Result { + let listed = self + .client + .list_objects_v2() + .bucket(bucket) + .prefix(key) + .max_keys(1) + .send() + .await?; + Ok(listed.contents().iter().any(|object| object.key() == Some(key))) + } + + /// Panics unless `key` is stored locally with exactly `expected` bytes. + /// Presence is checked through listing first so a missing object fails + /// here instead of being pulled from the source by the GET. + pub async fn assert_local_present(&self, bucket: &str, key: &str, expected: &[u8]) { + assert!( + self.local_key_listed(bucket, key) + .await + .unwrap_or_else(|error| panic!("listing {bucket}/{key} failed: {error}")), + "{bucket}/{key} must be present locally" + ); + let body = self + .client + .get_object() + .bucket(bucket) + .key(key) + .send() + .await + .unwrap_or_else(|error| panic!("GET {bucket}/{key} failed: {error}")) + .body + .collect() + .await + .unwrap_or_else(|error| panic!("reading {bucket}/{key} failed: {error}")) + .into_bytes(); + assert_eq!(body.as_ref(), expected, "{bucket}/{key} local content mismatch"); + } + + /// Panics if `key` is listed locally. + pub async fn assert_local_absent(&self, bucket: &str, key: &str) { + assert!( + !self + .local_key_listed(bucket, key) + .await + .unwrap_or_else(|error| panic!("listing {bucket}/{key} failed: {error}")), + "{bucket}/{key} must be absent locally" + ); + } +} + +/// S3 client for the fake source with retries disabled (see +/// [`OdmTestEnv::source_client`]). +pub fn fake_source_client(source: &FakeS3Target) -> Client { + let credentials = Credentials::new(FAKE_ACCESS_KEY, FAKE_SECRET_KEY, None, None, "odm-fake-source"); + Client::from_conf( + aws_sdk_s3::Config::builder() + .credentials_provider(credentials) + .region(Region::new(FAKE_SOURCE_REGION)) + .endpoint_url(source.endpoint()) + .force_path_style(true) + .behavior_version_latest() + .retry_config(RetryConfig::standard().with_max_attempts(1)) + .http_client(SmithyHttpClientBuilder::new().build_http()) + .build(), + ) +} + +/// Start a second, fully independent RustFS process (own port, data +/// directory, and default credentials) to act as a real S3 source. It is +/// spawned the same way `reliant::tiering` starts its cold tier; the process +/// is stopped and its directory removed when the returned environment drops. +pub async fn start_source_rustfs() -> Result { + let mut source = RustFSTestEnvironment::new().await?; + source.start_rustfs_server_without_cleanup(vec![]).await?; + Ok(source) +} diff --git a/crates/e2e_test/src/on_demand_migration/harness_self_test.rs b/crates/e2e_test/src/on_demand_migration/harness_self_test.rs new file mode 100644 index 000000000..6c19bc7e5 --- /dev/null +++ b/crates/e2e_test/src/on_demand_migration/harness_self_test.rs @@ -0,0 +1,606 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Self-test of the ODM harness (rustfs/backlog#2151): the fake source's +//! migration-facing surface (ListObjectsV2 paging, `Range`, unversioned +//! buckets, metadata replay, fault actions) and the two-server environment. +//! No ODM behavior is exercised here. + +use super::common::{OdmTestEnv, SeedObject, fake_source_client, start_source_rustfs}; +use crate::fake_s3_target::{BucketMode, FakeS3Target, FakeS3TargetOptions, FaultAction, Operation, SeedMetadata}; +use aws_sdk_s3::Client; +use aws_sdk_s3::error::ProvideErrorMetadata; +use aws_sdk_s3::primitives::{ByteStream, DateTime}; +use bytes::Bytes; +use std::collections::BTreeSet; +use std::time::{Duration, Instant}; + +type TestResult = Result<(), Box>; + +const SOURCE_BUCKET: &str = "odm-source"; + +/// Position-dependent payload so a misaligned range read is caught. +fn payload(len: usize) -> Bytes { + (0..len).map(|index| (index % 251) as u8).collect::>().into() +} + +async fn fake_source() -> Result<(FakeS3Target, Client), Box> { + let source = FakeS3Target::start().await?; + source.create_bucket(SOURCE_BUCKET); + let client = fake_source_client(&source); + Ok((source, client)) +} + +/// Full ListObjectsV2 traversal. Returns `(keys, common prefixes, pages)` and +/// checks the page shape on the way: every page except the last is full and +/// truncated, the last carries no continuation token. +async fn list_all( + client: &Client, + prefix: Option<&str>, + delimiter: Option<&str>, + start_after: Option<&str>, + max_keys: i32, +) -> Result<(Vec, Vec, usize), Box> { + let mut keys = Vec::new(); + let mut prefixes = Vec::new(); + let mut pages = 0usize; + let mut token: Option = None; + loop { + let page = client + .list_objects_v2() + .bucket(SOURCE_BUCKET) + .set_prefix(prefix.map(str::to_string)) + .set_delimiter(delimiter.map(str::to_string)) + .set_start_after(start_after.map(str::to_string)) + .max_keys(max_keys) + .set_continuation_token(token.clone()) + .send() + .await?; + pages += 1; + let page_keys: Vec = page + .contents() + .iter() + .filter_map(|object| object.key().map(str::to_string)) + .collect(); + let page_prefixes: Vec = page + .common_prefixes() + .iter() + .filter_map(|common| common.prefix().map(str::to_string)) + .collect(); + let entries = page_keys.len() + page_prefixes.len(); + assert_eq!(page.key_count(), Some(entries as i32), "KeyCount must count keys and prefixes"); + assert_eq!(page.continuation_token(), token.as_deref(), "the request token must be echoed"); + keys.extend(page_keys); + prefixes.extend(page_prefixes); + if page.is_truncated() == Some(true) { + assert_eq!(entries as i32, max_keys, "every truncated page must be full"); + token = Some( + page.next_continuation_token() + .expect("truncated page must carry a continuation token") + .to_string(), + ); + } else { + assert!(page.next_continuation_token().is_none(), "final page must not carry a token"); + return Ok((keys, prefixes, pages)); + } + } +} + +#[tokio::test] +async fn fake_source_list_objects_v2_paginates_with_delimiter() -> TestResult { + let (source, client) = fake_source().await?; + let mut expected_keys = BTreeSet::new(); + for directory in 0..30 { + for file in 0..30 { + expected_keys.insert(format!("d{directory:02}/k{file:03}")); + } + } + for index in 0..100 { + expected_keys.insert(format!("top-{index:03}")); + } + assert_eq!(expected_keys.len(), 1000); + for key in &expected_keys { + source.put_seed_object(SOURCE_BUCKET, key.clone(), Bytes::from(key.clone()), &SeedMetadata::new()); + } + // A key whose current version is a delete marker must stay hidden. + client + .put_object() + .bucket(SOURCE_BUCKET) + .key("hidden/marker") + .body(ByteStream::from_static(b"gone")) + .send() + .await?; + client + .delete_object() + .bucket(SOURCE_BUCKET) + .key("hidden/marker") + .send() + .await?; + let expected_sorted: Vec = expected_keys.iter().cloned().collect(); + let expected_prefixes: Vec = (0..30).map(|directory| format!("d{directory:02}/")).collect(); + let expected_top: Vec = (0..100).map(|index| format!("top-{index:03}")).collect(); + + // Flat traversal in byte order, 1000 keys in pages of 7. + let (keys, prefixes, pages) = list_all(&client, None, None, None, 7).await?; + assert_eq!(keys, expected_sorted); + assert!(prefixes.is_empty()); + assert_eq!(pages, 143); + + // Delimiter folding: 30 common prefixes then 100 top-level keys, pages of 7. + let (keys, prefixes, pages) = list_all(&client, None, Some("/"), None, 7).await?; + assert_eq!(prefixes, expected_prefixes); + assert_eq!(keys, expected_top); + assert_eq!(pages, 19); + + // Empty prefix equals no prefix. + let (keys, _, _) = list_all(&client, Some(""), None, None, 1000).await?; + assert_eq!(keys, expected_sorted); + + // No match: empty, not truncated, no token. + let (keys, prefixes, pages) = list_all(&client, Some("zzz/"), Some("/"), None, 7).await?; + assert!(keys.is_empty() && prefixes.is_empty()); + assert_eq!(pages, 1); + let (keys, _, _) = list_all(&client, Some("hidden/"), None, None, 7).await?; + assert!(keys.is_empty(), "a current delete marker must hide its key"); + + // Exact page boundary: 30 keys under one directory, max-keys=30 -> one + // untruncated page. + let (keys, prefixes, pages) = list_all(&client, Some("d05/"), Some("/"), None, 30).await?; + assert_eq!(keys.len(), 30); + assert!(prefixes.is_empty()); + assert_eq!(pages, 1); + + // start-after skips keys at or before the marker. + let (keys, _, _) = list_all(&client, None, None, Some("top-097"), 1000).await?; + assert_eq!(keys, ["top-098", "top-099"]); + + // max-keys is clamped to 1000; exactly 1000 keys fit in one page. + let (keys, _, pages) = list_all(&client, None, None, None, 5000).await?; + assert_eq!(keys.len(), 1000); + assert_eq!(pages, 1); + + let listings: Vec<_> = source + .requests() + .into_iter() + .filter(|record| record.operation == Operation::ListObjectsV2) + .collect(); + assert!(listings.len() >= 143 + 19); + assert!(listings.iter().any(|record| record.prefix.as_deref() == Some("d05/"))); + assert!( + listings.iter().any(|record| record.continuation_token.is_some()), + "resumed pages must journal their continuation token" + ); + assert!(listings.iter().all(|record| record.user_agent.is_some())); + source.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn fake_source_range_get_variants_and_416() -> TestResult { + let (source, client) = fake_source().await?; + let body = payload(1000); + source.put_seed_object(SOURCE_BUCKET, "ranged", body.clone(), &SeedMetadata::new()); + + for (range, expected_range, expected_slice) in [ + ("bytes=10-19", "bytes 10-19/1000", &body[10..20]), + ("bytes=990-", "bytes 990-999/1000", &body[990..]), + ("bytes=-5", "bytes 995-999/1000", &body[995..]), + ("bytes=0-5000", "bytes 0-999/1000", &body[..]), + ] { + let output = client + .get_object() + .bucket(SOURCE_BUCKET) + .key("ranged") + .range(range) + .send() + .await?; + assert_eq!(output.content_range(), Some(expected_range), "{range}"); + assert_eq!(output.accept_ranges(), Some("bytes"), "{range}"); + assert_eq!(output.content_length(), Some(expected_slice.len() as i64), "{range}"); + let collected = output.body.collect().await?.into_bytes(); + assert_eq!(collected.as_ref(), expected_slice, "{range}"); + } + let head = client + .head_object() + .bucket(SOURCE_BUCKET) + .key("ranged") + .range("bytes=10-19") + .send() + .await?; + assert_eq!(head.content_range(), Some("bytes 10-19/1000")); + assert_eq!(head.content_length(), Some(10)); + + for range in ["bytes=1000-", "bytes=-0"] { + let error = client + .get_object() + .bucket(SOURCE_BUCKET) + .key("ranged") + .range(range) + .send() + .await + .expect_err("unsatisfiable range must fail"); + let response = error.raw_response().expect("416 must retain the raw response"); + assert_eq!(response.status().as_u16(), 416, "{range}"); + assert_eq!(response.headers().get("content-range"), Some("bytes */1000"), "{range}"); + assert_eq!(error.code(), Some("InvalidRange"), "{range}"); + } + + let ranged = source + .requests() + .into_iter() + .find(|record| record.operation == Operation::GetObject && record.range.as_deref() == Some("bytes=10-19")) + .expect("the Range header must be journaled verbatim"); + assert_eq!(ranged.key.as_deref(), Some("ranged")); + assert!(source.count_requests(Operation::GetObject, "ranged") >= 6); + source.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn fake_source_unversioned_bucket_overwrites_and_deletes() -> TestResult { + let (source, client) = fake_source().await?; + source.create_bucket_with_mode("plain-source", BucketMode::Unversioned); + let versioning = client.get_bucket_versioning().bucket("plain-source").send().await?; + assert!(versioning.status().is_none(), "unversioned bucket must report no versioning status"); + + let first = client + .put_object() + .bucket("plain-source") + .key("doc") + .body(ByteStream::from_static(b"first")) + .send() + .await?; + assert!(first.version_id().is_none()); + let second = client + .put_object() + .bucket("plain-source") + .key("doc") + .body(ByteStream::from_static(b"second")) + .send() + .await?; + assert!(second.version_id().is_none()); + let get = client.get_object().bucket("plain-source").key("doc").send().await?; + assert!(get.version_id().is_none(), "GET must not return x-amz-version-id"); + assert_eq!(get.body.collect().await?.into_bytes().as_ref(), b"second"); + let head = client.head_object().bucket("plain-source").key("doc").send().await?; + assert!(head.version_id().is_none(), "HEAD must not return x-amz-version-id"); + assert_eq!(source.stored_versions("plain-source", "doc").len(), 1, "overwrite must replace in place"); + + let deleted = client.delete_object().bucket("plain-source").key("doc").send().await?; + assert!(deleted.delete_marker().is_none() && deleted.version_id().is_none()); + let missing = client + .get_object() + .bucket("plain-source") + .key("doc") + .send() + .await + .expect_err("deleted object must be gone"); + assert_eq!(missing.raw_response().map(|response| response.status().as_u16()), Some(404)); + assert_eq!(missing.code(), Some("NoSuchKey")); + let missing_head = client + .head_object() + .bucket("plain-source") + .key("doc") + .send() + .await + .expect_err("deleted object must fail HEAD"); + assert_eq!(missing_head.raw_response().map(|response| response.status().as_u16()), Some(404)); + assert!(source.stored_versions("plain-source", "doc").is_empty(), "DELETE must not leave a marker"); + + // The versioned bucket on the same target keeps its version ids. + let versioned = client + .put_object() + .bucket(SOURCE_BUCKET) + .key("doc") + .body(ByteStream::from_static(b"versioned")) + .send() + .await?; + assert!(versioned.version_id().is_some()); + source.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn fake_source_replays_standard_and_user_metadata() -> TestResult { + let (source, client) = fake_source().await?; + let body = payload(4096); + let expected_etag = format!("\"{}\"", { + use md5::Digest as _; + hex_simd::encode_to_string(md5::Md5::digest(&body), hex_simd::AsciiCase::Lower) + }); + // 2026-01-01T00:00:00Z rendered as an HTTP date by the SDK. + let expires = DateTime::from_secs(1_767_225_600); + client + .put_object() + .bucket(SOURCE_BUCKET) + .key("meta") + .body(ByteStream::from(body.clone())) + .content_type("application/x-odm") + .content_encoding("gzip") + .content_disposition("attachment; filename=\"meta.bin\"") + .content_language("en-US") + .cache_control("max-age=60") + .expires(expires) + .metadata("Foo-Bar", "mixed case name") + .metadata("UPPER", "upper name") + .metadata("already-lower", "lower name") + .send() + .await?; + + let head = client.head_object().bucket(SOURCE_BUCKET).key("meta").send().await?; + let get = client.get_object().bucket(SOURCE_BUCKET).key("meta").send().await?; + for (label, content_type, content_encoding, content_disposition, content_language, cache_control, expires_string, e_tag) in [ + ( + "HEAD", + head.content_type(), + head.content_encoding(), + head.content_disposition(), + head.content_language(), + head.cache_control(), + head.expires_string(), + head.e_tag(), + ), + ( + "GET", + get.content_type(), + get.content_encoding(), + get.content_disposition(), + get.content_language(), + get.cache_control(), + get.expires_string(), + get.e_tag(), + ), + ] { + assert_eq!(content_type, Some("application/x-odm"), "{label}"); + assert_eq!(content_encoding, Some("gzip"), "{label}"); + assert_eq!(content_disposition, Some("attachment; filename=\"meta.bin\""), "{label}"); + assert_eq!(content_language, Some("en-US"), "{label}"); + assert_eq!(cache_control, Some("max-age=60"), "{label}"); + assert_eq!(expires_string, Some("Thu, 01 Jan 2026 00:00:00 GMT"), "{label}"); + assert_eq!(e_tag, Some(expected_etag.as_str()), "{label}"); + } + for metadata in [head.metadata(), get.metadata()] { + let metadata = metadata.expect("user metadata must be replayed"); + assert_eq!(metadata.get("foo-bar").map(String::as_str), Some("mixed case name")); + assert_eq!(metadata.get("upper").map(String::as_str), Some("upper name")); + assert_eq!(metadata.get("already-lower").map(String::as_str), Some("lower name")); + assert!(!metadata.contains_key("Foo-Bar") && !metadata.contains_key("UPPER")); + } + assert!(head.last_modified().is_some()); + assert_eq!(head.last_modified(), get.last_modified()); + assert_eq!(head.content_length(), Some(4096)); + assert_eq!(get.body.collect().await?.into_bytes(), body); + + // Seeded objects replay the same way. + let seeded_etag = source.put_seed_object( + SOURCE_BUCKET, + "seeded", + Bytes::from_static(b"seeded"), + &SeedMetadata::new() + .content_type("text/plain") + .content_encoding("identity") + .cache_control("no-store") + .user_metadata("Origin", "seed"), + ); + let seeded = client.head_object().bucket(SOURCE_BUCKET).key("seeded").send().await?; + assert_eq!(seeded.e_tag(), Some(format!("\"{seeded_etag}\"").as_str())); + assert_eq!(seeded.content_type(), Some("text/plain")); + assert_eq!(seeded.content_encoding(), Some("identity")); + assert_eq!(seeded.cache_control(), Some("no-store")); + assert_eq!( + seeded + .metadata() + .and_then(|metadata| metadata.get("origin")) + .map(String::as_str), + Some("seed") + ); + source.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn fake_source_fault_actions_truncate_stall_and_status() -> TestResult { + let (source, client) = fake_source().await?; + let body = payload(4096); + source.put_seed_object(SOURCE_BUCKET, "faulty", body.clone(), &SeedMetadata::new()); + + // TruncateBodyAt: headers promise 4096 bytes, the body ends after 100. + source.inject_for_key(Operation::GetObject, "faulty", FaultAction::TruncateBodyAt(100), 1); + let truncated = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?; + assert_eq!(truncated.content_length(), Some(4096)); + let short_read = truncated + .body + .collect() + .await + .expect_err("a truncated body must fail to collect"); + let short_read = short_read.to_string(); + assert!(!short_read.is_empty()); + + // ResponseStatus: arbitrary status with the matching S3 error code. + for (code, expected_code) in [ + (429u16, "SlowDown"), + (404, "NoSuchKey"), + (500, "InternalError"), + (503, "ServiceUnavailable"), + ] { + source.inject(Operation::GetObject, FaultAction::ResponseStatus(code), 1); + let error = client + .get_object() + .bucket(SOURCE_BUCKET) + .key("faulty") + .send() + .await + .expect_err("scripted status must fail"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(code)); + assert_eq!(error.code(), Some(expected_code)); + } + + // Stall: the fully computed response is held before its first byte. + source.inject(Operation::HeadObject, FaultAction::Stall(Duration::from_millis(400)), 1); + let started = Instant::now(); + let stalled = client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?; + assert!(started.elapsed() >= Duration::from_millis(350), "stall must delay the first byte"); + assert_eq!(stalled.content_length(), Some(4096)); + let unstalled_started = Instant::now(); + client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?; + assert!(unstalled_started.elapsed() < Duration::from_millis(350), "stall is consumed once"); + + // The object is intact once the script is drained. + let intact = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?; + assert_eq!(intact.body.collect().await?.into_bytes(), body); + + assert_eq!(source.count_requests(Operation::GetObject, "faulty"), 6); + assert_eq!(source.count_requests(Operation::HeadObject, "faulty"), 2); + assert_eq!(source.count_requests(Operation::GetObject, "other"), 0); + let records = source.requests(); + assert!( + records.iter().all(|record| record + .user_agent + .as_deref() + .is_some_and(|agent| agent.contains("aws-sdk-rust"))), + "the SDK user agent must be journaled" + ); + assert!( + records + .iter() + .any(|record| record.fault == Some(FaultAction::TruncateBodyAt(100))) + ); + assert!( + records + .iter() + .any(|record| record.fault == Some(FaultAction::Stall(Duration::from_millis(400)))) + ); + source.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn fake_source_raised_object_cap_accepts_large_put() -> TestResult { + let source = FakeS3Target::start_with_options(FakeS3TargetOptions { + max_object_bytes: 96 * 1024 * 1024, + }) + .await?; + source.create_bucket(SOURCE_BUCKET); + let client = fake_source_client(&source); + let len = 64 * 1024 * 1024 + 1; + client + .put_object() + .bucket(SOURCE_BUCKET) + .key("large") + .body(ByteStream::from(vec![7u8; len])) + .send() + .await?; + let head = client.head_object().bucket(SOURCE_BUCKET).key("large").send().await?; + assert_eq!(head.content_length(), Some(len as i64)); + let tail = client + .get_object() + .bucket(SOURCE_BUCKET) + .key("large") + .range("bytes=-1") + .send() + .await?; + assert_eq!(tail.content_range(), Some(format!("bytes {}-{}/{len}", len - 1, len - 1).as_str())); + source.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn odm_env_starts_rustfs_and_fake_source() -> TestResult { + let env = OdmTestEnv::start().await?; + env.source.create_bucket(SOURCE_BUCKET); + let local_bucket = "odm-local"; + env.rustfs.create_test_bucket(local_bucket).await?; + + let etags = env.seed_source( + SOURCE_BUCKET, + &[ + SeedObject::new("seed/a", Bytes::from_static(b"alpha")), + SeedObject::new("seed/b", Bytes::from_static(b"beta")) + .with_metadata(SeedMetadata::new().content_type("text/plain").user_metadata("Kind", "seed")), + ], + ); + assert_eq!(etags.len(), 2); + assert!(env.source.requests().is_empty(), "seeding must not touch the journal"); + let source_client = env.source_client(); + let seeded = source_client.head_object().bucket(SOURCE_BUCKET).key("seed/b").send().await?; + assert_eq!(seeded.content_type(), Some("text/plain")); + assert_eq!(seeded.e_tag(), Some(format!("\"{}\"", etags[1]).as_str())); + assert_eq!(env.source.count_requests(Operation::HeadObject, "seed/b"), 1); + + env.assert_local_absent(local_bucket, "seed/a").await; + env.client + .put_object() + .bucket(local_bucket) + .key("seed/a") + .body(ByteStream::from_static(b"alpha")) + .send() + .await?; + env.assert_local_present(local_bucket, "seed/a", b"alpha").await; + env.assert_local_absent(local_bucket, "seed/b").await; + + let spec = env.fake_source_spec(SOURCE_BUCKET).to_json(); + assert_eq!(spec["version"], 1); + assert_eq!(spec["enabled"], true); + assert_eq!(spec["source"]["provider"], "s3"); + assert_eq!(spec["source"]["endpoint"], env.source.endpoint()); + assert_eq!(spec["source"]["bucket"], SOURCE_BUCKET); + assert_eq!(spec["source"]["credentials"]["secret_key"], "fake-secret"); + assert_eq!(spec["policy"]["source_timeout"]["first_byte_ms"], 15_000); + assert!(spec["policy"]["bandwidth_limit_bytes_per_sec"].is_null()); + let debug = format!("{:?}", env.fake_source_spec(SOURCE_BUCKET)); + assert!(!debug.contains("fake-secret"), "Debug output must redact the secret"); + Ok(()) +} + +#[tokio::test] +async fn start_source_rustfs_round_trips_put_get() -> TestResult { + let env = OdmTestEnv::start().await?; + let source = start_source_rustfs().await?; + assert_ne!(source.url, env.rustfs.url, "the source must be a separate instance"); + + source.create_test_bucket(SOURCE_BUCKET).await?; + let source_client = source.create_s3_client(); + let body = payload(70_000); + let put = source_client + .put_object() + .bucket(SOURCE_BUCKET) + .key("real/object") + .body(ByteStream::from(body.clone())) + .content_type("application/octet-stream") + .send() + .await?; + assert!(put.e_tag().is_some()); + let get = source_client + .get_object() + .bucket(SOURCE_BUCKET) + .key("real/object") + .send() + .await?; + assert_eq!(get.content_type(), Some("application/octet-stream")); + assert_eq!(get.body.collect().await?.into_bytes(), body); + + let visible_to_primary = env + .client + .list_buckets() + .send() + .await? + .buckets() + .iter() + .any(|bucket| bucket.name() == Some(SOURCE_BUCKET)); + assert!(!visible_to_primary, "the two servers must not share state"); + let spec = super::common::OdmSourceSpec::for_rustfs_source(&source, SOURCE_BUCKET).to_json(); + assert_eq!(spec["source"]["provider"], "rustfs"); + assert_eq!(spec["source"]["endpoint"], source.url); + Ok(()) +} diff --git a/crates/e2e_test/src/on_demand_migration/mod.rs b/crates/e2e_test/src/on_demand_migration/mod.rs new file mode 100644 index 000000000..ffa4d6874 --- /dev/null +++ b/crates/e2e_test/src/on_demand_migration/mod.rs @@ -0,0 +1,24 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! On-demand migration (ODM) end-to-end suite (rustfs/backlog#2147). +//! +//! `common` is the shared environment: one RustFS under test, one programmable +//! fake S3 source, admin-API wrappers, seeding and local-state assertions. +//! `harness_self_test` proves the harness itself; ODM behavior scenarios are +//! separate modules wired by later tasks. + +pub mod common; + +mod harness_self_test;