From d04611ba3889c93f0d2bf27e504eab5d45c3fb86 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 3 Sep 2026 07:21:51 +0800 Subject: [PATCH] feat(object): serve GET misses from the on-demand migration source (#7084) feat(rustfs): serve GET misses from the migration source Wire the on-demand migration read-through into the GET path, after the local read and the replication proxy have both missed (rustfs/backlog#2156). A source HEAD supplies size, validators and metadata. Conditional headers are evaluated locally against it and never forwarded, so a source 304/412 cannot be mistaken for a source failure. An object within inline_max_bytes is teed: the primary streams to the client while the secondary commits the local copy in a background task, so a client disconnect still stores the whole object and a failed write-back never touches the client stream. Range reads and larger objects stream straight through and queue a background pull per policy. Concurrent misses of one key share the singleflight slot: the leader tees, followers re-read local after it commits or degrade to passthrough after first_byte_ms. Version reads, partNumber reads, anti-loop marked requests and a respected local delete marker keep their original 404. Source answers carry x-rustfs-on-demand-migration: source; local hits are untouched, and the local hit path gains no await or lock. --- .../src/on_demand_migration/common.rs | 69 +- .../src/on_demand_migration/get_basic_test.rs | 265 ++++ .../e2e_test/src/on_demand_migration/mod.rs | 5 +- rustfs/src/app/object/get.rs | 1192 ++++++++++++++++- rustfs/src/app/object/shared.rs | 87 +- rustfs/src/app/storage_api.rs | 4 +- 6 files changed, 1601 insertions(+), 21 deletions(-) create mode 100644 crates/e2e_test/src/on_demand_migration/get_basic_test.rs diff --git a/crates/e2e_test/src/on_demand_migration/common.rs b/crates/e2e_test/src/on_demand_migration/common.rs index 7405a2650..4058e2a59 100644 --- a/crates/e2e_test/src/on_demand_migration/common.rs +++ b/crates/e2e_test/src/on_demand_migration/common.rs @@ -22,7 +22,7 @@ //! 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 crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FakeS3TargetOptions, Operation, SeedMetadata}; use aws_config::retry::RetryConfig; use aws_sdk_s3::Client; use aws_sdk_s3::config::{Credentials, Region}; @@ -30,12 +30,16 @@ use aws_smithy_http_client::Builder as SmithyHttpClientBuilder; use bytes::Bytes; use serde::Serialize; use std::fmt; +use std::time::{Duration, Instant}; 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"; +/// The source client shares the replication egress guard, which rejects the +/// fake source's loopback endpoint unless this switch is set. +pub const ALLOW_LOOPBACK_SOURCE_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET"; /// 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). @@ -235,6 +239,21 @@ impl AdminResponse { } } +/// Raw S3 response (status, headers, body) for assertions on headers the +/// SDK does not surface, such as `x-rustfs-on-demand-migration`. +#[derive(Debug, Clone)] +pub struct RawResponse { + pub status: u16, + pub headers: http::HeaderMap, + pub body: Bytes, +} + +impl RawResponse { + pub fn header(&self, name: &str) -> Option<&str> { + self.headers.get(name).and_then(|value| value.to_str().ok()) + } +} + /// One object to seed into the source. #[derive(Clone)] pub struct SeedObject { @@ -277,7 +296,7 @@ impl OdmTestEnv { 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")]) + .start_rustfs_server_with_env(vec![], &[(ODM_MODULE_SWITCH_ENV, "true"), (ALLOW_LOOPBACK_SOURCE_ENV, "true")]) .await?; let client = rustfs.create_s3_client(); Ok(Self { rustfs, source, client }) @@ -412,6 +431,52 @@ impl OdmTestEnv { assert_eq!(body.as_ref(), expected, "{bucket}/{key} local content mismatch"); } + /// Polls the listing until `key` is present locally or `timeout` elapses + /// (background pulls land after the response that triggered them). + pub async fn wait_local_listed(&self, bucket: &str, key: &str, timeout: Duration) -> Result { + let deadline = Instant::now() + timeout; + loop { + if self.local_key_listed(bucket, key).await? { + return Ok(true); + } + if Instant::now() >= deadline { + return Ok(false); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + + /// Raw signed `GET /{bucket}/{key}` against the RustFS under test. + pub async fn raw_get(&self, bucket: &str, key: &str) -> Result { + let url = format!("{}/{bucket}/{key}", self.rustfs.url); + let response = + signed_request(http::Method::GET, &url, &self.rustfs.access_key, &self.rustfs.secret_key, None, None).await?; + Ok(RawResponse { + status: response.status().as_u16(), + headers: response.headers().clone(), + body: response.bytes().await?, + }) + } + + /// Waits until the runtime consults the source for `bucket`: a config + /// install is applied asynchronously after the admin call returns. The + /// probe is a HEAD on a key that exists nowhere, so nothing is pulled and + /// only that key enters the negative cache. + pub async fn wait_until_source_consulted(&self, bucket: &str) -> Result<(), BoxError> { + const PROBE_KEY: &str = "_odm-readiness-probe"; + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let _ = self.client.head_object().bucket(bucket).key(PROBE_KEY).send().await; + if self.source.count_requests(Operation::HeadObject, PROBE_KEY) > 0 { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(format!("on-demand migration runtime for {bucket} did not consult the source in time").into()); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + /// Panics if `key` is listed locally. pub async fn assert_local_absent(&self, bucket: &str, key: &str) { assert!( diff --git a/crates/e2e_test/src/on_demand_migration/get_basic_test.rs b/crates/e2e_test/src/on_demand_migration/get_basic_test.rs new file mode 100644 index 000000000..f9cde81fe --- /dev/null +++ b/crates/e2e_test/src/on_demand_migration/get_basic_test.rs @@ -0,0 +1,265 @@ +// 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. + +//! Basic GET read-through scenarios (rustfs/backlog#2156): inline pull and +//! local persistence, large-object passthrough with background backfill, +//! Range passthrough, source 404, `versionId` reads, and a disabled bucket. +//! Every source-side expectation is asserted on the fake source's journal. + +use super::common::{BoxError, OdmSourceSpec, OdmTestEnv, SeedObject}; +use crate::fake_s3_target::{BucketMode, Operation}; +use aws_sdk_s3::error::ProvideErrorMetadata; +use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration}; +use bytes::Bytes; +use std::time::Duration; + +type TestResult = Result<(), BoxError>; + +const SOURCE_BUCKET: &str = "odm-get-source"; +const ODM_RESPONSE_HEADER: &str = "x-rustfs-on-demand-migration"; +/// Background pulls run after the response; generous for a loaded CI host. +const BACKFILL_WAIT: Duration = Duration::from_secs(60); + +/// Position-dependent payload so a misaligned or truncated copy is caught. +fn payload(len: usize) -> Bytes { + (0..len).map(|index| (index % 251) as u8).collect::>().into() +} + +/// RustFS with `local_bucket` migrating from `SOURCE_BUCKET` on the fake +/// source (unversioned, like a plain migration source); `adjust` tweaks the +/// policy before it is installed. Returns once the runtime consults the +/// source. +async fn configured_env(local_bucket: &str, adjust: impl FnOnce(&mut OdmSourceSpec)) -> Result { + let env = OdmTestEnv::start().await?; + env.source.create_bucket_with_mode(SOURCE_BUCKET, BucketMode::Unversioned); + env.rustfs.create_test_bucket(local_bucket).await?; + let mut spec = env.fake_source_spec(SOURCE_BUCKET); + adjust(&mut spec); + let response = env.configure_source(local_bucket, &spec).await?; + assert_eq!(response.status, 200, "configure on-demand migration: {}", response.body); + env.wait_until_source_consulted(local_bucket).await?; + Ok(env) +} + +fn source_get_ranges(env: &OdmTestEnv, key: &str) -> Vec> { + env.source + .requests() + .into_iter() + .filter(|record| record.operation == Operation::GetObject && record.key.as_deref() == Some(key)) + .map(|record| record.range) + .collect() +} + +#[tokio::test] +async fn get_miss_pulls_inline_and_serves_locally_afterwards() -> TestResult { + let bucket = "odm-get-inline"; + let env = configured_env(bucket, |_| {}).await?; + let key = "inline/report.bin"; + let body = payload(200 * 1024); + let etag = env + .seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]) + .remove(0); + let quoted_etag = format!("\"{etag}\""); + + let first = env.raw_get(bucket, key).await?; + assert_eq!(first.status, 200, "{}", String::from_utf8_lossy(&first.body)); + assert_eq!(first.header(ODM_RESPONSE_HEADER), Some("source"), "a source answer is marked"); + assert_eq!(first.header("etag"), Some(quoted_etag.as_str()), "inline answers carry the source ETag"); + assert_eq!(first.header("content-length"), Some(body.len().to_string().as_str())); + assert_eq!(first.header("accept-ranges"), Some("bytes")); + assert_eq!(first.body, body, "the client receives the source bytes"); + assert_eq!(env.source.count_requests(Operation::GetObject, key), 1, "exactly one source GET"); + assert_eq!(env.source.count_requests(Operation::HeadObject, key), 1); + + assert!( + env.wait_local_listed(bucket, key, BACKFILL_WAIT).await?, + "the inline pull must store the object locally" + ); + let second = env.raw_get(bucket, key).await?; + assert_eq!(second.status, 200); + assert_eq!(second.header(ODM_RESPONSE_HEADER), None, "a local hit carries no source marker"); + assert_eq!(second.body, body, "the local copy is the source bytes"); + assert_eq!(second.header("etag"), Some(quoted_etag.as_str()), "preserve_etag keeps the source ETag"); + assert_eq!( + env.source.count_requests(Operation::GetObject, key), + 1, + "the second GET is served locally" + ); + assert_eq!(env.source.count_requests(Operation::HeadObject, key), 1); + Ok(()) +} + +#[tokio::test] +async fn get_large_object_streams_through_and_backfills_in_background() -> TestResult { + let bucket = "odm-get-large"; + let env = configured_env(bucket, |spec| spec.policy.inline_max_bytes = 4096).await?; + let key = "large/archive.bin"; + let body = payload(512 * 1024); + env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]); + + let response = env.raw_get(bucket, key).await?; + assert_eq!(response.status, 200, "{}", String::from_utf8_lossy(&response.body)); + assert_eq!(response.header(ODM_RESPONSE_HEADER), Some("source")); + assert_eq!(response.header("content-length"), Some(body.len().to_string().as_str())); + assert_eq!(response.body, body, "the passthrough streams the whole object"); + + assert!( + env.wait_local_listed(bucket, key, BACKFILL_WAIT).await?, + "the background pull must store the object locally" + ); + env.assert_local_present(bucket, key, &body).await; + assert_eq!( + source_get_ranges(&env, key), + vec![None, None], + "one passthrough GET plus one background pull, both unranged" + ); + Ok(()) +} + +#[tokio::test] +async fn get_range_streams_206_and_backfills_the_whole_object() -> TestResult { + let bucket = "odm-get-range"; + let env = configured_env(bucket, |_| {}).await?; + let key = "range/video.bin"; + let body = payload(100_000); + let etag = env + .seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]) + .remove(0); + + let response = env + .client + .get_object() + .bucket(bucket) + .key(key) + .range("bytes=10-19") + .send() + .await?; + assert_eq!(response.content_range(), Some("bytes 10-19/100000"), "the source's 206 is passed through"); + assert_eq!(response.content_length(), Some(10)); + assert_eq!(response.e_tag(), Some(format!("\"{etag}\"").as_str())); + assert_eq!(response.body.collect().await?.into_bytes(), body.slice(10..20)); + assert_eq!( + source_get_ranges(&env, key), + vec![Some("bytes=10-19".to_string())], + "the Range is forwarded" + ); + + assert!( + env.wait_local_listed(bucket, key, BACKFILL_WAIT).await?, + "serve_and_backfill must pull the whole object" + ); + env.assert_local_present(bucket, key, &body).await; + assert_eq!( + source_get_ranges(&env, key), + vec![Some("bytes=10-19".to_string()), None], + "the background pull fetches the whole object" + ); + Ok(()) +} + +#[tokio::test] +async fn get_source_not_found_is_404_and_negative_cached() -> TestResult { + let bucket = "odm-get-missing"; + let env = configured_env(bucket, |_| {}).await?; + let key = "missing/nowhere.bin"; + + for attempt in 1..=2 { + let err = env + .client + .get_object() + .bucket(bucket) + .key(key) + .send() + .await + .expect_err("a key missing on both sides is 404"); + assert_eq!(err.code(), Some("NoSuchKey"), "attempt {attempt}: {err:?}"); + } + assert_eq!(env.source.count_requests(Operation::GetObject, key), 0, "a source miss never pulls"); + assert_eq!( + env.source.count_requests(Operation::HeadObject, key), + 1, + "the second miss stops at the negative cache" + ); + env.assert_local_absent(bucket, key).await; + Ok(()) +} + +#[tokio::test] +async fn get_with_version_id_does_not_consult_the_source() -> TestResult { + let bucket = "odm-get-versioned"; + let env = configured_env(bucket, |_| {}).await?; + env.client + .put_bucket_versioning() + .bucket(bucket) + .versioning_configuration( + VersioningConfiguration::builder() + .status(BucketVersioningStatus::Enabled) + .build(), + ) + .send() + .await?; + let key = "versioned/doc.bin"; + let body = payload(1024); + env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]); + + let err = env + .client + .get_object() + .bucket(bucket) + .key(key) + .version_id("11111111-2222-4333-8444-555555555555") + .send() + .await + .expect_err("a version read cannot be answered by the source"); + assert!( + matches!(err.code(), Some("NoSuchVersion") | Some("NoSuchKey")), + "unexpected error: {err:?}" + ); + assert_eq!(env.source.count_requests(Operation::HeadObject, key), 0); + assert_eq!(env.source.count_requests(Operation::GetObject, key), 0); + env.assert_local_absent(bucket, key).await; + + // The same key without versionId is still migrated: the gate is per request. + let response = env.raw_get(bucket, key).await?; + assert_eq!(response.status, 200, "{}", String::from_utf8_lossy(&response.body)); + assert_eq!(response.header(ODM_RESPONSE_HEADER), Some("source")); + assert_eq!(response.body, body); + assert_eq!(env.source.count_requests(Operation::GetObject, key), 1); + Ok(()) +} + +#[tokio::test] +async fn get_after_disable_does_not_consult_the_source() -> TestResult { + let bucket = "odm-get-disabled"; + let env = configured_env(bucket, |_| {}).await?; + let key = "disabled/doc.bin"; + env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, payload(1024))]); + + let response = env.disable(bucket).await?; + assert_eq!(response.status, 204, "{}", response.body); + + let err = env + .client + .get_object() + .bucket(bucket) + .key(key) + .send() + .await + .expect_err("a disabled bucket answers locally"); + assert_eq!(err.code(), Some("NoSuchKey"), "{err:?}"); + assert_eq!(env.source.count_requests(Operation::HeadObject, key), 0); + assert_eq!(env.source.count_requests(Operation::GetObject, key), 0); + env.assert_local_absent(bucket, key).await; + Ok(()) +} diff --git a/crates/e2e_test/src/on_demand_migration/mod.rs b/crates/e2e_test/src/on_demand_migration/mod.rs index ffa4d6874..4dd04dced 100644 --- a/crates/e2e_test/src/on_demand_migration/mod.rs +++ b/crates/e2e_test/src/on_demand_migration/mod.rs @@ -16,9 +16,10 @@ //! //! `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. +//! `harness_self_test` proves the harness itself; `get_basic_test` covers the +//! GET read-through (rustfs/backlog#2156). pub mod common; +mod get_basic_test; mod harness_self_test; diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index c0bdfe859..2e658ed8a 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -15,6 +15,13 @@ //! GetObject / GetObjectAttributes read path: cold fill, resume, stream tuning. use super::*; +use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ + BucketOdmState, OdmLookup, OdmOp, OdmOutcome, OnDemandMigrationSys, PullError, PullLeader, PullOutcome, PullReason, PullSlot, + RangeGetPolicy, SourceClient, SourceError, SourceGet, SourceHead, commit_inline, +}; +use crate::app::storage_api::object_usecase::on_demand_migration::WriteBackBody; +use rustfs_rio::{TeeOptions, TeePrimary, tee_reader_with_options}; +use tokio_stream::wrappers::ReceiverStream; struct ColdFillDiskPermitMetric { owner: ColdFillDiskPermitOwner, @@ -3818,6 +3825,58 @@ impl DefaultObjectUsecase { } } + /// On-demand migration read-through for a GET miss (rustfs/backlog#2156): + /// consulted only after the local read and the replication proxy both + /// missed. `None` means the runtime does not intervene and the caller + /// keeps its original 404. + #[allow(clippy::too_many_arguments)] + async fn on_demand_migration_get( + &self, + req: &S3Request, + store: &Arc, + bucket: &str, + key: &str, + range: Option<&HTTPRangeSpec>, + opts: &ObjectOptions, + part_number: Option, + ) -> Option { + if !odm_get_may_consult_source(opts, part_number) { + return None; + } + let lookup = OnDemandMigrationSys::get().resolve(bucket, key)?; + let (state, client) = match odm_get_verdict(lookup) { + OdmGetVerdict::Fail(err) => return Some(OdmGetOutcome::Respond(Err(err))), + OdmGetVerdict::Consult { state, client } => (state, client), + }; + let policy = &state.config().policy; + // The read path reports a latest delete marker as a plain 404, so the + // marker is classified here, and only where one can exist. + if policy.respect_local_delete_marker && (opts.versioned || opts.version_suspended) { + let lookup = store.get_object_info(bucket, key, opts).await; + match odm_local_miss(lookup.as_ref()) { + Some(miss) if odm_policy_admits_miss(policy, miss) => {} + Some(_) => return None, + // The object appeared meanwhile, or the lookup failed for a + // reason the source cannot answer: the local read decides. + None => return Some(OdmGetOutcome::RetryLocal), + } + } + let request_context = req.extensions.get::().cloned(); + let reply = odm_get_from_source(&state, client.as_ref(), &req.headers, key, range, request_context).await; + Some(match reply { + OdmGetReply::Served { output, backfill } => { + if let Some(reason) = backfill { + // Queue outcomes are the queue's own accounting + // (`queue_full`); the response does not depend on them. + let _ = state.enqueue_pull(key, reason); + } + OdmGetOutcome::Respond(Ok(output)) + } + OdmGetReply::Error(err) => OdmGetOutcome::Respond(Err(err)), + OdmGetReply::RetryLocal => OdmGetOutcome::RetryLocal, + }) + } + #[instrument(name = "execute_get_object", level = "trace", skip(self, req))] pub async fn execute_get_object(&self, req: S3Request) -> S3Result> { self.execute_get_object_boxed(req).await @@ -3941,7 +4000,7 @@ impl DefaultObjectUsecase { let manager = get_concurrency_manager(); - let prepared_read = match self + let mut prepared_read = self .prepare_get_object_read_execution( &req, manager, @@ -3950,29 +4009,70 @@ impl DefaultObjectUsecase { &timeout_config, &bucket, &key, - rs, + rs.clone(), &opts, part_number, - object_traffic_health, + object_traffic_health.clone(), ) - .await + .await; + // An object missing locally (and only missing — other errors keep + // their semantics) may still be served by a remote copy. + if let Err(err) = &prepared_read + && matches!(*err.code(), S3ErrorCode::NoSuchKey | S3ErrorCode::NoSuchVersion) { - Ok(prepared_read) => prepared_read, - Err(err) => { - // Active-active replication lag window: an object missing - // locally (and only missing — other errors keep their - // semantics) may still be served by proxying the GET to a - // replication target (backlog#1675 P1-5). - if matches!(*err.code(), S3ErrorCode::NoSuchKey | S3ErrorCode::NoSuchVersion) - && let Some(output) = Self::proxy_get_object_to_replication_targets(&req, &bucket, &key, &opts).await - { + // Active-active replication lag window: proxy the GET to a + // replication target (backlog#1675 P1-5). + if let Some(output) = Self::proxy_get_object_to_replication_targets(&req, &bucket, &key, &opts).await { + lifecycle.finish_ok(); + let mut response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await; + inject_accept_ranges_header(&mut response.headers); + let result = Ok(response); + let _ = helper.version_id(version_id_for_event).complete(&result); + return result; + } + // On-demand migration source (rustfs/backlog#2156): the + // authoritative external copy, after the cheaper replication + // proxy. + match self + .on_demand_migration_get(&req, &store, &bucket, &key, rs.as_ref(), &opts, part_number) + .await + { + None => {} + Some(OdmGetOutcome::Respond(Ok(output))) => { lifecycle.finish_ok(); - let mut response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await; + let mut response = wrap_response_with_cors(&bucket, &req.method, &req.headers, *output).await; inject_accept_ranges_header(&mut response.headers); + mark_on_demand_migration_response(&mut response.headers); let result = Ok(response); let _ = helper.version_id(version_id_for_event).complete(&result); return result; } + Some(OdmGetOutcome::Respond(Err(err))) => { + lifecycle.finish_err(); + return Self::complete_get_object_error(helper.version_id(version_id_for_event), err); + } + Some(OdmGetOutcome::RetryLocal) => { + prepared_read = self + .prepare_get_object_read_execution( + &req, + manager, + store.clone(), + &wrapper, + &timeout_config, + &bucket, + &key, + rs, + &opts, + part_number, + object_traffic_health, + ) + .await; + } + } + } + let prepared_read = match prepared_read { + Ok(prepared_read) => prepared_read, + Err(err) => { lifecycle.finish_err(); return Self::complete_get_object_error(helper.version_id(version_id_for_event), err); } @@ -4331,6 +4431,1070 @@ fn object_attributes_requested(object_attributes: &[ObjectAttributes], name: &'s }) } +/// Bytes the inline tee may queue for the write-back ahead of the client. +const ODM_INLINE_TEE_BUFFER_BYTES: usize = 1024 * 1024; +/// Chunks the client-side pump may run ahead of the response body. +const ODM_INLINE_CLIENT_CHANNEL_CHUNKS: usize = 8; +/// Read size of the source body streams handed to the client. +const ODM_SOURCE_BODY_CHUNK_BYTES: usize = 64 * 1024; + +/// Source seam for the on-demand migration GET read-through: production +/// goes through [`SourceClient`], tests script the answers. +pub(super) trait OdmGetSource { + async fn head_object(&self, key: &str) -> Result; + async fn get_object(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result; + async fn get_object_tagging(&self, key: &str) -> Result, SourceError>; +} + +impl OdmGetSource for SourceClient { + async fn head_object(&self, key: &str) -> Result { + SourceClient::head_object(self, key).await + } + + async fn get_object(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result { + SourceClient::get_object(self, key, range).await + } + + async fn get_object_tagging(&self, key: &str) -> Result, SourceError> { + SourceClient::get_object_tagging(self, key).await + } +} + +/// What the on-demand migration runtime decided for a GET miss before any +/// source traffic. +pub(super) enum OdmGetVerdict { + /// Answer with this error without touching the source. + Fail(S3Error), + /// Consult the source through `client`. + Consult { + state: Arc, + client: Arc, + }, +} + +/// How the source flow answers a GET miss. +pub(super) enum OdmGetReply { + /// Stream this source-backed body to the client; `backfill` names the + /// background pull to queue once the response is on its way. + Served { + output: Box, + backfill: Option, + }, + Error(S3Error), + /// A concurrent leader committed the object locally; read local again. + RetryLocal, +} + +/// What the GET handler does after the on-demand migration branch. +pub(super) enum OdmGetOutcome { + Respond(S3Result>), + RetryLocal, +} + +/// Request-level gate for the GET read-through, on top of the shared +/// [`odm_request_may_consult_source`] (version reads and anti-loop marked +/// requests): a `partNumber` read has no source counterpart, since the +/// source's part layout is its own, so it keeps the original 404. +pub(super) fn odm_get_may_consult_source(opts: &ObjectOptions, part_number: Option) -> bool { + part_number.is_none() && odm_request_may_consult_source(opts) +} + +/// Applies the lookup verdict of [`OnDemandMigrationSys::resolve`] to a GET +/// miss, recording the outcome for every request that stops here. Unlike +/// HEAD, an open breaker follows `policy.source_error` (the spec's +/// "source error / breaker open" row). +pub(super) fn odm_get_verdict(lookup: OdmLookup) -> OdmGetVerdict { + let state = Arc::clone(lookup.state()); + let policy = &state.config().policy; + let stats = state.stats(); + let source_error = |class: &'static str| { + stats.record_request(OdmOp::Get, OdmOutcome::SourceError); + OdmGetVerdict::Fail(odm_source_error_response(policy, class)) + }; + match &lookup { + OdmLookup::NegativeCached { .. } => { + stats.record_request(OdmOp::Get, OdmOutcome::NegativeCached); + OdmGetVerdict::Fail(S3Error::new(S3ErrorCode::NoSuchKey)) + } + OdmLookup::BreakerOpen { .. } => { + stats.record_request(OdmOp::Get, OdmOutcome::BreakerOpen); + OdmGetVerdict::Fail(odm_source_error_response(policy, "breaker_open")) + } + OdmLookup::Unavailable { error, .. } => source_error(odm_state_error_class(error)), + OdmLookup::Ready { .. } => match state.client() { + Ok(client) => OdmGetVerdict::Consult { + client: Arc::clone(client), + state: Arc::clone(&state), + }, + Err(error) => source_error(odm_state_error_class(error)), + }, + } +} + +/// Runs one source call and feeds its latency and error into the bucket +/// runtime (breaker scoring, negative cache, `last_source_error`). +async fn odm_observe( + state: &BucketOdmState, + key: &str, + call: impl std::future::Future>, +) -> Result { + let started = Instant::now(); + let result = call.await; + state.observe_source(started.elapsed(), key, result.as_ref().err()); + result +} + +/// Client-facing error for a failed source call, recording the GET outcome: +/// 404 for a source miss, 424 for an unsupported source object (SSE-C), +/// `policy.source_error` for everything else. +fn odm_get_source_failure(state: &BucketOdmState, err: &SourceError) -> S3Error { + let stats = state.stats(); + match err { + SourceError::NotFound => { + stats.record_request(OdmOp::Get, OdmOutcome::SourceMiss); + S3Error::new(S3ErrorCode::NoSuchKey) + } + SourceError::Unsupported(_) => { + stats.record_request(OdmOp::Get, OdmOutcome::Unsupported); + odm_source_unavailable_error(err.class_label()) + } + _ => { + stats.record_request(OdmOp::Get, OdmOutcome::SourceError); + odm_source_error_response(&state.config().policy, err.class_label()) + } + } +} + +fn odm_content_length(size: u64) -> S3Result { + i64::try_from(size) + .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "source object size exceeds the content-length range")) +} + +/// Maps a source body onto the s3s output. Only what the source can vouch +/// for is returned: the ETag and Last-Modified are the source's (the object +/// is not local yet); no version id, SSE headers, storage class or checksums. +fn odm_get_output(head: &SourceHead, content_length: i64, content_range: Option, body: StreamingBlob) -> GetObjectOutput { + GetObjectOutput { + body: Some(body), + content_length: Some(content_length), + content_range, + content_type: head.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()), + content_encoding: head.content_encoding.clone(), + content_disposition: head.content_disposition.clone(), + content_language: head.content_language.clone(), + cache_control: head.cache_control.clone(), + expires: head.expires.clone(), + accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()), + e_tag: head.etag.as_deref().map(to_s3s_etag), + last_modified: head.last_modified.map(OffsetDateTime::from).map(Timestamp::from), + metadata: (!head.user_metadata.is_empty()).then(|| head.user_metadata.clone()), + ..Default::default() + } +} + +/// Feeds the tee primary to the client through a channel: the primary owns a +/// `!Sync` boxed source while the response body must be `Sync`. Dropping the +/// body ends the pump, which drops the primary and hands the remaining +/// source bytes to the tee's drain task (`drain_on_primary_drop`). +fn odm_inline_client_body(primary: TeePrimary) -> StreamingBlob { + let (tx, rx) = tokio::sync::mpsc::channel::>(ODM_INLINE_CLIENT_CHANNEL_CHUNKS); + spawn_traced(async move { + let mut chunks = tokio_util::io::ReaderStream::with_capacity(primary, ODM_SOURCE_BODY_CHUNK_BYTES); + while let Some(chunk) = chunks.next().await { + let failed = chunk.is_err(); + if tx.send(chunk).await.is_err() || failed { + break; + } + } + }); + StreamingBlob::wrap(ReceiverStream::new(rx)) +} + +/// Streams one source GET straight to the client (no local persistence); +/// `backfill` is the background pull the caller queues. +async fn odm_get_passthrough( + state: &Arc, + source: &S, + key: &str, + range: Option<&HTTPRangeSpec>, + backfill: Option, +) -> OdmGetReply { + let get = match odm_observe(state, key, source.get_object(key, range)).await { + Ok(get) => get, + Err(err) => return OdmGetReply::Error(odm_get_source_failure(state, &err)), + }; + let content_length = match odm_content_length(get.head.size) { + Ok(length) => length, + Err(err) => { + state.stats().record_request(OdmOp::Get, OdmOutcome::SourceError); + return OdmGetReply::Error(err); + } + }; + let body = StreamingBlob::wrap(tokio_util::io::ReaderStream::with_capacity( + get.body.into_async_read(), + ODM_SOURCE_BODY_CHUNK_BYTES, + )); + state.stats().record_request(OdmOp::Get, OdmOutcome::SourceHit); + OdmGetReply::Served { + output: Box::new(odm_get_output(&get.head, content_length, get.content_range, body)), + backfill, + } +} + +/// Leader side of an inline pull: one source GET teed between the client +/// (primary) and [`commit_inline`] (secondary, in a background task that +/// survives the request). The write-back result completes the singleflight +/// slot; a failed write-back never affects the client stream. +async fn odm_get_inline( + state: &Arc, + source: &S, + key: &str, + leader: PullLeader, + request_context: Option, +) -> OdmGetReply { + let policy = &state.config().policy; + let tags = if policy.copy_tags { + match odm_observe(state, key, source.get_object_tagging(key)).await { + Ok(tags) => Some(tags), + Err(err) => { + leader.complete(Err(PullError::from(&err))); + return OdmGetReply::Error(odm_get_source_failure(state, &err)); + } + } + } else { + None + }; + let get = match odm_observe(state, key, source.get_object(key, None)).await { + Ok(get) => get, + Err(err) => { + leader.complete(Err(PullError::from(&err))); + return OdmGetReply::Error(odm_get_source_failure(state, &err)); + } + }; + let SourceGet { + head, + body, + content_range, + } = get; + // The object outgrew the inline budget between HEAD and GET: followers + // stream through on their own and the background pull stores it. + if head.size > policy.inline_max_bytes { + leader.complete(Err(PullError::canceled("source object exceeds the inline budget"))); + let content_length = match odm_content_length(head.size) { + Ok(length) => length, + Err(err) => { + state.stats().record_request(OdmOp::Get, OdmOutcome::SourceError); + return OdmGetReply::Error(err); + } + }; + let body = StreamingBlob::wrap(tokio_util::io::ReaderStream::with_capacity( + body.into_async_read(), + ODM_SOURCE_BODY_CHUNK_BYTES, + )); + state.stats().record_request(OdmOp::Get, OdmOutcome::SourceHit); + return OdmGetReply::Served { + output: Box::new(odm_get_output(&head, content_length, content_range, body)), + backfill: Some(PullReason::LargeObject), + }; + } + // `inline_max_bytes` is bounded far below `i64::MAX`. + let content_length = head.size as i64; + let options = TeeOptions { + drain_on_primary_drop: true, + max_drain_bytes: usize::try_from(policy.inline_max_bytes).unwrap_or(usize::MAX), + }; + let (primary, secondary) = tee_reader_with_options(Box::pin(body.into_async_read()), ODM_INLINE_TEE_BUFFER_BYTES, options); + let output = Box::new(odm_get_output(&head, content_length, content_range, odm_inline_client_body(primary))); + let commit_state = Arc::clone(state); + let commit_key = key.to_string(); + spawn_background_with_context(request_context, async move { + let body: WriteBackBody = Box::pin(secondary.into_stream()); + let result = commit_inline(&commit_state, &commit_key, head, tags, body).await; + leader.complete(result.map(|outcome| PullOutcome { + etag: outcome.etag, + size: outcome.size, + })); + }); + state.stats().record_request(OdmOp::Get, OdmOutcome::SourceHit); + OdmGetReply::Served { output, backfill: None } +} + +/// One GET miss against the source (rustfs/backlog#2156). Source HEAD first +/// (size, validators, metadata); conditional headers are evaluated locally +/// against it, never forwarded. Then one of: passthrough for a Range GET or +/// an object above `inline_max_bytes` (plus a queued background pull), or +/// the inline tee for a small object. Concurrent misses of one key share +/// the singleflight slot: only the leader tees; followers re-read local once +/// it commits, or stream through (without queueing) after `first_byte_ms` +/// or when the leader fails. +pub(super) async fn odm_get_from_source( + state: &Arc, + source: &S, + headers: &HeaderMap, + key: &str, + range: Option<&HTTPRangeSpec>, + request_context: Option, +) -> OdmGetReply { + let head = match odm_observe(state, key, source.head_object(key)).await { + Ok(head) => head, + Err(err) => return OdmGetReply::Error(odm_get_source_failure(state, &err)), + }; + let stats = state.stats(); + if let Err(err) = odm_check_source_preconditions(headers, &head) { + stats.record_request(OdmOp::Get, OdmOutcome::SourceHit); + return OdmGetReply::Error(err); + } + let policy = &state.config().policy; + if let Some(range) = range { + let backfill = (policy.range_get == RangeGetPolicy::ServeAndBackfill).then_some(PullReason::RangeGet); + return odm_get_passthrough(state, source, key, Some(range), backfill).await; + } + if head.size > policy.inline_max_bytes { + return odm_get_passthrough(state, source, key, None, Some(PullReason::LargeObject)).await; + } + let slot = match state.acquire_pull_slot(key).await { + Ok(slot) => slot, + // The bucket state was torn down under this request: serve it + // without queueing anything on the old state. + Err(_) => return odm_get_passthrough(state, source, key, None, None).await, + }; + match slot { + PullSlot::Leader(leader) => odm_get_inline(state, source, key, leader, request_context).await, + PullSlot::Follower(follower) => { + let first_byte = Duration::from_millis(policy.source_timeout.first_byte_ms); + match tokio::time::timeout(first_byte, follower.wait()).await { + Ok(Ok(_)) => { + stats.record_request(OdmOp::Get, OdmOutcome::SourceHit); + OdmGetReply::RetryLocal + } + Ok(Err(_)) | Err(_) => odm_get_passthrough(state, source, key, None, None).await, + } + } + } +} + +#[cfg(test)] +mod on_demand_migration_tests { + use super::*; + use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ + BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OdmStateError, OnDemandMigrationConfig, PathStyle, PolicyConfig, + Provider, SourceConfig, SourceCredentials, SourceErrorPolicy, TlsConfig, + }; + use crate::app::storage_api::object_usecase::on_demand_migration::{ + LocalObject, OdmWriteBack, WriteBackError, WriteBackOutcome, WriteBackPart, WriteBackRequest, + }; + use async_trait::async_trait; + use aws_sdk_s3::primitives::ByteStream as AwsByteStream; + use std::collections::VecDeque; + use std::sync::atomic::AtomicBool; + use std::time::SystemTime; + + const KEY: &str = "docs/report.bin"; + + /// A configured, enabled bucket source pointing at an unreachable + /// endpoint; the client is built (no network) and only the scripted + /// source below is ever called. + fn odm_config(policy: PolicyConfig) -> OnDemandMigrationConfig { + OnDemandMigrationConfig { + version: 1, + enabled: true, + source: SourceConfig { + provider: Provider::Minio, + endpoint: Some("https://source.example.invalid:9000".to_string()), + region: "auto".to_string(), + bucket: "legacy".to_string(), + path_style: PathStyle::Auto, + credentials: Some(SourceCredentials { + access_key: "AK".to_string(), + secret_key: "SK".to_string(), + session_token: None, + }), + tls: TlsConfig::default(), + }, + filter: FilterConfig { + prefix: None, + source_prefix: None, + }, + policy, + } + } + + /// Write-back double: collects every committed body, optionally fails. + #[derive(Default)] + struct RecordingWriteBack { + puts: Mutex)>>, + fail_puts: AtomicBool, + } + + impl RecordingWriteBack { + fn puts(&self) -> Vec<(WriteBackRequest, Vec)> { + self.puts.lock().expect("write-back lock").clone() + } + + async fn wait_for_put(&self) -> (WriteBackRequest, Vec) { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Some(put) = self.puts().pop() { + return put; + } + assert!(Instant::now() < deadline, "write-back did not commit in time"); + tokio::time::sleep(Duration::from_millis(5)).await; + } + } + } + + #[async_trait] + impl OdmWriteBack for RecordingWriteBack { + async fn local_object(&self, _bucket: &str, _key: &str) -> Result, WriteBackError> { + Ok(None) + } + + async fn put_object( + &self, + request: &WriteBackRequest, + mut body: WriteBackBody, + ) -> Result { + let mut bytes = Vec::new(); + while let Some(chunk) = body.next().await { + let chunk = chunk.map_err(|err| WriteBackError::Local(err.to_string()))?; + bytes.extend_from_slice(&chunk); + } + if self.fail_puts.load(Ordering::SeqCst) { + return Err(WriteBackError::Local("scripted local failure".to_string())); + } + let size = bytes.len() as u64; + self.puts.lock().expect("write-back lock").push((request.clone(), bytes)); + Ok(WriteBackOutcome { + etag: request.head.etag.clone(), + size, + version_id: None, + }) + } + + async fn create_multipart_upload(&self, _request: &WriteBackRequest) -> Result { + Err(WriteBackError::Local("multipart is not part of the inline path".to_string())) + } + + async fn upload_part( + &self, + _request: &WriteBackRequest, + _upload_id: &str, + _part_number: usize, + _size: u64, + _body: WriteBackBody, + ) -> Result { + Err(WriteBackError::Local("multipart is not part of the inline path".to_string())) + } + + async fn complete_multipart_upload( + &self, + _request: &WriteBackRequest, + _upload_id: &str, + _parts: Vec, + ) -> Result { + Err(WriteBackError::Local("multipart is not part of the inline path".to_string())) + } + + async fn abort_multipart_upload(&self, _bucket: &str, _key: &str, _upload_id: &str) -> Result<(), WriteBackError> { + Ok(()) + } + } + + struct TestRuntime { + sys: OnDemandMigrationSys, + write_back: Arc, + } + + async fn runtime(bucket: &str, policy: PolicyConfig) -> TestRuntime { + let sys = OnDemandMigrationSys::new(); + sys.set_module_enabled(true); + let write_back = Arc::new(RecordingWriteBack::default()); + sys.set_write_back(Arc::clone(&write_back) as Arc); + sys.apply(bucket, Some(&odm_config(policy))).await; + TestRuntime { sys, write_back } + } + + impl TestRuntime { + fn state(&self, bucket: &str) -> Arc { + self.sys.state(bucket).expect("bucket is configured") + } + } + + type ScriptedGet = Result<(SourceHead, Vec, Option), SourceError>; + + struct ScriptedSource { + heads: Mutex>>, + gets: Mutex>, + head_calls: AtomicUsize, + get_calls: AtomicUsize, + tag_calls: AtomicUsize, + ranges: Mutex>>, + } + + impl ScriptedSource { + fn new(heads: Vec>, gets: Vec) -> Self { + Self { + heads: Mutex::new(heads.into_iter().collect()), + gets: Mutex::new(gets.into_iter().collect()), + head_calls: AtomicUsize::new(0), + get_calls: AtomicUsize::new(0), + tag_calls: AtomicUsize::new(0), + ranges: Mutex::new(Vec::new()), + } + } + + fn for_object(body: &[u8]) -> Self { + let head = source_head(body); + Self::new(vec![Ok(head.clone())], vec![Ok((head, body.to_vec(), None))]) + } + + fn head_calls(&self) -> usize { + self.head_calls.load(Ordering::SeqCst) + } + + fn get_calls(&self) -> usize { + self.get_calls.load(Ordering::SeqCst) + } + + fn ranges(&self) -> Vec> { + self.ranges.lock().expect("ranges lock").clone() + } + } + + impl OdmGetSource for ScriptedSource { + async fn head_object(&self, _key: &str) -> Result { + self.head_calls.fetch_add(1, Ordering::SeqCst); + self.heads + .lock() + .expect("heads lock") + .pop_front() + .expect("test script must provide a response for every source HEAD") + } + + async fn get_object(&self, _key: &str, range: Option<&HTTPRangeSpec>) -> Result { + self.get_calls.fetch_add(1, Ordering::SeqCst); + self.ranges + .lock() + .expect("ranges lock") + .push(range.map(|r| (r.is_suffix_length, r.start, r.end))); + let scripted = self + .gets + .lock() + .expect("gets lock") + .pop_front() + .expect("test script must provide a response for every source GET"); + scripted.map(|(head, body, content_range)| SourceGet { + head, + body: AwsByteStream::from(body), + content_range, + }) + } + + async fn get_object_tagging(&self, _key: &str) -> Result, SourceError> { + self.tag_calls.fetch_add(1, Ordering::SeqCst); + Ok(HashMap::from([("team".to_string(), "docs".to_string())])) + } + } + + fn payload(len: usize) -> Vec { + (0..len).map(|index| (index % 251) as u8).collect() + } + + fn source_head(body: &[u8]) -> SourceHead { + SourceHead { + etag: Some(Md5::digest(body).iter().map(|byte| format!("{byte:02x}")).collect()), + size: body.len() as u64, + last_modified: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_767_225_600)), + content_type: Some("application/octet-stream".to_string()), + cache_control: Some("max-age=60".to_string()), + user_metadata: HashMap::from([("owner".to_string(), "alice".to_string())]), + ..Default::default() + } + } + + async fn collect_body(output: &mut GetObjectOutput) -> Vec { + let mut body = output.body.take().expect("source-backed output carries a body"); + let mut bytes = Vec::new(); + while let Some(chunk) = body.next().await { + bytes.extend_from_slice(&chunk.expect("body chunk")); + } + bytes + } + + fn get_count(state: &BucketOdmState, outcome: OdmOutcome) -> u64 { + state.stats().snapshot(state.breaker().state()).requests_total["get"][outcome.as_str()] + } + + fn served(reply: OdmGetReply) -> (GetObjectOutput, Option) { + match reply { + OdmGetReply::Served { output, backfill } => (*output, backfill), + OdmGetReply::Error(err) => panic!("expected Served, got {err:?}"), + OdmGetReply::RetryLocal => panic!("expected Served, got RetryLocal"), + } + } + + fn failed(reply: OdmGetReply) -> S3Error { + match reply { + OdmGetReply::Error(err) => err, + OdmGetReply::Served { .. } => panic!("expected Error, got Served"), + OdmGetReply::RetryLocal => panic!("expected Error, got RetryLocal"), + } + } + + fn consult(sys: &OnDemandMigrationSys, bucket: &str) -> Arc { + match odm_get_verdict(sys.resolve(bucket, KEY).expect("bucket is configured")) { + OdmGetVerdict::Consult { state, .. } => state, + OdmGetVerdict::Fail(err) => panic!("expected Consult, got {err:?}"), + } + } + + fn fail(sys: &OnDemandMigrationSys, bucket: &str) -> S3Error { + match odm_get_verdict(sys.resolve(bucket, KEY).expect("bucket is configured")) { + OdmGetVerdict::Fail(err) => err, + OdmGetVerdict::Consult { .. } => panic!("expected Fail, got Consult"), + } + } + + #[test] + fn odm_get_gate_rejects_part_reads_and_version_reads() { + let plain = ObjectOptions::default(); + assert!(odm_get_may_consult_source(&plain, None)); + assert!(!odm_get_may_consult_source(&plain, Some(1)), "a partNumber read keeps its local 404"); + + let versioned_read = ObjectOptions { + version_id: Some(uuid::Uuid::new_v4().to_string()), + ..Default::default() + }; + assert!(!odm_get_may_consult_source(&versioned_read, None)); + + let proxy_marked = ObjectOptions { + proxy_header_set: true, + ..Default::default() + }; + assert!(!odm_get_may_consult_source(&proxy_marked, None)); + } + + #[tokio::test] + async fn odm_get_inline_streams_to_client_and_commits_the_same_bytes() { + let rt = runtime("b", PolicyConfig::default()).await; + let state = rt.state("b"); + let data = payload(300 * 1024); + let source = ScriptedSource::for_object(&data); + + let (mut output, backfill) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(backfill, None, "inline objects are stored by the tee, not queued"); + assert_eq!(output.content_length, Some(data.len() as i64)); + assert_eq!(output.e_tag, Some(to_s3s_etag(source_head(&data).etag.as_deref().unwrap()))); + assert_eq!( + output.content_type.as_ref().map(|v| v.to_string()), + Some("application/octet-stream".to_string()) + ); + assert_eq!(output.cache_control.as_deref(), Some("max-age=60")); + assert_eq!(output.metadata, Some(HashMap::from([("owner".to_string(), "alice".to_string())]))); + assert_eq!(output.version_id, None, "no x-amz-version-id for a source answer"); + assert_eq!(output.server_side_encryption, None); + assert_eq!(output.storage_class, None); + assert_eq!(output.checksum_sha256, None); + assert_eq!(collect_body(&mut output).await, data, "the client receives the source bytes"); + + let (request, stored) = rt.write_back.wait_for_put().await; + assert_eq!(stored, data, "the local copy is the source bytes"); + assert_eq!(request.bucket, "b"); + assert_eq!(request.key, KEY); + assert_eq!(request.head, source_head(&data)); + assert_eq!(request.tags, None, "copy_tags is off by default"); + assert_eq!(source.get_calls(), 1, "exactly one source GET"); + assert_eq!(source.head_calls(), 1); + assert_eq!(source.tag_calls.load(Ordering::SeqCst), 0); + + // The leader released the key; the next miss would lead again. + let deadline = Instant::now() + Duration::from_secs(5); + while state.inflight_keys() != 0 { + assert!(Instant::now() < deadline, "leader must release the key after the commit"); + tokio::time::sleep(Duration::from_millis(5)).await; + } + assert_eq!(get_count(&state, OdmOutcome::SourceHit), 1); + let snapshot = state.stats().snapshot(state.breaker().state()); + assert_eq!(snapshot.pulled_objects_total["inline"], 1); + assert_eq!(snapshot.pulled_bytes_total, data.len() as u64); + assert_eq!(snapshot.source_latency.count, 2, "HEAD and GET are both observed"); + } + + #[tokio::test] + async fn odm_get_inline_copies_tags_when_configured() { + let rt = runtime( + "t", + PolicyConfig { + copy_tags: true, + ..Default::default() + }, + ) + .await; + let state = rt.state("t"); + let data = payload(1024); + let source = ScriptedSource::for_object(&data); + + let (mut output, _) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(collect_body(&mut output).await, data); + let (request, _) = rt.write_back.wait_for_put().await; + assert_eq!(request.tags, Some(HashMap::from([("team".to_string(), "docs".to_string())]))); + assert_eq!(source.tag_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn odm_get_inline_client_disconnect_still_stores_the_whole_object() { + let rt = runtime("d", PolicyConfig::default()).await; + let state = rt.state("d"); + let data = payload(4 * 1024 * 1024); + let source = ScriptedSource::for_object(&data); + + let (mut output, _) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + let mut body = output.body.take().expect("body"); + let mut received = 0usize; + while received < data.len() / 10 { + let chunk = body.next().await.expect("body chunk").expect("body chunk"); + received += chunk.len(); + } + drop(body); + + let (_, stored) = rt.write_back.wait_for_put().await; + assert_eq!(stored, data, "the drain completes the local copy after the client left"); + assert_eq!(source.get_calls(), 1); + } + + #[tokio::test] + async fn odm_get_inline_write_back_failure_does_not_touch_the_client_stream() { + let rt = runtime("f", PolicyConfig::default()).await; + rt.write_back.fail_puts.store(true, Ordering::SeqCst); + let state = rt.state("f"); + let data = payload(64 * 1024); + let source = ScriptedSource::for_object(&data); + + let (mut output, _) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(collect_body(&mut output).await, data); + + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let snapshot = state.stats().snapshot(state.breaker().state()); + if snapshot.pull_failures_total["local_write"] == 1 { + assert_eq!(snapshot.pulled_objects_total["inline"], 0); + break; + } + assert!(Instant::now() < deadline, "write-back failure must be counted"); + tokio::time::sleep(Duration::from_millis(5)).await; + } + assert!(rt.write_back.puts().is_empty()); + // No retry: the next miss goes to the source again. + assert!(matches!(rt.sys.resolve("f", KEY), Some(OdmLookup::Ready { .. }))); + } + + #[tokio::test] + async fn odm_get_large_object_streams_through_and_queues_a_background_pull() { + let rt = runtime( + "l", + PolicyConfig { + inline_max_bytes: 1024, + ..Default::default() + }, + ) + .await; + let state = rt.state("l"); + let data = payload(4096); + let source = ScriptedSource::for_object(&data); + + let (mut output, backfill) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(backfill, Some(PullReason::LargeObject)); + assert_eq!(output.content_length, Some(4096)); + assert_eq!(collect_body(&mut output).await, data); + assert_eq!(source.ranges(), vec![None], "the whole object is streamed"); + assert_eq!(source.get_calls(), 1); + assert_eq!(get_count(&state, OdmOutcome::SourceHit), 1); + assert_eq!(state.inflight_keys(), 0, "passthrough never takes the singleflight slot"); + tokio::time::sleep(Duration::from_millis(20)).await; + assert!(rt.write_back.puts().is_empty(), "passthrough writes nothing inline"); + } + + #[tokio::test] + async fn odm_get_range_streams_206_and_queues_per_policy() { + let data = payload(10_000); + let range = HTTPRangeSpec { + is_suffix_length: false, + start: 10, + end: 19, + }; + let slice = data[10..20].to_vec(); + let mut slice_head = source_head(&data); + slice_head.size = 10; + let script = || { + ScriptedSource::new( + vec![Ok(source_head(&data))], + vec![Ok((slice_head.clone(), slice.clone(), Some("bytes 10-19/10000".to_string())))], + ) + }; + + let rt = runtime("r", PolicyConfig::default()).await; + let state = rt.state("r"); + let source = script(); + let (mut output, backfill) = + served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, Some(&range), None).await); + assert_eq!(backfill, Some(PullReason::RangeGet), "serve_and_backfill queues the whole object"); + assert_eq!(output.content_range.as_deref(), Some("bytes 10-19/10000")); + assert_eq!(output.content_length, Some(10)); + assert_eq!(collect_body(&mut output).await, slice); + assert_eq!(source.ranges(), vec![Some((false, 10, 19))], "the Range is passed through"); + assert_eq!(source.get_calls(), 1); + assert_eq!(state.inflight_keys(), 0); + + let rt = runtime( + "o", + PolicyConfig { + range_get: RangeGetPolicy::ServeOnly, + ..Default::default() + }, + ) + .await; + let state = rt.state("o"); + let source = script(); + let (_, backfill) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, Some(&range), None).await); + assert_eq!(backfill, None, "serve_only never queues"); + } + + #[tokio::test] + async fn odm_get_conditional_headers_are_answered_from_the_source_head() { + let rt = runtime("c", PolicyConfig::default()).await; + let state = rt.state("c"); + let data = payload(512); + let etag = source_head(&data).etag.expect("etag"); + + let source = ScriptedSource::for_object(&data); + let mut headers = HeaderMap::new(); + headers.insert(http::header::IF_NONE_MATCH, HeaderValue::from_str(&format!("\"{etag}\"")).unwrap()); + let err = failed(odm_get_from_source(&state, &source, &headers, KEY, None, None).await); + assert_eq!(err.code(), &S3ErrorCode::NotModified); + assert_eq!(source.head_calls(), 1); + assert_eq!(source.get_calls(), 0, "a 304 never pulls"); + + let source = ScriptedSource::for_object(&data); + let mut headers = HeaderMap::new(); + headers.insert(http::header::IF_MATCH, HeaderValue::from_static("\"another-etag\"")); + let err = failed(odm_get_from_source(&state, &source, &headers, KEY, None, None).await); + assert_eq!(err.code(), &S3ErrorCode::PreconditionFailed); + assert_eq!(source.get_calls(), 0, "a 412 never pulls"); + assert_eq!(get_count(&state, OdmOutcome::SourceHit), 2); + assert_eq!(state.inflight_keys(), 0); + assert!(rt.write_back.puts().is_empty()); + } + + #[tokio::test] + async fn odm_get_source_not_found_is_404_and_negative_cached() { + let rt = runtime("n", PolicyConfig::default()).await; + let state = rt.state("n"); + let source = ScriptedSource::new(vec![Err(SourceError::NotFound)], vec![]); + + let err = failed(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(get_count(&state, OdmOutcome::SourceMiss), 1); + + let err = fail(&rt.sys, "n"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(get_count(&state, OdmOutcome::NegativeCached), 1); + assert_eq!(source.head_calls(), 1, "the negative cache stops the second miss"); + assert!(matches!(rt.sys.resolve("n", "other"), Some(OdmLookup::Ready { .. }))); + } + + #[tokio::test] + async fn odm_get_unsupported_source_object_is_424() { + let rt = runtime("s", PolicyConfig::default()).await; + let state = rt.state("s"); + let source = ScriptedSource::new( + vec![Err(SourceError::Unsupported( + "source object is encrypted with SSE-C".to_string(), + ))], + vec![], + ); + + let err = failed(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY)); + assert_eq!(err.code(), &S3ErrorCode::Custom(ODM_SOURCE_UNAVAILABLE_CODE.into())); + assert_eq!(err.message(), Some("unsupported")); + assert_eq!(get_count(&state, OdmOutcome::Unsupported), 1); + assert_eq!(source.get_calls(), 0); + } + + #[tokio::test] + async fn odm_get_source_errors_follow_policy_and_open_the_breaker() { + let rt = runtime("e", PolicyConfig::default()).await; + let state = rt.state("e"); + let source = ScriptedSource::new(vec![Err(SourceError::ServerError(503)), Err(SourceError::Timeout)], vec![]); + let err = failed(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY)); + assert_eq!(err.message(), Some("server_error")); + let err = failed(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(err.message(), Some("timeout")); + assert_eq!(get_count(&state, OdmOutcome::SourceError), 2); + + // A GET whose body fetch fails is a source error too, and releases the slot. + let data = payload(64); + let source = ScriptedSource::new(vec![Ok(source_head(&data))], vec![Err(SourceError::AccessDenied)]); + let err = failed(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(err.message(), Some("access_denied")); + assert_eq!(state.inflight_keys(), 0, "a failed leader releases the key"); + + let hidden = runtime( + "h", + PolicyConfig { + source_error: SourceErrorPolicy::NotFound, + ..Default::default() + }, + ) + .await; + let hidden_state = hidden.state("h"); + let source = ScriptedSource::new(vec![Err(SourceError::ServerError(503))], vec![]); + let err = failed(odm_get_from_source(&hidden_state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + + let breaker = runtime("k", PolicyConfig::default()).await; + let source = ScriptedSource::new( + (0..BREAKER_FAILURE_THRESHOLD) + .map(|_| Err(SourceError::ServerError(503))) + .collect(), + vec![], + ); + for _ in 0..BREAKER_FAILURE_THRESHOLD { + let state = consult(&breaker.sys, "k"); + let _ = failed(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + } + let state = breaker.state("k"); + assert_eq!(state.breaker().state(), BreakerState::Open); + let err = fail(&breaker.sys, "k"); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY), "breaker open propagates"); + assert_eq!(err.message(), Some("breaker_open")); + assert_eq!(get_count(&state, OdmOutcome::BreakerOpen), 1); + } + + #[tokio::test] + async fn odm_get_verdict_reports_an_unusable_client() { + let sys = OnDemandMigrationSys::new(); + sys.set_module_enabled(true); + let mut config = odm_config(PolicyConfig::default()); + config.source.credentials = None; + sys.apply("a", Some(&config)).await; + let lookup = sys.resolve("a", KEY).expect("bucket is configured"); + assert!(matches!( + &lookup, + OdmLookup::Unavailable { + error: OdmStateError::AnonymousUnsupported, + .. + } + )); + let err = fail(&sys, "a"); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY)); + assert_eq!(err.message(), Some("unsupported")); + } + + #[tokio::test] + async fn odm_get_follower_rereads_local_after_the_leader_commits() { + let rt = runtime("g", PolicyConfig::default()).await; + let state = rt.state("g"); + let data = payload(256); + let leader = match state.acquire_pull_slot(KEY).await.expect("slot") { + PullSlot::Leader(leader) => leader, + PullSlot::Follower(_) => panic!("first caller leads"), + }; + let source = ScriptedSource::for_object(&data); + let headers = HeaderMap::new(); + + let follower = odm_get_from_source(&state, &source, &headers, KEY, None, None); + let release = async { + tokio::time::sleep(Duration::from_millis(50)).await; + leader.complete(Ok(PullOutcome { + etag: source_head(&data).etag, + size: data.len() as u64, + })); + }; + let (reply, ()) = tokio::join!(follower, release); + assert!(matches!(reply, OdmGetReply::RetryLocal)); + assert_eq!(source.head_calls(), 1, "the follower still validates the miss against the source HEAD"); + assert_eq!(source.get_calls(), 0, "the follower never pulls"); + assert_eq!(get_count(&state, OdmOutcome::SourceHit), 1); + assert!(rt.write_back.puts().is_empty()); + } + + #[tokio::test] + async fn odm_get_follower_degrades_to_passthrough_on_timeout_or_leader_failure() { + let mut policy = PolicyConfig::default(); + policy.source_timeout.first_byte_ms = 100; + let rt = runtime("w", policy).await; + let state = rt.state("w"); + let data = payload(256); + let leader = match state.acquire_pull_slot(KEY).await.expect("slot") { + PullSlot::Leader(leader) => leader, + PullSlot::Follower(_) => panic!("first caller leads"), + }; + + let source = ScriptedSource::for_object(&data); + let (mut output, backfill) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await); + assert_eq!(backfill, None, "a degraded follower does not queue; the leader stores the object"); + assert_eq!(collect_body(&mut output).await, data); + assert_eq!(source.get_calls(), 1); + + // A leader that fails while the follower waits: the follower streams + // through on its own instead of re-reading local. + let source = ScriptedSource::for_object(&data); + let headers = HeaderMap::new(); + let follower = odm_get_from_source(&state, &source, &headers, KEY, None, None); + let fail_leader = async { + tokio::time::sleep(Duration::from_millis(20)).await; + leader.complete(Err(PullError::canceled("scripted leader failure"))); + }; + let (reply, ()) = tokio::join!(follower, fail_leader); + let (mut output, backfill) = served(reply); + assert_eq!(backfill, None); + assert_eq!(collect_body(&mut output).await, data); + assert_eq!(source.get_calls(), 1); + assert!(rt.write_back.puts().is_empty(), "followers never write back"); + } + + #[tokio::test] + async fn odm_get_concurrent_misses_pull_once() { + let rt = runtime("p", PolicyConfig::default()).await; + let state = rt.state("p"); + let data = payload(128 * 1024); + let source = Arc::new(ScriptedSource::new( + (0..32).map(|_| Ok(source_head(&data))).collect(), + vec![Ok((source_head(&data), data.clone(), None))], + )); + + let mut tasks = Vec::new(); + for _ in 0..32 { + let state = Arc::clone(&state); + let source = Arc::clone(&source); + tasks.push(tokio::spawn(async move { + odm_get_from_source(&state, source.as_ref(), &HeaderMap::new(), KEY, None, None).await + })); + } + let mut leaders = 0; + let mut followers = 0; + for task in tasks { + match task.await.expect("task") { + OdmGetReply::Served { mut output, backfill } => { + assert_eq!(backfill, None); + assert_eq!(collect_body(&mut output).await, data); + leaders += 1; + } + OdmGetReply::RetryLocal => followers += 1, + OdmGetReply::Error(err) => panic!("unexpected error {err:?}"), + } + } + assert_eq!(leaders, 1, "exactly one caller streams from the source"); + assert_eq!(followers, 31); + assert_eq!(source.get_calls(), 1); + let (_, stored) = rt.write_back.wait_for_put().await; + assert_eq!(stored, data); + assert_eq!(rt.write_back.puts().len(), 1, "exactly one local commit"); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/rustfs/src/app/object/shared.rs b/rustfs/src/app/object/shared.rs index 42273a08a..b636fdfa8 100644 --- a/rustfs/src/app/object/shared.rs +++ b/rustfs/src/app/object/shared.rs @@ -15,7 +15,9 @@ //! Cross-cutting helpers shared by the object use-case modules. use super::*; -use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{OdmStateError, PolicyConfig, SourceErrorPolicy}; +use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ + OdmStateError, PolicyConfig, SourceErrorPolicy, SourceHead, +}; pub(super) const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id"; @@ -870,6 +872,21 @@ pub(crate) fn mark_on_demand_migration_response(headers: &mut HeaderMap) { headers.insert(ON_DEMAND_MIGRATION_HEADER, ON_DEMAND_MIGRATION_SOURCE); } +/// Evaluates the request's conditional headers (`If-Match`, `If-None-Match`, +/// `If-Modified-Since`, `If-Unmodified-Since`) against the source's view of +/// the object, with the same S3 semantics the local path applies to +/// `ObjectInfo` (rustfs/backlog#2156). Conditional headers are never +/// forwarded to the source: a 304/412 answered by the source would be +/// indistinguishable from a source failure. +pub(crate) fn odm_check_source_preconditions(headers: &HeaderMap, head: &SourceHead) -> S3Result<()> { + let info = ObjectInfo { + etag: head.etag.clone(), + mod_time: head.last_modified.map(OffsetDateTime::from), + ..Default::default() + }; + check_preconditions(headers, &info) +} + #[cfg(test)] mod tests { use super::*; @@ -1859,4 +1876,72 @@ mod on_demand_migration_tests { mark_on_demand_migration_response(&mut headers); assert_eq!(headers.get("x-rustfs-on-demand-migration").and_then(|v| v.to_str().ok()), Some("source")); } + + fn conditional_source_head() -> SourceHead { + SourceHead { + etag: Some("0123456789abcdef0123456789abcdef".to_string()), + size: 7, + // Thu, 01 Jan 2026 00:00:00 GMT + last_modified: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1_767_225_600)), + ..Default::default() + } + } + + fn headers_with(name: http::header::HeaderName, value: &'static str) -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert(name, HeaderValue::from_static(value)); + headers + } + + #[test] + fn odm_source_preconditions_follow_local_semantics() { + let head = conditional_source_head(); + assert!(odm_check_source_preconditions(&HeaderMap::new(), &head).is_ok()); + + // If-None-Match on the source ETag: 304 carrying the source ETag and Last-Modified. + let err = odm_check_source_preconditions( + &headers_with(http::header::IF_NONE_MATCH, "\"0123456789abcdef0123456789abcdef\""), + &head, + ) + .expect_err("matching If-None-Match is 304"); + assert_eq!(err.code(), &S3ErrorCode::NotModified); + assert_eq!(err.status_code(), Some(StatusCode::NOT_MODIFIED)); + let echoed = err.headers().expect("304 echoes validators"); + assert_eq!( + echoed.get("etag").and_then(|v| v.to_str().ok()), + Some("\"0123456789abcdef0123456789abcdef\"") + ); + assert_eq!( + echoed.get("last-modified").and_then(|v| v.to_str().ok()), + Some("Thu, 01 Jan 2026 00:00:00 GMT") + ); + assert!(odm_check_source_preconditions(&headers_with(http::header::IF_NONE_MATCH, "\"other\""), &head).is_ok()); + + // If-Match on another ETag: 412. + let err = odm_check_source_preconditions(&headers_with(http::header::IF_MATCH, "\"other\""), &head) + .expect_err("mismatching If-Match is 412"); + assert_eq!(err.code(), &S3ErrorCode::PreconditionFailed); + assert!( + odm_check_source_preconditions(&headers_with(http::header::IF_MATCH, "\"0123456789abcdef0123456789abcdef\""), &head) + .is_ok() + ); + + // Date validators compare against the source Last-Modified. + let err = odm_check_source_preconditions( + &headers_with(http::header::IF_MODIFIED_SINCE, "Fri, 02 Jan 2026 00:00:00 GMT"), + &head, + ) + .expect_err("not modified since a later date is 304"); + assert_eq!(err.code(), &S3ErrorCode::NotModified); + let err = odm_check_source_preconditions( + &headers_with(http::header::IF_UNMODIFIED_SINCE, "Wed, 31 Dec 2025 00:00:00 GMT"), + &head, + ) + .expect_err("modified since an earlier date is 412"); + assert_eq!(err.code(), &S3ErrorCode::PreconditionFailed); + + // A source without validators cannot fail a precondition. + let bare = SourceHead::default(); + assert!(odm_check_source_preconditions(&headers_with(http::header::IF_MATCH, "\"other\""), &bare).is_ok()); + } } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 358900aed..3a903e385 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -626,7 +626,7 @@ pub(crate) mod bucket { pub(crate) mod on_demand_migration { pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::source_client::{ - SourceClient, SourceError, SourceHead, + SourceClient, SourceError, SourceGet, SourceHead, }; #[cfg(test)] pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ @@ -635,7 +635,7 @@ pub(crate) mod bucket { }; pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OdmStateError, OnDemandMigrationSys, PolicyConfig, - SourceErrorPolicy, + PullError, PullLeader, PullOutcome, PullReason, PullSlot, RangeGetPolicy, SourceErrorPolicy, commit_inline, }; }