diff --git a/rustfs/src/app/object/head.rs b/rustfs/src/app/object/head.rs index 7294eb088..6ed8de0cd 100644 --- a/rustfs/src/app/object/head.rs +++ b/rustfs/src/app/object/head.rs @@ -15,8 +15,166 @@ //! HeadObject path. use super::*; +use crate::app::storage_api::object_usecase::bucket::on_demand_migration::{ + BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OnDemandMigrationSys, SourceClient, SourceError, SourceHead, +}; + +/// Source HEAD seam for the on-demand migration passthrough: production goes +/// through [`SourceClient`], tests script the answers. +pub(super) trait OdmHeadSource { + async fn head_object(&self, key: &str) -> Result; +} + +impl OdmHeadSource for SourceClient { + async fn head_object(&self, key: &str) -> Result { + SourceClient::head_object(self, key).await + } +} + +/// What the on-demand migration runtime decided for a HEAD miss before any +/// source traffic. +pub(super) enum OdmHeadVerdict { + /// Fall through to the local 404. + Ignore, + /// Answer with this error without touching the source. + Fail(S3Error), + /// Consult the source through `client`. + Consult { + state: Arc, + client: Arc, + }, +} + +/// Applies the bucket policy and the lookup verdict of +/// [`OnDemandMigrationSys::resolve`] to a HEAD miss, recording the outcome +/// for every request that stops here. +pub(super) fn odm_head_verdict(lookup: OdmLookup, miss: OdmLocalMiss) -> OdmHeadVerdict { + let state = Arc::clone(lookup.state()); + let policy = &state.config().policy; + if !odm_policy_admits_miss(policy, miss) { + return OdmHeadVerdict::Ignore; + } + let stats = state.stats(); + let unavailable = |error| { + stats.record_request(OdmOp::Head, OdmOutcome::SourceError); + OdmHeadVerdict::Fail(odm_source_error_response(policy, odm_state_error_class(error))) + }; + match &lookup { + OdmLookup::NegativeCached { .. } => { + stats.record_request(OdmOp::Head, OdmOutcome::NegativeCached); + OdmHeadVerdict::Fail(S3Error::new(S3ErrorCode::NoSuchKey)) + } + OdmLookup::BreakerOpen { .. } => { + stats.record_request(OdmOp::Head, OdmOutcome::BreakerOpen); + OdmHeadVerdict::Fail(S3Error::new(S3ErrorCode::NoSuchKey)) + } + OdmLookup::Unavailable { error, .. } => unavailable(error), + OdmLookup::Ready { .. } => { + if policy.head == HeadPolicy::LocalOnly { + stats.record_request(OdmOp::Head, OdmOutcome::Filtered); + return OdmHeadVerdict::Fail(S3Error::new(S3ErrorCode::NoSuchKey)); + } + match state.client() { + Ok(client) => OdmHeadVerdict::Consult { + client: Arc::clone(client), + state: Arc::clone(&state), + }, + Err(error) => unavailable(error), + } + } + } +} + +/// Maps the source's HEAD 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, no SSE headers and no storage +/// class are reported. +pub(super) fn odm_head_output(head: SourceHead) -> S3Result { + let content_length = i64::try_from(head.size) + .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "source object size exceeds the content-length range"))?; + Ok(HeadObjectOutput { + content_length: Some(content_length), + content_type: head.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()), + content_encoding: head.content_encoding, + content_disposition: head.content_disposition, + content_language: head.content_language, + cache_control: head.cache_control, + expires: head.expires, + 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_some(head.user_metadata), + ..Default::default() + }) +} + +/// One source HEAD: latency, breaker and negative cache go through +/// `observe_source`; the outcome counter and the client-facing error follow +/// the bucket's `source_error` policy. Nothing is written locally. +pub(super) async fn odm_head_from_source( + state: &BucketOdmState, + source: &S, + key: &str, +) -> S3Result { + let started = Instant::now(); + let result = source.head_object(key).await; + state.observe_source(started.elapsed(), key, result.as_ref().err()); + let stats = state.stats(); + match result { + Ok(head) => { + stats.record_request(OdmOp::Head, OdmOutcome::SourceHit); + odm_head_output(head) + } + Err(SourceError::NotFound) => { + stats.record_request(OdmOp::Head, OdmOutcome::SourceMiss); + Err(S3Error::new(S3ErrorCode::NoSuchKey)) + } + Err(err @ SourceError::Unsupported(_)) => { + stats.record_request(OdmOp::Head, OdmOutcome::Unsupported); + Err(odm_source_unavailable_error(err.class_label())) + } + Err(err) => { + stats.record_request(OdmOp::Head, OdmOutcome::SourceError); + Err(odm_source_error_response(&state.config().policy, err.class_label())) + } + } +} impl DefaultObjectUsecase { + /// On-demand migration HEAD passthrough (rustfs/backlog#2155): consulted + /// only after the local lookup and the replication proxy both missed. + /// `None` means the runtime does not intervene and the caller keeps its + /// original 404. The source answer is never written back or queued. + async fn on_demand_migration_head( + bucket: &str, + key: &str, + opts: &ObjectOptions, + miss: OdmLocalMiss, + ) -> Option> { + if !odm_request_may_consult_source(opts) { + return None; + } + let lookup = OnDemandMigrationSys::get().resolve(bucket, key)?; + match odm_head_verdict(lookup, miss) { + OdmHeadVerdict::Ignore => None, + OdmHeadVerdict::Fail(err) => Some(Err(err)), + OdmHeadVerdict::Consult { state, client } => Some(odm_head_from_source(&state, client.as_ref(), key).await), + } + } + + async fn finish_on_demand_migration_head( + req: &S3Request, + bucket: &str, + helper: OperationHelper, + output: HeadObjectOutput, + ) -> S3Result> { + let mut response = wrap_response_with_cors(bucket, &req.method, &req.headers, output).await; + mark_on_demand_migration_response(&mut response.headers); + let result = Ok(response); + let _ = helper.complete(&result); + result + } + /// Serve a HEAD whose local lookup failed with not-found by proxying to /// the bucket's replication targets (MinIO `proxyHeadToRepTarget`). async fn proxy_head_object_to_replication_targets( @@ -157,7 +315,11 @@ impl DefaultObjectUsecase { .map_err(ApiError::from)?; // Modification Points: Explicitly handles get_object_info errors, distinguishing between object absence and other errors - let info = match store.get_object_info(&bucket, &key, &opts).await { + let lookup = store.get_object_info(&bucket, &key, &opts).await; + // Single classification point for the on-demand migration gate + // (rustfs/backlog#2155): a not-found error or a latest delete marker. + let odm_miss = odm_local_miss(lookup.as_ref()); + let info = match lookup { Ok(info) => info, Err(err) => { // If the error indicates the object or its version was not found, return 404 (NoSuchKey) @@ -184,6 +346,11 @@ impl DefaultObjectUsecase { .complete(&result); return result; } + if let Some(miss) = odm_miss + && let Some(result) = Self::on_demand_migration_head(&bucket, &key, &opts, miss).await + { + return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await; + } return Err(S3Error::new(S3ErrorCode::NoSuchKey)); } // Other errors, such as insufficient permissions, still return the original error @@ -192,6 +359,13 @@ impl DefaultObjectUsecase { }; if info.delete_marker { if opts.version_id.is_none() { + // A latest delete marker is a local miss the source may still + // answer when the bucket policy says so. + if let Some(miss) = odm_miss + && let Some(result) = Self::on_demand_migration_head(&bucket, &key, &opts, miss).await + { + return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await; + } return Err(S3Error::new(S3ErrorCode::NoSuchKey)); } return Err(S3Error::new(S3ErrorCode::MethodNotAllowed)); @@ -464,7 +638,490 @@ impl DefaultObjectUsecase { #[cfg(test)] mod 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 http::Method; + use std::collections::VecDeque; + use std::time::SystemTime; + + /// 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, + } + } + + async fn odm_sys(bucket: &str, policy: PolicyConfig) -> OnDemandMigrationSys { + let sys = OnDemandMigrationSys::new(); + sys.set_module_enabled(true); + sys.apply(bucket, Some(&odm_config(policy))).await; + sys + } + + fn head_count(state: &BucketOdmState, outcome: OdmOutcome) -> u64 { + state.stats().snapshot(state.breaker().state()).requests_total["head"][outcome.as_str()] + } + + fn assert_no_head_traffic(state: &BucketOdmState) { + let snapshot = state.stats().snapshot(state.breaker().state()); + assert!( + snapshot.requests_total["head"].values().all(|count| *count == 0), + "HEAD must not have entered the runtime: {:?}", + snapshot.requests_total["head"] + ); + } + + struct ScriptedSource { + responses: Mutex>>, + calls: AtomicUsize, + } + + impl ScriptedSource { + fn new(responses: Vec>) -> Self { + Self { + responses: Mutex::new(responses.into_iter().collect()), + calls: AtomicUsize::new(0), + } + } + + fn calls(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } + } + + impl OdmHeadSource for ScriptedSource { + async fn head_object(&self, _key: &str) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + self.responses + .lock() + .expect("scripted source lock should not be poisoned") + .pop_front() + .expect("test script must provide a response for every source HEAD") + } + } + + fn source_head() -> SourceHead { + SourceHead { + etag: Some("d41d8cd98f00b204e9800998ecf8427e-3".to_string()), + size: 1234, + last_modified: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_445_412_480)), + content_type: Some("text/plain".to_string()), + content_encoding: Some("gzip".to_string()), + content_disposition: Some("attachment; filename=\"a.txt\"".to_string()), + content_language: Some("en".to_string()), + cache_control: Some("max-age=60".to_string()), + expires: Some("Thu, 01 Jan 2026 00:00:00 GMT".to_string()), + user_metadata: HashMap::from([("owner".to_string(), "alice".to_string())]), + version_id: Some("v1".to_string()), + storage_class: Some("STANDARD_IA".to_string()), + sse: Some( + crate::storage::storage_api::ecstore_bucket::on_demand_migration::source_client::SourceSse::Kms { + key_id: Some("key-1".to_string()), + }, + ), + is_multipart_etag: true, + } + } + + fn consult(sys: &OnDemandMigrationSys, bucket: &str, key: &str) -> Arc { + match odm_head_verdict(sys.resolve(bucket, key).expect("bucket is configured"), OdmLocalMiss::NotFound) { + OdmHeadVerdict::Consult { state, .. } => state, + OdmHeadVerdict::Ignore => panic!("expected Consult, got Ignore"), + OdmHeadVerdict::Fail(err) => panic!("expected Consult, got {err:?}"), + } + } + + fn fail(sys: &OnDemandMigrationSys, bucket: &str, key: &str, miss: OdmLocalMiss) -> S3Error { + match odm_head_verdict(sys.resolve(bucket, key).expect("bucket is configured"), miss) { + OdmHeadVerdict::Fail(err) => err, + OdmHeadVerdict::Ignore => panic!("expected Fail, got Ignore"), + OdmHeadVerdict::Consult { .. } => panic!("expected Fail, got Consult"), + } + } + + #[test] + fn odm_head_output_maps_source_fields_and_hides_local_only_headers() { + let output = odm_head_output(source_head()).expect("source HEAD maps"); + assert_eq!(output.content_length, Some(1234)); + assert_eq!(output.content_type.as_ref().map(|v| v.to_string()), Some("text/plain".to_string())); + assert_eq!(output.content_encoding.as_deref(), Some("gzip")); + assert_eq!(output.content_disposition.as_deref(), Some("attachment; filename=\"a.txt\"")); + assert_eq!(output.content_language.as_deref(), Some("en")); + assert_eq!(output.cache_control.as_deref(), Some("max-age=60")); + assert_eq!(output.expires.as_deref(), Some("Thu, 01 Jan 2026 00:00:00 GMT")); + assert_eq!(output.accept_ranges.as_deref(), Some("bytes")); + assert_eq!( + output.e_tag, + Some(ETag::Strong("d41d8cd98f00b204e9800998ecf8427e-3".to_string())), + "the source ETag is returned as-is" + ); + let last_modified: OffsetDateTime = output.last_modified.expect("source Last-Modified is returned").into(); + assert_eq!(last_modified.unix_timestamp(), 1_445_412_480); + 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.ssekms_key_id, None); + assert_eq!(output.sse_customer_algorithm, None); + assert_eq!(output.storage_class, None); + assert_eq!(output.replication_status, None); + + let bare = odm_head_output(SourceHead::default()).expect("empty source HEAD maps"); + assert_eq!(bare.content_length, Some(0)); + assert_eq!(bare.e_tag, None); + assert_eq!(bare.metadata, None, "no metadata header family for an empty map"); + assert_eq!(bare.last_modified, None); + } + + #[tokio::test] + async fn odm_head_source_hit_returns_output_and_writes_nothing_back() { + let sys = odm_sys("b", PolicyConfig::default()).await; + let state = consult(&sys, "b", "k"); + let source = ScriptedSource::new(vec![Ok(source_head())]); + + let output = odm_head_from_source(&state, &source, "k").await.expect("source hit"); + assert_eq!(output.content_length, Some(1234)); + assert_eq!(source.calls(), 1); + assert_eq!(head_count(&state, OdmOutcome::SourceHit), 1); + let snapshot = state.stats().snapshot(state.breaker().state()); + assert_eq!(snapshot.source_latency.count, 1, "source latency is observed once per HEAD"); + assert_eq!(snapshot.queue_depth, 0, "HEAD never queues a pull"); + assert_eq!(snapshot.inflight_pulls, 0); + assert_eq!(state.inflight_keys(), 0); + assert_eq!(snapshot.pulled_objects_total.values().sum::(), 0); + assert!( + matches!(sys.resolve("b", "k"), Some(OdmLookup::Ready { .. })), + "a hit leaves the key consultable" + ); + } + + #[tokio::test] + async fn odm_head_source_not_found_is_404_and_negative_cached() { + let sys = odm_sys("b", PolicyConfig::default()).await; + let state = consult(&sys, "b", "gone"); + let source = ScriptedSource::new(vec![Err(SourceError::NotFound)]); + + let err = odm_head_from_source(&state, &source, "gone").await.expect_err("source 404"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(head_count(&state, OdmOutcome::SourceMiss), 1); + + // The second HEAD stops at the negative cache: no source call. + let err = fail(&sys, "b", "gone", OdmLocalMiss::NotFound); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(head_count(&state, OdmOutcome::NegativeCached), 1); + assert_eq!(source.calls(), 1); + assert!( + matches!(sys.resolve("b", "other"), Some(OdmLookup::Ready { .. })), + "other keys stay consultable" + ); + } + + #[tokio::test] + async fn odm_head_source_errors_follow_policy_and_open_the_breaker() { + let sys = odm_sys("b", PolicyConfig::default()).await; + let state = consult(&sys, "b", "k"); + let source = ScriptedSource::new(vec![ + Err(SourceError::ServerError(503)), + Err(SourceError::Timeout), + Err(SourceError::AccessDenied), + Err(SourceError::Other("boom".to_string())), + ]); + + let err = odm_head_from_source(&state, &source, "k").await.expect_err("503 propagates"); + 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("server_error"), "message carries the class only"); + let err = odm_head_from_source(&state, &source, "k") + .await + .expect_err("timeout propagates"); + assert_eq!(err.message(), Some("timeout")); + let err = odm_head_from_source(&state, &source, "k") + .await + .expect_err("access denied propagates"); + assert_eq!(err.message(), Some("access_denied")); + let err = odm_head_from_source(&state, &source, "k") + .await + .expect_err("other propagates"); + assert_eq!(err.message(), Some("other")); + assert_eq!(head_count(&state, OdmOutcome::SourceError), 4); + assert_eq!(state.stats().last_source_error().map(|e| e.class), Some("other".to_string())); + assert_eq!(state.breaker().state(), BreakerState::Closed, "neutral classes do not score"); + + let hidden_sys = odm_sys( + "h", + PolicyConfig { + source_error: SourceErrorPolicy::NotFound, + ..Default::default() + }, + ) + .await; + let hidden = consult(&hidden_sys, "h", "k"); + let source = ScriptedSource::new(vec![Err(SourceError::ServerError(503))]); + let err = odm_head_from_source(&hidden, &source, "k").await.expect_err("503 hidden"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(head_count(&hidden, OdmOutcome::SourceError), 1); + + // Consecutive scoring failures open the breaker; later HEADs stop + // before the source. + let breaker_sys = odm_sys("c", PolicyConfig::default()).await; + let state = consult(&breaker_sys, "c", "k"); + let source = ScriptedSource::new( + (0..BREAKER_FAILURE_THRESHOLD) + .map(|_| Err(SourceError::ServerError(503))) + .collect(), + ); + for _ in 0..BREAKER_FAILURE_THRESHOLD { + let state = consult(&breaker_sys, "c", "k"); + let err = odm_head_from_source(&state, &source, "k").await.expect_err("503"); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY)); + } + assert_eq!(state.breaker().state(), BreakerState::Open); + let err = fail(&breaker_sys, "c", "k", OdmLocalMiss::NotFound); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(head_count(&state, OdmOutcome::BreakerOpen), 1); + assert_eq!( + source.calls(), + BREAKER_FAILURE_THRESHOLD as usize, + "an open breaker never reaches the source" + ); + } + + #[tokio::test] + async fn odm_head_unsupported_source_object_is_424_regardless_of_policy() { + let sys = odm_sys( + "b", + PolicyConfig { + source_error: SourceErrorPolicy::NotFound, + ..Default::default() + }, + ) + .await; + let state = consult(&sys, "b", "k"); + let source = ScriptedSource::new(vec![Err(SourceError::Unsupported("SSE-C".to_string()))]); + + let err = odm_head_from_source(&state, &source, "k").await.expect_err("unsupported"); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY)); + assert_eq!(err.message(), Some("unsupported")); + assert_eq!(head_count(&state, OdmOutcome::Unsupported), 1); + assert_eq!(head_count(&state, OdmOutcome::SourceError), 0); + assert_eq!(state.breaker().state(), BreakerState::Closed); + } + + #[tokio::test] + async fn odm_head_local_only_policy_is_404_without_source_traffic() { + let sys = odm_sys( + "b", + PolicyConfig { + head: HeadPolicy::LocalOnly, + ..Default::default() + }, + ) + .await; + let err = fail(&sys, "b", "k", OdmLocalMiss::NotFound); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + let state = sys.state("b").expect("configured"); + assert_eq!(head_count(&state, OdmOutcome::Filtered), 1); + assert_eq!(state.stats().snapshot(state.breaker().state()).source_latency.count, 0); + } + + #[tokio::test] + async fn odm_head_verdict_respects_local_delete_marker_by_policy() { + let sys = odm_sys("b", PolicyConfig::default()).await; + let lookup = sys.resolve("b", "k").expect("configured"); + assert!(matches!(odm_head_verdict(lookup, OdmLocalMiss::DeleteMarker), OdmHeadVerdict::Ignore)); + assert_no_head_traffic(&sys.state("b").expect("configured")); + + let sys = odm_sys( + "o", + PolicyConfig { + respect_local_delete_marker: false, + ..Default::default() + }, + ) + .await; + let lookup = sys.resolve("o", "k").expect("configured"); + assert!(matches!( + odm_head_verdict(lookup, OdmLocalMiss::DeleteMarker), + OdmHeadVerdict::Consult { .. } + )); + } + + #[tokio::test] + async fn odm_head_unavailable_client_follows_source_error_policy() { + let sys = OnDemandMigrationSys::new(); + sys.set_module_enabled(true); + let mut cfg = odm_config(PolicyConfig::default()); + cfg.source.credentials = None; + sys.apply("b", Some(&cfg)).await; + let state = sys.state("b").expect("configured"); + assert_eq!(state.client().err(), Some(&OdmStateError::AnonymousUnsupported)); + + let err = fail(&sys, "b", "k", OdmLocalMiss::NotFound); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY)); + assert_eq!(err.message(), Some("unsupported")); + assert_eq!(head_count(&state, OdmOutcome::SourceError), 1); + + cfg.policy.source_error = SourceErrorPolicy::NotFound; + sys.apply("b", Some(&cfg)).await; + let err = fail(&sys, "b", "k", OdmLocalMiss::NotFound); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + } + + fn head_input(bucket: &str, key: &str, version_id: Option) -> S3Request { + let input = HeadObjectInput::builder() + .bucket(bucket.to_string()) + .key(key.to_string()) + .version_id(version_id) + .build() + .unwrap(); + build_request(input, Method::HEAD) + } + + /// Drives `execute_head_object` against a real store with the global + /// runtime configured as `head = local_only`: any HEAD that wrongly + /// enters the runtime shows up as a `filtered` count, so a zero counter + /// proves the gate held. + #[tokio::test] + #[serial_test::serial] + async fn execute_head_object_odm_gate_against_real_store() { + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + + let store = crate::app::gating_test_env::shared_gating_ecstore().await; + if current_app_context().is_none() { + crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await; + } + let usecase = DefaultObjectUsecase::from_global(); + + let bucket = format!("odm-head-{}", Uuid::new_v4().simple()); + store + .make_bucket( + &bucket, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("create versioned ODM test bucket"); + let mut reader = PutObjReader::from_vec(b"present".to_vec()); + store + .put_object( + &bucket, + "present", + &mut reader, + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("put local object"); + + let sys = OnDemandMigrationSys::get(); + sys.set_module_enabled(true); + let mut cfg = odm_config(PolicyConfig { + head: HeadPolicy::LocalOnly, + ..Default::default() + }); + sys.apply(&bucket, Some(&cfg)).await; + let state = sys.state(&bucket).expect("bucket runtime installed"); + + // Local hit: served locally, the runtime is never entered. + let response = Box::pin(usecase.execute_head_object(head_input(&bucket, "present", None))) + .await + .expect("local object is served"); + assert_eq!(response.output.content_length, Some(7)); + assert!(response.headers.get(ON_DEMAND_MIGRATION_HEADER).is_none()); + assert_no_head_traffic(&state); + + // HEAD ?versionId on a missing object never consults the runtime. + let err = Box::pin(usecase.execute_head_object(head_input(&bucket, "missing", Some(Uuid::new_v4().to_string())))) + .await + .expect_err("missing version"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_no_head_traffic(&state); + + // The anti-loop marker (any value) keeps the answer local. + let mut req = head_input(&bucket, "missing", None); + req.headers + .insert("x-minio-source-proxy-request", HeaderValue::from_static("false")); + let err = Box::pin(usecase.execute_head_object(req)).await.expect_err("missing object"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_no_head_traffic(&state); + + // A plain miss reaches the runtime; local_only answers 404 there. + let err = Box::pin(usecase.execute_head_object(head_input(&bucket, "missing", None))) + .await + .expect_err("missing object"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(head_count(&state, OdmOutcome::Filtered), 1); + + // Latest delete marker: respected by default, a miss once overridden. + store + .delete_object( + &bucket, + "present", + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("write delete marker"); + let err = Box::pin(usecase.execute_head_object(head_input(&bucket, "present", None))) + .await + .expect_err("delete marker hides the object"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!( + head_count(&state, OdmOutcome::Filtered), + 1, + "respected delete marker never enters the runtime" + ); + + cfg.policy.respect_local_delete_marker = false; + sys.apply(&bucket, Some(&cfg)).await; + let err = Box::pin(usecase.execute_head_object(head_input(&bucket, "present", None))) + .await + .expect_err("local_only still answers 404"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(head_count(&state, OdmOutcome::Filtered), 2, "overridden delete marker is a miss"); + + // A bucket without a runtime keeps the plain 404. + let plain = format!("odm-head-plain-{}", Uuid::new_v4().simple()); + store + .make_bucket(&plain, &MakeBucketOptions::default()) + .await + .expect("create plain bucket"); + assert!(sys.state(&plain).is_none()); + let err = Box::pin(usecase.execute_head_object(head_input(&plain, "missing", None))) + .await + .expect_err("missing object"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + + sys.remove(&bucket); + } #[tokio::test] async fn execute_head_object_rejects_range_with_part_number() { @@ -479,7 +1136,7 @@ mod tests { let req = build_request(input, Method::HEAD); let usecase = DefaultObjectUsecase::without_context(); - let err = usecase.execute_head_object(req).await.unwrap_err(); + let err = Box::pin(usecase.execute_head_object(req)).await.unwrap_err(); assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); } } diff --git a/rustfs/src/app/object/shared.rs b/rustfs/src/app/object/shared.rs index 8698bb908..42273a08a 100644 --- a/rustfs/src/app/object/shared.rs +++ b/rustfs/src/app/object/shared.rs @@ -15,6 +15,7 @@ //! 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}; pub(super) const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id"; @@ -791,6 +792,84 @@ pub(super) fn object_lock_checks_required_for_state(state: &metadata_sys::Object } } +/// Response header marking a HEAD/GET answered by the on-demand migration +/// source instead of local storage (rustfs/backlog#2155). +pub(crate) const ON_DEMAND_MIGRATION_HEADER: http::HeaderName = http::HeaderName::from_static("x-rustfs-on-demand-migration"); +pub(crate) const ON_DEMAND_MIGRATION_SOURCE: HeaderValue = HeaderValue::from_static("source"); + +/// Custom S3 error code for a source failure surfaced under +/// `policy.source_error = propagate`; carried on HTTP 424. +pub(crate) const ODM_SOURCE_UNAVAILABLE_CODE: &str = "SourceUnavailable"; + +/// How the local lookup missed, as seen by the on-demand migration gate. +/// +/// Only a versioned bucket can report `DeleteMarker`: an unversioned bucket +/// keeps nothing after a delete, so its lookup reports `NotFound` and +/// `policy.respect_local_delete_marker` cannot hold the source back there. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum OdmLocalMiss { + NotFound, + DeleteMarker, +} + +/// Classifies a local `get_object_info` result for the on-demand migration +/// gate. `None` means the object is present or the lookup failed for a +/// reason the source cannot answer (permissions, quorum, corruption). +pub(crate) fn odm_local_miss(lookup: Result<&ObjectInfo, &EcstoreError>) -> Option { + match lookup { + Ok(info) if info.delete_marker => Some(OdmLocalMiss::DeleteMarker), + Ok(_) => None, + Err(err) if is_err_object_not_found(err) || is_err_version_not_found(err) => Some(OdmLocalMiss::NotFound), + Err(_) => None, + } +} + +/// Request-level gate shared by HEAD (ODM-08) and GET (ODM-09): a read of a +/// specific version can only be answered locally, and a request carrying the +/// `source-proxy-request` anti-loop marker (whatever its value) comes from a +/// peer that must be answered locally, never bounced to the source. +pub(crate) fn odm_request_may_consult_source(opts: &ObjectOptions) -> bool { + opts.version_id.is_none() && !opts.proxy_header_set +} + +/// Bucket-policy gate for a local miss: a delete marker is honored as the +/// final answer while `respect_local_delete_marker` is set. +pub(crate) fn odm_policy_admits_miss(policy: &PolicyConfig, miss: OdmLocalMiss) -> bool { + match miss { + OdmLocalMiss::NotFound => true, + OdmLocalMiss::DeleteMarker => !policy.respect_local_delete_marker, + } +} + +/// HTTP 424 with the `SourceUnavailable` code. The message carries only the +/// error class: a source message may echo endpoint or key details. +pub(crate) fn odm_source_unavailable_error(class: &'static str) -> S3Error { + let mut err = S3Error::with_message(S3ErrorCode::Custom(ODM_SOURCE_UNAVAILABLE_CODE.into()), class); + err.set_status_code(StatusCode::FAILED_DEPENDENCY); + err +} + +/// Client-facing error for a source failure other than not-found, under +/// `policy.source_error`. +pub(crate) fn odm_source_error_response(policy: &PolicyConfig, class: &'static str) -> S3Error { + match policy.source_error { + SourceErrorPolicy::Propagate => odm_source_unavailable_error(class), + SourceErrorPolicy::NotFound => S3Error::new(S3ErrorCode::NoSuchKey), + } +} + +/// Metrics/message label for a bucket whose source client could not be built. +pub(crate) fn odm_state_error_class(error: &OdmStateError) -> &'static str { + match error { + OdmStateError::AnonymousUnsupported => "unsupported", + OdmStateError::ClientBuild(_) => "client_build", + } +} + +pub(crate) fn mark_on_demand_migration_response(headers: &mut HeaderMap) { + headers.insert(ON_DEMAND_MIGRATION_HEADER, ON_DEMAND_MIGRATION_SOURCE); +} + #[cfg(test)] mod tests { use super::*; @@ -1680,3 +1759,104 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable); } } + +#[cfg(test)] +mod on_demand_migration_tests { + use super::*; + + #[test] + fn odm_local_miss_classifies_lookup_results() { + let present = ObjectInfo::default(); + assert_eq!(odm_local_miss(Ok(&present)), None, "a present object is never a miss"); + + let marker = ObjectInfo { + delete_marker: true, + ..Default::default() + }; + assert_eq!(odm_local_miss(Ok(&marker)), Some(OdmLocalMiss::DeleteMarker)); + + let not_found = EcstoreError::ObjectNotFound("b".to_string(), "k".to_string()); + assert_eq!(odm_local_miss(Err(¬_found)), Some(OdmLocalMiss::NotFound)); + let version_not_found = EcstoreError::VersionNotFound("b".to_string(), "k".to_string(), "v".to_string()); + assert_eq!(odm_local_miss(Err(&version_not_found)), Some(OdmLocalMiss::NotFound)); + + let other = EcstoreError::MethodNotAllowed; + assert_eq!(odm_local_miss(Err(&other)), None, "non-404 failures are not the source's to answer"); + } + + #[test] + fn odm_request_gate_rejects_version_reads_and_anti_loop_marker() { + let plain = ObjectOptions::default(); + assert!(odm_request_may_consult_source(&plain)); + + let versioned_read = ObjectOptions { + version_id: Some(Uuid::new_v4().to_string()), + ..Default::default() + }; + assert!( + !odm_request_may_consult_source(&versioned_read), + "HEAD ?versionId never consults the source" + ); + + // The marker disables the passthrough by presence alone, so a peer's + // "false" convergence probe is answered locally too. + let proxy_marked = ObjectOptions { + proxy_header_set: true, + proxy_request: false, + ..Default::default() + }; + assert!(!odm_request_may_consult_source(&proxy_marked)); + } + + #[test] + fn odm_policy_gate_honors_delete_marker_only_when_configured() { + let respecting = PolicyConfig::default(); + assert!(respecting.respect_local_delete_marker, "default policy respects local delete markers"); + assert!(odm_policy_admits_miss(&respecting, OdmLocalMiss::NotFound)); + assert!(!odm_policy_admits_miss(&respecting, OdmLocalMiss::DeleteMarker)); + + let overriding = PolicyConfig { + respect_local_delete_marker: false, + ..Default::default() + }; + assert!(odm_policy_admits_miss(&overriding, OdmLocalMiss::DeleteMarker)); + } + + #[test] + fn odm_source_unavailable_error_is_424_with_class_only() { + let err = odm_source_unavailable_error("server_error"); + 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("server_error")); + } + + #[test] + fn odm_source_error_response_follows_policy() { + let propagate = PolicyConfig::default(); + assert_eq!(propagate.source_error, SourceErrorPolicy::Propagate); + let err = odm_source_error_response(&propagate, "timeout"); + assert_eq!(err.status_code(), Some(StatusCode::FAILED_DEPENDENCY)); + assert_eq!(err.message(), Some("timeout")); + + let hide = PolicyConfig { + source_error: SourceErrorPolicy::NotFound, + ..Default::default() + }; + let err = odm_source_error_response(&hide, "timeout"); + assert_eq!(err.code(), &S3ErrorCode::NoSuchKey); + assert_eq!(err.status_code(), Some(StatusCode::NOT_FOUND)); + } + + #[test] + fn odm_state_error_class_is_stable() { + assert_eq!(odm_state_error_class(&OdmStateError::AnonymousUnsupported), "unsupported"); + assert_eq!(odm_state_error_class(&OdmStateError::ClientBuild("tls".to_string())), "client_build"); + } + + #[test] + fn mark_on_demand_migration_response_sets_header() { + let mut headers = HeaderMap::new(); + 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")); + } +} diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index a1a682f6f..c9b1b1c7a 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -613,6 +613,21 @@ 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, + }; + #[cfg(test)] + pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ + BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OnDemandMigrationConfig, PathStyle, Provider, SourceConfig, + SourceCredentials, TlsConfig, + }; + pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{ + BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OdmStateError, OnDemandMigrationSys, PolicyConfig, + SourceErrorPolicy, + }; + } + pub(crate) mod policy_sys { pub(crate) type PolicySys = crate::storage::storage_api::ecstore_bucket::policy_sys::PolicySys; }