diff --git a/CHANGELOG.md b/CHANGELOG.md index ebf882167..4f6601c07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Read paths: an object at or below `policy.inline_max_bytes` (16 MiB by default) is teed to the client and to the local store in a single source read; a larger object or a Range read streams through and a background pull stores the whole object. A HEAD miss is proxied to the source and stores nothing (`policy.head = local_only` disables it). Every source-backed response carries `x-rustfs-on-demand-migration: source` - Protections: a per-source circuit breaker, a per-key negative cache, singleflight per key, a concurrency limit and a bounded pull queue shared by the inline and background paths, an optional bandwidth limit, an anti-loop request marker, and the shared outbound-endpoint (SSRF) policy - Metrics under `rustfs_on_demand_migration_*` (`requests_total`, `pulled_bytes_total`, `pulled_objects_total`, `pull_failures_total`, `inflight_pulls`, `queue_depth`, `source_latency_seconds_*`, `breaker_state`), mirrored per node by the admin status route - - Limitations: listings show only local objects (the source is not merged into `ListObjectsV2`); PUT and DELETE never reach the source; a source object updated after it was pulled is not re-fetched; SSE-C source objects are unsupported and answer 424; `Last-Modified` on a pulled object is the local write time, with the source timestamp kept in metadata + - Listings: `ListObjects` v1 remains local with ordinary key markers. `ListObjectsV2` can merge source objects when `policy.list_through = true`; this is off by default + - Upgrade and rollback: finish upgrading every node before enabling ODM. An rc.5 node that writes bucket configuration drops the ODM fields from metadata; neither a later restart nor moving the service out of ECStore recovers them. Before rollback, disable ODM and securely retain the original full configuration and credentials. After every node returns to a compatible version, restore and validate that configuration. Redacted exports cannot replace the credential backup; source-only objects are unavailable through RustFS while ODM is disabled. See the upgrade and rollback section of `docs/operations/on-demand-migration.md` + - Optional Google dependencies: default and `full` server builds retain native GCS support. `cargo build -p rustfs --no-default-features --features ftps,webdav` excludes Google SDKs while preserving configuration decoding and redaction; native GCS ODM and tier operations require the `gcs` feature. Do not use that build with existing GCS-tiered data + - Limitations: PUT and DELETE never reach the source; a source object updated after it was pulled is not re-fetched; SSE-C source objects are unsupported and answer 424; `Last-Modified` on a pulled object is the local write time, with the source timestamp kept in metadata - **NATS JetStream Publish Path**: Opt-in at-least-once delivery for the NATS notify and audit targets. A NATS Core publish flushes to the connection without awaiting a broker acknowledgement, so an event can be lost across a broker restart or a reconnect after the send queue has already cleared it. A queued event now clears only after the JetStream `PublishAck`, so bucket notifications survive those interruptions. Off by default and byte-identical to the NATS Core path when disabled. - Three configuration keys per target: `JETSTREAM_ENABLE`, `JETSTREAM_STREAM_NAME`, and `JETSTREAM_ACK_TIMEOUT_SECS`, under the `RUSTFS_NOTIFY_NATS_` and `RUSTFS_AUDIT_NATS_` prefixes - Durable store-and-forward with a stable dedup id sent as the `Nats-Msg-Id` header, so a replay after a crash is collapsed by the server duplicate window diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 7b36a1647..17f185b63 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -50,7 +50,7 @@ use uuid::Uuid; /// Opaque bucket configuration notifications for application-owned services. /// `None` withdraws a configuration; consumers validate nonempty bytes. -pub type BucketConfigPublishHook = Box) + Send + Sync>; +pub type BucketConfigPublishHook = Box) + Send + Sync>; pub static BUCKET_CONFIG_PUBLISH_HOOK: std::sync::OnceLock = std::sync::OnceLock::new(); const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60); @@ -405,7 +405,8 @@ fn sync_on_demand_migration(bucket: &str, bm: &BucketMetadata) { hook( bucket, super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, - bm.on_demand_migration_config(), + bm.on_demand_migration_config() + .map(|(bytes, stamp)| (bytes, stamp, bm.bucket_incarnation_id)), ); } } @@ -4374,23 +4375,26 @@ mod tests { const ODM_JSON: &[u8] = br#"{"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#; + type RecordedOdmConfig = Option<(Vec, OffsetDateTime, Uuid)>; + type RecordedOdmHookCall = (String, RecordedOdmConfig); + /// Every `(bucket, config)` the recording hook has seen. Tests filter by /// their own bucket name; the hook is process-wide and set once. - static ODM_HOOK_CALLS: std::sync::Mutex, OffsetDateTime)>)>> = std::sync::Mutex::new(Vec::new()); + static ODM_HOOK_CALLS: std::sync::Mutex> = std::sync::Mutex::new(Vec::new()); fn install_recording_odm_hook() { BUCKET_CONFIG_PUBLISH_HOOK.get_or_init(|| { Box::new(|bucket, config_file, config| { assert_eq!(config_file, super::super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG); - ODM_HOOK_CALLS - .lock() - .unwrap() - .push((bucket.to_string(), config.map(|(bytes, stamp)| (bytes.to_vec(), stamp)))); + ODM_HOOK_CALLS.lock().unwrap().push(( + bucket.to_string(), + config.map(|(bytes, stamp, incarnation)| (bytes.to_vec(), stamp, incarnation)), + )); }) }); } - fn odm_hook_calls(bucket: &str) -> Vec, OffsetDateTime)>> { + fn odm_hook_calls(bucket: &str) -> Vec { ODM_HOOK_CALLS .lock() .unwrap() @@ -4414,18 +4418,21 @@ mod tests { std::fs::create_dir_all(dir.path().join(bucket)).expect("physical bucket should exist"); } + let incarnation = Uuid::new_v4(); let expect_publish = |before: usize, label: &str| { let calls = odm_hook_calls(bucket); assert_eq!(calls.len(), before + 1, "{label} must publish exactly once"); assert_eq!( - calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()), + calls.last().unwrap().as_ref().map(|(bytes, _, _)| bytes.as_slice()), Some(ODM_JSON), "{label} must publish the stored bytes" ); + assert_eq!(calls.last().unwrap().as_ref().map(|(_, _, id)| *id), Some(incarnation)); }; // set (via persist_new_and_set, which installs through `set`). let mut bm = BucketMetadata::new(bucket); + bm.bucket_incarnation_id = incarnation; bm.update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec()) .unwrap(); let writer = BucketMetadataSys::new(ecstore.clone()); @@ -4475,7 +4482,7 @@ mod tests { let calls = odm_hook_calls(bucket); assert_eq!(calls.len(), before + 1); assert_eq!( - calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()), + calls.last().unwrap().as_ref().map(|(bytes, _, _)| bytes.as_slice()), Some(b"not-json".as_slice()), "the application validates opaque config bytes" ); diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 210118cda..c648255ad 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -329,11 +329,11 @@ impl ECStore { /// reuse its result, which is sound because bucket deletion/recreation /// requires the lifecycle WRITE lock and therefore cannot have run while /// any read guard was continuously held. - pub(crate) async fn acquire_bucket_incarnation_fence( + pub async fn acquire_bucket_incarnation_fence( &self, bucket: &str, expected: uuid::Uuid, - ) -> Result { + ) -> Result { let inner = self.acquire_bucket_lifecycle_read_lock(bucket).await?; let pieces = super::bucket_fence::FencePieces { registry: self.bucket_fence_registry.clone(), diff --git a/crates/ecstore/src/store/bucket_fence.rs b/crates/ecstore/src/store/bucket_fence.rs index 1800d53a3..66b1f9d01 100644 --- a/crates/ecstore/src/store/bucket_fence.rs +++ b/crates/ecstore/src/store/bucket_fence.rs @@ -150,7 +150,7 @@ impl BucketFenceRegistry { /// A held bucket lifecycle read lock plus its registration in the fence /// registry. Dropping the guard deregisters it; the memo is cleared when the /// last guard for the bucket drops (or a lost lock is observed). -pub(crate) struct BucketIncarnationFenceGuard { +pub struct BucketIncarnationFenceGuard { inner: Option, registry: Arc, bucket: String, @@ -158,6 +158,14 @@ pub(crate) struct BucketIncarnationFenceGuard { } impl BucketIncarnationFenceGuard { + /// Propagate lifecycle lock loss into the storage commit checks. + /// The caller still owns this guard until the complete write tail drains. + pub fn attach_to_object_options(&self, opts: &mut crate::object_api::ObjectOptions) { + if let Some(guard) = self.namespace_lock_guard() { + opts.add_bucket_lifecycle_lock_guard(guard); + } + } + pub(crate) fn is_lock_lost(&self) -> bool { self.inner.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost) } @@ -346,6 +354,36 @@ mod tests { first_pieces.abandon("b", first.token); } + #[tokio::test] + async fn checkpoint_options_inherit_bucket_fence_lock_loss() { + let lock = NamespaceLock::new("bucket-fence-options".to_string(), Arc::new(LocalClient::new())); + let inner = lock + .acquire_guard(&lock_request("options")) + .await + .expect("acquire") + .expect("quorum"); + let pieces = FencePieces { + registry: Arc::default(), + inner, + }; + let registration = pieces.enter("b"); + let fence = pieces.into_guard("b", registration.token); + let mut opts = crate::object_api::ObjectOptions::default(); + fence.attach_to_object_options(&mut opts); + let inherited = opts + .bucket_lifecycle_lock_fence + .as_ref() + .expect("checkpoint inherits lifecycle guard"); + assert!(!inherited.is_lock_lost()); + tokio::time::timeout( + Duration::from_secs(2), + fence.namespace_lock_guard().expect("held guard").lock_lost_notified(), + ) + .await + .expect("distributed guard expires"); + assert!(inherited.is_lock_lost(), "the actual pre-rename options must observe lifecycle lock loss"); + } + #[test] fn buckets_are_isolated() { let reg = BucketFenceRegistry::default(); diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index b2f2965f6..b54a69438 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -417,6 +417,7 @@ const MAX_UPLOADS_LIST: usize = 10000; mod bucket; mod bucket_fence; pub(crate) use bucket::await_bucket_namespace_operation; +pub use bucket_fence::BucketIncarnationFenceGuard; mod heal; mod heal_walk; pub use heal_walk::HealWalkVersion; diff --git a/docs/architecture/crate-boundaries.md b/docs/architecture/crate-boundaries.md index 6fdca19ab..09a308d8f 100644 --- a/docs/architecture/crate-boundaries.md +++ b/docs/architecture/crate-boundaries.md @@ -87,6 +87,12 @@ The guard requires the documents and section headings listed in its `require_sou ## On-Demand Migration Service +Read-through, backfill and external pull orchestration belong in an application +service under `rustfs/src//`. ECStore owns the storage primitives they +need, including atomic commits, lifecycle locks and on-disk metadata. A service +may use these primitives without moving its provider clients or scheduling +policy into the engine. + `rustfs/src/on_demand_migration/` owns source clients, pull scheduling, list merging, runtime state and backfill orchestration. Its `storage_api.rs` is the only ECStore facade boundary. Object write-back still enters the application's @@ -101,6 +107,13 @@ deployment constraints in the admin use case before the incarnation-fenced metadata update. Backfill reads metadata from its store's instance context and preserves the checkpoint ETag compare-and-set, lease and tail-drained writes. +An ODM runtime is bound to the bucket incarnation published with its metadata, +not just its name. Source reads and write-back reject a different incarnation. +Checkpoint writes hold the user bucket's lifecycle fence through their complete +commit and read-back, even if their caller stops waiting; the storage commit +also observes lock loss. Deleting and recreating a bucket must not let work for +its previous incarnation repopulate objects or checkpoints. + Observability owns its metric DTOs and accepts application snapshot callbacks; it does not depend on the ODM runtime. The application registers both bucket and backfill snapshots during startup, before metadata and metric collection. diff --git a/docs/operations/bucket-metadata-recovery.md b/docs/operations/bucket-metadata-recovery.md index ff513e9db..92227c359 100644 --- a/docs/operations/bucket-metadata-recovery.md +++ b/docs/operations/bucket-metadata-recovery.md @@ -12,7 +12,7 @@ To inspect readable configurations while identifying failures, use the same auth ## Recover unreadable replication targets -MinIO target configuration may be an array or KMS-encrypted data that RustFS cannot decode. Diagnosis preserves the failure instead of interpreting it as an empty target set. +RustFS currently accepts the documented `{"targets": [...]}` object format. It cannot decrypt MinIO KMS-encrypted target metadata. Unreadable target payloads remain failures instead of being interpreted as an empty target set; diagnostic export and replacement import do not add MinIO KMS decryption support. 1. Inspect the diagnostic manifest to identify affected buckets. Preserve a separate backup of the original source configuration and any credentials needed for recovery. 2. Prepare a ZIP containing `/bucket-targets.json` with a valid RustFS replacement, whose top-level shape is `{"targets": [...]}`. Supply the intended target settings and credentials; exported credentials are redacted. Use `{"targets": []}` only when intentionally clearing all targets, and reconcile any replication rules that reference removed targets. diff --git a/rustfs/src/app/bucket_list_through.rs b/rustfs/src/app/bucket_list_through.rs index 586f5b0ae..86497e36f 100644 --- a/rustfs/src/app/bucket_list_through.rs +++ b/rustfs/src/app/bucket_list_through.rs @@ -28,7 +28,7 @@ use super::storage_api::bucket_usecase::StorageObjectOptions; use super::storage_api::bucket_usecase::bucket::versioning_sys::BucketVersioningSys; use super::storage_api::bucket_usecase::contract::list::{ListObjectsV2Info as StorageListObjectsV2Info, ListOperations as _}; use super::storage_api::bucket_usecase::contract::object::ObjectOperations as _; -use super::storage_api::bucket_usecase::s3::{S3Error, S3ErrorCode, S3Result}; +use super::storage_api::bucket_usecase::s3::{S3Error, S3ErrorCode, S3Request, S3Result}; use super::storage_api::bucket_usecase::s3_api::bucket::ListObjectsV2Params; use crate::app::object::shared::{odm_source_unavailable_error, odm_state_error_class}; use crate::error::ApiError; @@ -101,16 +101,38 @@ fn invalid_continuation_token(err: &ListThroughTokenError) -> S3Error { /// bucket has no source, `list_through` is off, or the request carries the /// `source-proxy-request` anti-loop marker and therefore comes from a peer /// that must be answered locally. -pub(crate) fn list_through_state(bucket: &str, headers: &HeaderMap) -> Option> { - if get_header(headers, SUFFIX_SOURCE_PROXY_REQUEST).is_some() { - return None; +pub(crate) async fn list_through_state( + store: &ECStore, + bucket: &str, + req: &S3Request, + params: &ListObjectsV2Params, +) -> S3Result>> { + if get_header(&req.headers, SUFFIX_SOURCE_PROXY_REQUEST).is_some() { + return Ok(None); } let sys = OnDemandMigrationSys::get(); if !sys.is_module_enabled() { - return None; + return Ok(None); } - let state = sys.state(bucket)?; - state.config().policy.list_through.then_some(state) + let Some(state) = sys.state(bucket).filter(|state| state.config().policy.list_through) else { + return Ok(None); + }; + if params.max_keys == 0 + || matches!( + source_list_plan(¶ms.prefix, state.config().filter.prefix.as_deref(), params.delimiter.as_deref()), + SourceListPlan::Skip, + ) + { + return Ok(None); + } + let Some(expected_incarnation) = super::storage_api::bucket_usecase::access::odm_read_generation(req, bucket)? else { + return Ok(None); + }; + let incarnation = store.bucket_incarnation_id(bucket).await.map_err(ApiError::from)?; + if incarnation != expected_incarnation { + return Ok(None); + } + Ok(state.filter_incarnation(incarnation)) } /// A merged page plus whether the source had to be left out of it. @@ -444,10 +466,11 @@ mod tests { use crate::app::bucket_usecase::DefaultBucketUsecase; use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore}; use crate::app::storage_api::bucket_usecase::s3::{ - ListObjectsInput, ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response, XmlSerialize, XmlSerializer, + GetObjectInput, HeadObjectInput, ListObjectsInput, ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response, + XmlSerialize, XmlSerializer, }; use crate::app::storage_api::test::StoragePutObjReader; - use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions}; use crate::app::storage_api::test::contract::object::ObjectIO as _; use crate::on_demand_migration::{ FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, @@ -727,7 +750,12 @@ mod tests { ..Default::default() }, }; - sys.apply(&bucket, Some(&config)).await; + sys.apply_for_incarnation( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket identity"), + Some(&config), + ) + .await; assert!( sys.state(&bucket).expect("ODM state installed").client().is_ok(), "fake source client must build" @@ -797,6 +825,128 @@ mod tests { (result, requests) } + #[test] + #[serial_test::serial] + fn stale_bucket_state_cannot_send_get_head_or_list_to_the_source() { + run_large_stack_test("list-through-incarnation", || async { + temp_env::async_with_vars( + [ + ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")), + ("HTTP_PROXY", None), + ("HTTPS_PROXY", None), + ("ALL_PROXY", None), + ("http_proxy", None), + ("https_proxy", None), + ("all_proxy", None), + ("NO_PROXY", Some("*")), + ("no_proxy", Some("*")), + ], + async { + let (endpoint, server, stop) = list_source(std::iter::repeat(source_xml(None, false, Some("source")))).await; + let (_guard, input) = source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, None).await; + let sys = OnDemandMigrationSys::get(); + let old = sys.state(&input.bucket).expect("original source state"); + let store = shared_gating_ecstore().await; + let get = S3Request { + input: GetObjectInput { + bucket: input.bucket.clone(), + key: "missing".into(), + ..Default::default() + }, + method: http::Method::GET, + uri: http::Uri::from_static("/missing"), + headers: HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + let authorized_generation = + super::super::storage_api::bucket_usecase::access::load_bucket_generation_from_store( + &store, + &get, + &input.bucket, + ) + .await + .expect("capture the identity before authorization"); + store + .delete_bucket( + &input.bucket, + &DeleteBucketOptions { + force: true, + ..Default::default() + }, + ) + .await + .expect("delete original bucket"); + store + .make_bucket(&input.bucket, &MakeBucketOptions::default()) + .await + .expect("recreate bucket"); + let replacement = store + .bucket_incarnation_id(&input.bucket) + .await + .expect("replacement identity"); + assert_ne!(replacement, old.incarnation_id()); + sys.remove(&input.bucket); + let mut without_source = get.clone(); + super::super::storage_api::bucket_usecase::access::prepare_odm_read_generation( + &store, + &mut without_source, + &input.bucket, + ) + .await; + for state_incarnation in [old.incarnation_id(), replacement] { + sys.apply_for_incarnation(&input.bucket, state_incarnation, Some(old.config())) + .await; + // Both a stale runtime and a newly published replacement must reject + // requests already authorized for the deleted incarnation. + for capture in 0..3 { + if capture == 0 && state_incarnation == replacement { + continue; + } + let mut get = if capture == 2 { without_source.clone() } else { get.clone() }; + if capture == 1 { + get.extensions.insert(authorized_generation.clone()); + } + let mut head = get.clone().map_input(|_| HeadObjectInput { + bucket: input.bucket.clone(), + key: "missing".into(), + ..Default::default() + }); + head.method = http::Method::HEAD; + let mut list = get.clone().map_input(|_| input.clone()); + list.uri = http::Uri::from_static("/?list-type=2"); + let usecase = crate::app::object::DefaultObjectUsecase::from_global(); + let get_error = tokio::time::timeout(Duration::from_secs(10), usecase.execute_get_object(get)) + .await + .expect("GET stays local") + .expect_err("local object is absent"); + assert_eq!(*get_error.code(), S3ErrorCode::NoSuchKey); + let head_error = tokio::time::timeout(Duration::from_secs(10), usecase.execute_head_object(head)) + .await + .expect("HEAD stays local") + .expect_err("local object is absent"); + assert_eq!(*head_error.code(), S3ErrorCode::NoSuchKey); + let listing = tokio::time::timeout( + Duration::from_secs(10), + DefaultBucketUsecase::from_global().execute_list_objects_v2(list), + ) + .await + .expect("LIST stays local") + .expect("replacement bucket lists locally"); + assert_eq!(listing.output.key_count, Some(0)); + } + } + stop.cancel(); + assert!(server.await.expect("source server must remain unused").is_empty()); + }, + ) + .await; + }); + } + #[test] #[serial_test::serial] fn list_objects_v1_stays_local_with_xml_safe_key_markers() { diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index fe3b11e45..0459e2ea6 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -20,7 +20,7 @@ use super::storage_api::bucket_usecase::StorageObjectInfo as ObjectInfo; use super::storage_api::bucket_usecase::access::ReqInfo; use super::storage_api::bucket_usecase::access::{ authorize_request, bucket_config_mutation_incarnation, log_list_buckets_iam_implicit_deny, - prepare_list_buckets_iam_authorization, req_info_ref, + prepare_list_buckets_iam_authorization, prepare_odm_read_generation, req_info_ref, }; #[cfg(test)] use super::storage_api::bucket_usecase::bucket::target::BucketTarget; @@ -2729,7 +2729,7 @@ impl DefaultBucketUsecase { async fn execute_list_objects_v2_inner( &self, - req: S3Request, + mut req: S3Request, allow_list_through: bool, ) -> S3Result> { let ListObjectsV2Input { @@ -2742,7 +2742,7 @@ impl DefaultBucketUsecase { prefix, start_after, .. - } = req.input; + } = req.input.clone(); let params = parse_list_objects_v2_params(prefix, delimiter, max_keys, continuation_token, start_after)?; @@ -2757,10 +2757,13 @@ impl DefaultBucketUsecase { // The on-demand migration envelope is decoded whether or not this // bucket still merges: a token handed out under `list_through` must keep // paginating after the policy is turned off (rustfs/backlog#2164). + if allow_list_through { + prepare_odm_read_generation(&store, &mut req, &bucket).await; + } let (merged_token, source_state) = if allow_list_through { ( list_through::decode_list_cursor(params.decoded_continuation_token.as_deref())?, - list_through::list_through_state(&bucket, &req.headers), + list_through::list_through_state(&store, &bucket, &req, ¶ms).await?, ) } else { (None, None) diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index 591c4b4c6..aee2fabb1 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -3843,11 +3843,11 @@ impl DefaultObjectUsecase { 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 sys = OnDemandMigrationSys::get(); + if !sys.is_module_enabled() { + return None; + } + let state = sys.state(bucket).filter(|state| state.matches_prefix(key))?; 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. @@ -3861,6 +3861,24 @@ impl DefaultObjectUsecase { None => return Some(OdmGetOutcome::RetryLocal), } } + let expected_incarnation = match odm_read_generation(req, bucket) { + Ok(Some(incarnation)) => incarnation, + Ok(None) => return None, + Err(err) => return Some(OdmGetOutcome::Respond(Err(err))), + }; + match store.bucket_incarnation_id(bucket).await { + Ok(current) if current == expected_incarnation => {} + Ok(_) => return None, + Err(err) => return Some(OdmGetOutcome::Respond(Err(ApiError::from(err).into()))), + } + if !sys.is_module_enabled() { + return None; + } + let lookup = state.filter_incarnation(expected_incarnation)?.resolve_key(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 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 { @@ -3895,7 +3913,7 @@ impl DefaultObjectUsecase { result } - async fn execute_get_object_inner(&self, req: S3Request) -> S3Result> { + async fn execute_get_object_inner(&self, mut req: S3Request) -> S3Result> { let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event(); if let Some(context) = &self.context { @@ -3981,6 +3999,8 @@ impl DefaultObjectUsecase { return Self::complete_get_object_error(helper, err); } }; + let bucket = req.input.bucket.clone(); + prepare_odm_read_generation(&store, &mut req, &bucket).await; if let Some(request_context_start) = request_context_start { rustfs_io_metrics::record_get_object_stage_duration( "s3_handler", @@ -4919,7 +4939,7 @@ mod on_demand_migration_tests { 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> { + async fn abort_multipart_upload(&self, _request: &WriteBackRequest, _upload_id: &str) -> Result<(), WriteBackError> { Ok(()) } } diff --git a/rustfs/src/app/object/head.rs b/rustfs/src/app/object/head.rs index 7cb38bbbc..f20eb895a 100644 --- a/rustfs/src/app/object/head.rs +++ b/rustfs/src/app/object/head.rs @@ -146,6 +146,8 @@ impl DefaultObjectUsecase { /// `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( + req: &S3Request, + store: &ECStore, bucket: &str, key: &str, opts: &ObjectOptions, @@ -154,7 +156,33 @@ impl DefaultObjectUsecase { if !odm_request_may_consult_source(opts) { return None; } - let lookup = OnDemandMigrationSys::get().resolve(bucket, key)?; + let sys = OnDemandMigrationSys::get(); + if !sys.is_module_enabled() { + return None; + } + let state = sys.state(bucket).filter(|state| state.matches_prefix(key))?; + let policy = &state.config().policy; + if !odm_policy_admits_miss(policy, miss) { + return None; + } + if policy.head == HeadPolicy::LocalOnly { + state.stats().record_request(OdmOp::Head, OdmOutcome::Filtered); + return None; + } + let expected_incarnation = match odm_read_generation(req, bucket) { + Ok(Some(incarnation)) => incarnation, + Ok(None) => return None, + Err(err) => return Some(Err(err)), + }; + match store.bucket_incarnation_id(bucket).await { + Ok(current) if current == expected_incarnation => {} + Ok(_) => return None, + Err(err) => return Some(Err(ApiError::from(err).into())), + } + if !sys.is_module_enabled() { + return None; + } + let lookup = state.filter_incarnation(expected_incarnation)?.resolve_key(key)?; match odm_head_verdict(lookup, miss) { OdmHeadVerdict::Ignore => None, OdmHeadVerdict::Fail(err) => Some(Err(err)), @@ -269,7 +297,7 @@ impl DefaultObjectUsecase { } #[instrument(level = "debug", skip(self, req))] - pub async fn execute_head_object(&self, req: S3Request) -> S3Result> { + pub async fn execute_head_object(&self, mut req: S3Request) -> S3Result> { if let Some(context) = &self.context { let _ = context.object_store(); } @@ -314,6 +342,8 @@ impl DefaultObjectUsecase { .await .map_err(ApiError::from)?; + prepare_odm_read_generation(&store, &mut req, &bucket).await; + // Modification Points: Explicitly handles get_object_info errors, distinguishing between object absence and other errors let lookup = store.get_object_info(&bucket, &key, &opts).await; // Single classification point for the on-demand migration gate @@ -347,7 +377,7 @@ impl DefaultObjectUsecase { return result; } if let Some(miss) = odm_miss - && let Some(result) = Self::on_demand_migration_head(&bucket, &key, &opts, miss).await + && let Some(result) = Self::on_demand_migration_head(&req, &store, &bucket, &key, &opts, miss).await { return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await; } @@ -362,7 +392,7 @@ impl DefaultObjectUsecase { // 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 + && let Some(result) = Self::on_demand_migration_head(&req, &store, &bucket, &key, &opts, miss).await { return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await; } @@ -1053,7 +1083,12 @@ mod tests { head: HeadPolicy::LocalOnly, ..Default::default() }); - sys.apply(&bucket, Some(&cfg)).await; + sys.apply_for_incarnation( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + Some(&cfg), + ) + .await; let state = sys.state(&bucket).expect("bucket runtime installed"); // Local hit: served locally, the runtime is never entered. @@ -1109,7 +1144,12 @@ mod tests { ); cfg.policy.respect_local_delete_marker = false; - sys.apply(&bucket, Some(&cfg)).await; + sys.apply_for_incarnation( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + Some(&cfg), + ) + .await; let err = Box::pin(usecase.execute_head_object(head_input(&bucket, "present", None))) .await .expect_err("local_only still answers 404"); diff --git a/rustfs/src/app/object/internal_put.rs b/rustfs/src/app/object/internal_put.rs index bfd01429f..eb93d2610 100644 --- a/rustfs/src/app/object/internal_put.rs +++ b/rustfs/src/app/object/internal_put.rs @@ -48,6 +48,8 @@ use http::HeaderName; /// other internal provenance is written verbatim. pub(crate) struct InternalPutContext { pub(crate) bucket: String, + /// Pins background work to its original bucket across deletion and recreation. + pub(crate) expected_bucket_incarnation_id: Option, pub(crate) key: String, /// Plaintext object length. The single-object path requires it, exactly /// like S3 PutObject rejects an unknown `Content-Length`. @@ -239,6 +241,7 @@ impl DefaultObjectUsecase { let start_time = Instant::now(); let InternalPutContext { bucket, + expected_bucket_incarnation_id, key, size, expected_md5_hex, @@ -296,6 +299,7 @@ impl DefaultObjectUsecase { principal_id, emit_events, preserve_delete_marker, + expected_bucket_incarnation_id, }, }; let committed = self @@ -367,6 +371,8 @@ impl DefaultObjectUsecase { .await .map_err(ApiError::from)?; + opts.expected_bucket_incarnation_id = ctx.expected_bucket_incarnation_id; + let dsc = must_replicate_object( &ctx.bucket, &ctx.key, @@ -428,7 +434,10 @@ impl DefaultObjectUsecase { let bucket = ctx.bucket.as_str(); let key = ctx.key.as_str(); let store = self.object_store().ok_or_else(not_initialized)?; - let mut opts = ObjectOptions::default(); + let mut opts = ObjectOptions { + expected_bucket_incarnation_id: ctx.expected_bucket_incarnation_id, + ..Default::default() + }; let session = store .get_multipart_info(bucket, key, upload_id, &opts) .await @@ -542,6 +551,7 @@ impl DefaultObjectUsecase { } let mut opts = get_complete_multipart_upload_opts_with_replication_authorization(&headers, false).map_err(ApiError::from)?; + opts.expected_bucket_incarnation_id = ctx.expected_bucket_incarnation_id; opts.preserve_etag = ctx.preserve_etag.clone(); opts.preserve_delete_marker = ctx.preserve_delete_marker; let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; @@ -699,10 +709,24 @@ impl DefaultObjectUsecase { } /// Discard an internal multipart upload and its staged parts. - pub(crate) async fn internal_abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), ApiError> { + pub(crate) async fn internal_abort_multipart_upload( + &self, + bucket: &str, + key: &str, + upload_id: &str, + expected_bucket_incarnation_id: Option, + ) -> Result<(), ApiError> { let store = self.object_store().ok_or_else(not_initialized)?; store - .abort_multipart_upload(bucket, key, upload_id, &ObjectOptions::default()) + .abort_multipart_upload( + bucket, + key, + upload_id, + &ObjectOptions { + expected_bucket_incarnation_id, + ..Default::default() + }, + ) .await .map_err(ApiError::from)?; rustfs_scanner::record_dirty_usage_bucket(bucket); @@ -756,6 +780,7 @@ mod tests { fn internal_context(bucket: &str, key: &str, body: &[u8]) -> InternalPutContext { InternalPutContext { bucket: bucket.to_string(), + expected_bucket_incarnation_id: None, key: key.to_string(), size: Some(body.len() as u64), expected_md5_hex: Some(md5_hex(body)), @@ -1138,9 +1163,14 @@ mod tests { )) .await .expect("part of the aborted upload must stage"); - Box::pin(usecase.internal_abort_multipart_upload(&bucket, &ctx.key, &aborted_upload_id)) - .await - .expect("internal abort must succeed"); + Box::pin(usecase.internal_abort_multipart_upload( + &bucket, + &ctx.key, + &aborted_upload_id, + ctx.expected_bucket_incarnation_id, + )) + .await + .expect("internal abort must succeed"); let uploads = Box::pin(store.list_multipart_uploads(&bucket, &ctx.key, None, None, None, 100)) .await .expect("list multipart uploads after abort"); diff --git a/rustfs/src/app/object/mod.rs b/rustfs/src/app/object/mod.rs index d5751d5c4..b9ea67269 100644 --- a/rustfs/src/app/object/mod.rs +++ b/rustfs/src/app/object/mod.rs @@ -21,8 +21,8 @@ use crate::storage_api::table::get_bucket_metadata; use super::storage_api::object_usecase::access::{ PostObjectRequestMarker, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard, authorize_request, - has_bypass_governance_header, load_bucket_generation_from_store, recursive_force_delete_is_authorized, - replication_request_authorized, req_info_mut, req_info_ref, + has_bypass_governance_header, load_bucket_generation_from_store, odm_read_generation, prepare_odm_read_generation, + recursive_force_delete_is_authorized, replication_request_authorized, req_info_mut, req_info_ref, }; #[cfg(test)] use super::storage_api::object_usecase::bucket::quota::BucketQuota; diff --git a/rustfs/src/app/object/on_demand_migration_put.rs b/rustfs/src/app/object/on_demand_migration_put.rs index 50e87b5cf..86ba3972a 100644 --- a/rustfs/src/app/object/on_demand_migration_put.rs +++ b/rustfs/src/app/object/on_demand_migration_put.rs @@ -172,6 +172,7 @@ pub(super) async fn write_back_context(request: &WriteBackRequest, single_part: }; InternalPutContext { bucket: request.bucket.clone(), + expected_bucket_incarnation_id: Some(request.bucket_incarnation_id), key: request.key.clone(), size: Some(head.size), expected_md5_hex: single_part.then(|| expected_md5_hex(head)).flatten(), @@ -285,9 +286,9 @@ impl OdmWriteBack for OnDemandMigrationWriteBack { .map_err(write_back_error) } - async fn abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), WriteBackError> { + async fn abort_multipart_upload(&self, request: &WriteBackRequest, upload_id: &str) -> Result<(), WriteBackError> { self.usecase() - .internal_abort_multipart_upload(bucket, key, upload_id) + .internal_abort_multipart_upload(&request.bucket, &request.key, upload_id, Some(request.bucket_incarnation_id)) .await .map_err(write_back_error) } @@ -303,7 +304,7 @@ mod tests { ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration, }; use crate::app::storage_api::test::bucket::utils::serialize; - use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions}; use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata}; use crate::on_demand_migration::{PullFailureReason, SourceSse}; use http::Method; @@ -345,9 +346,10 @@ mod tests { } } - fn request(bucket: &str, key: &str, head: SourceHead) -> WriteBackRequest { + fn request(bucket: &str, bucket_incarnation_id: Uuid, key: &str, head: SourceHead) -> WriteBackRequest { WriteBackRequest { bucket: bucket.to_string(), + bucket_incarnation_id, key: key.to_string(), head, source_label: SOURCE_LABEL.to_string(), @@ -475,7 +477,15 @@ mod tests { let body = b"pulled from the legacy bucket".to_vec(); let head = source_head(&body); let outcome = write_back - .put_object(&request(&bucket, "dir/obj.txt", head.clone()), body_stream(&body)) + .put_object( + &request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "dir/obj.txt", + head.clone(), + ), + body_stream(&body), + ) .await .expect("write-back must commit"); assert_eq!(outcome.etag, head.etag, "single-part source ETag is preserved"); @@ -518,7 +528,12 @@ mod tests { .await .expect("bucket"); let write_back = OnDemandMigrationWriteBack::new(); - let req = request(bucket, "key", source_head(b"source")); + let req = request( + bucket, + store.bucket_incarnation_id(bucket).await.expect("bucket incarnation"), + "key", + source_head(b"source"), + ); assert!(matches!( write_back.put_object(&req, body_stream(b"source")).await, Err(WriteBackError::Unsupported(_)) @@ -534,6 +549,75 @@ mod tests { assert_nothing_left(&store, bucket, "key").await; } + #[tokio::test] + #[serial_test::serial] + async fn stale_write_back_cannot_mutate_a_recreated_bucket() { + let (store, bucket) = write_back_test_bucket("odm-wb-incarnation", false).await; + let old_id = store.bucket_incarnation_id(&bucket).await.expect("old bucket incarnation"); + let stale = request(&bucket, old_id, "object", source_head(b"source")); + let (resume, wait) = tokio::sync::oneshot::channel(); + let delayed = { + let stale = stale.clone(); + tokio::spawn(async move { + wait.await.expect("resume old source pull"); + OnDemandMigrationWriteBack::new() + .put_object(&stale, body_stream(b"source")) + .await + }) + }; + store + .delete_bucket( + &bucket, + &DeleteBucketOptions { + force: true, + ..Default::default() + }, + ) + .await + .expect("delete original bucket"); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("recreate bucket"); + let new_id = store.bucket_incarnation_id(&bucket).await.expect("replacement incarnation"); + assert_ne!(old_id, new_id); + resume.send(()).expect("release delayed pull"); + assert!(delayed.await.expect("delayed pull task").is_err()); + assert_nothing_left(&store, &bucket, "object").await; + + let write_back = OnDemandMigrationWriteBack::new(); + assert!(write_back.create_multipart_upload(&stale).await.is_err()); + let current = request(&bucket, new_id, "object", source_head(b"current")); + let upload = write_back + .create_multipart_upload(¤t) + .await + .expect("create replacement upload"); + // A stale capability must fail independently of whether its upload ID + // happens to name a valid session in the replacement bucket. + assert!( + write_back + .upload_part(&stale, &upload, 1, 6, body_stream(b"source")) + .await + .is_err() + ); + let part = write_back + .upload_part(¤t, &upload, 1, 7, body_stream(b"current")) + .await + .expect("stage current part"); + assert!( + write_back + .complete_multipart_upload(&stale, &upload, vec![part.clone()]) + .await + .is_err() + ); + assert!(write_back.abort_multipart_upload(&stale, &upload).await.is_err()); + write_back + .complete_multipart_upload(¤t, &upload, vec![part]) + .await + .expect("stale cleanup preserves replacement upload"); + assert_eq!(raw_object_bytes(&store, &bucket, "object").await, b"current"); + } + #[tokio::test] #[serial_test::serial] async fn write_back_integrity_failure_leaves_nothing_behind() { @@ -543,7 +627,15 @@ mod tests { head.etag = Some(md5_hex(b"a different body")); let err = OnDemandMigrationWriteBack::new() - .put_object(&request(&bucket, "wrong.bin", head), body_stream(&body)) + .put_object( + &request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "wrong.bin", + head, + ), + body_stream(&body), + ) .await .expect_err("an ETag mismatch must fail the write-back"); assert_eq!(err, WriteBackError::Integrity); @@ -559,8 +651,18 @@ mod tests { let (store, bucket) = write_back_test_bucket("odm-wb-race", versioned).await; let source = b"old source bytes"; let client = b"new client bytes"; - let req = request(&bucket, "race", source_head(source)); - let client_req = request(&bucket, "race", source_head(client)); + let req = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "race", + source_head(source), + ); + let client_req = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "race", + source_head(client), + ); let mut client_ctx = write_back_context(&client_req, true).await; client_ctx.if_absent = false; let client_after = PutObjectCommitBarrier::install(&bucket, "race", PutObjectCommitPause::AfterNamespace); @@ -595,13 +697,27 @@ mod tests { async fn write_back_multipart_completion_preserves_a_client_put_after_staging() { let (store, bucket) = write_back_test_bucket("odm-mpu-race", false).await; let write_back = OnDemandMigrationWriteBack::new(); - let req = request(&bucket, "race", source_head(b"source")); + let req = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "race", + source_head(b"source"), + ); let upload_id = write_back.create_multipart_upload(&req).await.expect("create"); let part = write_back .upload_part(&req, &upload_id, 1, 6, body_stream(b"source")) .await .expect("stage"); - let mut client_ctx = write_back_context(&request(&bucket, "race", source_head(b"client")), true).await; + let mut client_ctx = write_back_context( + &request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "race", + source_head(b"client"), + ), + true, + ) + .await; client_ctx.if_absent = false; let committed = DefaultObjectUsecase::from_global() .internal_put_object(client_ctx, body_stream(b"client")) @@ -613,7 +729,7 @@ mod tests { "{result:?}" ); write_back - .abort_multipart_upload(&bucket, "race", &upload_id) + .abort_multipart_upload(&req, &upload_id) .await .expect("abort rejected upload"); let stored = stored_object(&store, &bucket, "race").await; @@ -628,7 +744,12 @@ mod tests { for multipart in [false, true] { let (store, bucket) = write_back_test_bucket("odm-wb-tombstone", true).await; let write_back = OnDemandMigrationWriteBack::new(); - let mut req = request(&bucket, "deleted", source_head(b"source")); + let mut req = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "deleted", + source_head(b"source"), + ); let staged = if multipart { let id = write_back.create_multipart_upload(&req).await.expect("create"); let part = write_back @@ -654,10 +775,7 @@ mod tests { assert!(marker.delete_marker); let rejected = if let Some((id, part)) = staged { let result = write_back.complete_multipart_upload(&req, &id, vec![part]).await; - write_back - .abort_multipart_upload(&bucket, "deleted", &id) - .await - .expect("abort"); + write_back.abort_multipart_upload(&req, &id).await.expect("abort"); result } else { write_back.put_object(&req, body_stream(b"source")).await @@ -691,7 +809,15 @@ mod tests { Err(io::Error::new(io::ErrorKind::BrokenPipe, "tee primary dropped before EOF")), ]); let err = OnDemandMigrationWriteBack::new() - .put_object(&request(&bucket, "torn.bin", head.clone()), torn) + .put_object( + &request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "torn.bin", + head.clone(), + ), + torn, + ) .await .expect_err("a broken stream must fail the write-back"); assert_ne!(err, WriteBackError::Integrity, "{err}"); @@ -700,7 +826,15 @@ mod tests { // A clean EOF short of the advertised size is just as fatal. let short = stream(vec![Ok(Bytes::copy_from_slice(&body[..64 * 1024]))]); let err = OnDemandMigrationWriteBack::new() - .put_object(&request(&bucket, "short.bin", head), short) + .put_object( + &request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "short.bin", + head, + ), + short, + ) .await .expect_err("a short body must fail the write-back"); assert!(matches!(err, WriteBackError::Local(_) | WriteBackError::Integrity), "{err}"); @@ -716,7 +850,12 @@ mod tests { let mut head = source_head(&body); head.etag = Some(format!("{}-2", md5_hex(&body))); head.is_multipart_etag = true; - let request = request(&bucket, "big/object.bin", head.clone()); + let request = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "big/object.bin", + head.clone(), + ); let write_back = OnDemandMigrationWriteBack::new(); let upload_id = write_back.create_multipart_upload(&request).await.expect("create"); @@ -754,10 +893,7 @@ mod tests { .upload_part(&request, &aborted, 1, 4096, body_stream(&body[..4096])) .await .expect("stage part"); - write_back - .abort_multipart_upload(&bucket, "big/object.bin", &aborted) - .await - .expect("abort"); + write_back.abort_multipart_upload(&request, &aborted).await.expect("abort"); let uploads = store .list_multipart_uploads(&bucket, "big/object.bin", None, None, None, 100) .await @@ -810,7 +946,12 @@ mod tests { let body = b"plaintext that must be encrypted at rest".to_vec(); let head = source_head(&body); - let request = request(&bucket, "secret.txt", head.clone()); + let request = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "secret.txt", + head.clone(), + ); // The source ETag is not forced onto an encrypted object; the // local ETag is whatever the SSE write path computes. assert_eq!(write_back_context(&request, true).await.preserve_etag, None); @@ -848,7 +989,15 @@ mod tests { let (store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("odm-wb-quota", 64).await; let body = vec![0x71; 4096]; let err = OnDemandMigrationWriteBack::new() - .put_object(&request(&bucket, "over.bin", source_head(&body)), body_stream(&body)) + .put_object( + &request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "over.bin", + source_head(&body), + ), + body_stream(&body), + ) .await .expect_err("a full quota must reject the write-back"); assert!(matches!(err, WriteBackError::Quota(_)), "{err}"); @@ -913,7 +1062,12 @@ mod tests { let body = b"replicate me".to_vec(); let head = source_head(&body); - let request = request(&bucket, "replicated.txt", head); + let request = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "replicated.txt", + head, + ); let ctx = write_back_context(&request, true).await; assert!(ctx.emit_events, "policy.emit_events reaches the creation event"); assert_eq!(ctx.principal_id, ON_DEMAND_MIGRATION_PRINCIPAL_ID); @@ -961,7 +1115,12 @@ mod tests { head.user_metadata .insert(forged_replica_key.to_string(), ReplicationStatusType::Replica.as_str().to_string()); - let mut write_request = request(&bucket, "unadmitted.txt", head); + let mut write_request = request( + &bucket, + store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"), + "unadmitted.txt", + head, + ); write_request.tags = Some(HashMap::from([("replicate".to_string(), "no".to_string())])); OnDemandMigrationWriteBack::new() .put_object(&write_request, body_stream(&body)) @@ -1059,7 +1218,7 @@ mod tests { #[test] fn provenance_and_tags_are_stable() { - let mut request = request("b", "k", source_head(b"x")); + let mut request = request("b", Uuid::nil(), "k", source_head(b"x")); let metadata = provenance_metadata(&request); assert_eq!(metadata.len(), 10, "five keys under two prefixes"); assert_provenance(&metadata, &request.head); @@ -1080,7 +1239,7 @@ mod tests { #[tokio::test] async fn write_back_context_applies_the_etag_and_event_policy() { let body = b"context".to_vec(); - let mut request = request("no-such-bucket", "k", source_head(&body)); + let mut request = request("no-such-bucket", Uuid::nil(), "k", source_head(&body)); let ctx = write_back_context(&request, true).await; assert_eq!(ctx.expected_md5_hex, Some(md5_hex(&body))); assert_eq!(ctx.preserve_etag, Some(md5_hex(&body))); diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index 443605791..f1873beea 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -953,6 +953,7 @@ pub(super) enum PutObjectOrigin<'a> { principal_id: &'static str, emit_events: bool, preserve_delete_marker: bool, + expected_bucket_incarnation_id: Option, }, } @@ -967,7 +968,13 @@ impl PutObjectOrigin<'_> { fn apply_bucket_generation_guard(&self, bucket: &str, opts: &mut ObjectOptions) -> S3Result<()> { match self { Self::S3 { req, .. } => apply_bucket_generation_guard(req, bucket, opts), - Self::Internal { .. } => Ok(()), + Self::Internal { + expected_bucket_incarnation_id, + .. + } => { + opts.expected_bucket_incarnation_id = *expected_bucket_incarnation_id; + Ok(()) + } } } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index da7af7c86..4e5ae8cc4 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -27,19 +27,20 @@ pub(crate) fn EndpointServerPools( /// S3 wire types for app-layer modules, funneled here so new files stay off /// the direct s3s surface (s3s footprint ratchet, `scripts/check_s3s_footprint.sh`). pub(crate) mod s3 { + #[cfg(test)] + pub(crate) use s3s::S3Response; #[cfg(test)] pub(crate) use s3s::dto::ListObjectsInput; #[cfg(test)] pub(crate) use s3s::dto::{ - BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ListObjectsV2Input, - ListObjectsV2Output, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, - ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration, + BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, GetObjectInput, + HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output, ReplicationConfiguration, ReplicationRule, + ReplicationRuleFilter, ReplicationRuleStatus, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, + ServerSideEncryptionRule, Tag, VersioningConfiguration, }; #[cfg(test)] pub(crate) use s3s::xml::{Serialize as XmlSerialize, Serializer as XmlSerializer}; - pub(crate) use s3s::{S3Error, S3ErrorCode, S3Result}; - #[cfg(test)] - pub(crate) use s3s::{S3Request, S3Response}; + pub(crate) use s3s::{S3Error, S3ErrorCode, S3Request, S3Result}; } pub(crate) mod admin { @@ -266,8 +267,9 @@ pub(crate) mod access { pub(crate) use crate::storage::storage_api::access_consumer::{ PostObjectRequestMarker, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard, authorize_request, bucket_config_mutation_incarnation, has_bypass_governance_header, load_bucket_generation_from_store, - log_list_buckets_iam_implicit_deny, prepare_list_buckets_iam_authorization, recursive_force_delete_is_authorized, - replication_request_authorized, req_info_mut, req_info_ref, + log_list_buckets_iam_implicit_deny, odm_read_generation, prepare_list_buckets_iam_authorization, + prepare_odm_read_generation, recursive_force_delete_is_authorized, replication_request_authorized, req_info_mut, + req_info_ref, }; } diff --git a/rustfs/src/on_demand_migration/backfill.rs b/rustfs/src/on_demand_migration/backfill.rs index f2d14b61d..28bb85219 100644 --- a/rustfs/src/on_demand_migration/backfill.rs +++ b/rustfs/src/on_demand_migration/backfill.rs @@ -370,6 +370,8 @@ pub type PullReport = Option; /// mock in unit tests. Production: [`BucketBackfillContext`]. #[async_trait] pub trait BackfillContext: Send + Sync { + /// The bucket incarnation captured by this context. + fn incarnation_id(&self) -> Uuid; /// One source page in the local key namespace. async fn list_page(&self, prefix: Option<&str>, token: Option<&str>, max_keys: i32) -> Result; /// Whether the breaker admits source traffic right now. @@ -411,6 +413,10 @@ impl BucketBackfillContext { #[async_trait] impl BackfillContext for BucketBackfillContext { + fn incarnation_id(&self) -> Uuid { + self.state.incarnation_id() + } + async fn list_page(&self, prefix: Option<&str>, token: Option<&str>, max_keys: i32) -> Result { let client = self.state.client().map_err(|err| SourceError::Unsupported(err.to_string()))?; let started = Instant::now(); @@ -661,8 +667,36 @@ pub async fn read_checkpoint(api: &Arc, bucket: &str) -> Result, bucket: &str, + incarnation_id: Uuid, checkpoint: &BackfillCheckpoint, expected_etag: Option<&str>, +) -> Result { + let api = Arc::clone(api); + let bucket = bucket.to_string(); + let checkpoint = checkpoint.clone(); + let expected_etag = expected_etag.map(str::to_string); + // The storage commit owns detached work. Keep its user-bucket fence alive + // even when a caller aborts its waiter before the erasure tail has drained. + tokio::spawn(async move { + let fence = api.acquire_bucket_incarnation_fence(&bucket, incarnation_id).await?; + let mut opts = ObjectOptions::default(); + fence.attach_to_object_options(&mut opts); + let result = write_checkpoint_while_fenced(&api, &bucket, &checkpoint, expected_etag.as_deref(), opts).await; + drop(fence); + result + }) + .await + .map_err(|err| StorageError::other(format!("backfill checkpoint task failed: {err}")))? +} + +/// The caller holds the destination bucket's lifecycle fence through the CAS +/// write and its read-back, including the drained erasure write tail. +async fn write_checkpoint_while_fenced( + api: &Arc, + bucket: &str, + checkpoint: &BackfillCheckpoint, + expected_etag: Option<&str>, + mut opts: ObjectOptions, ) -> Result { let data = checkpoint.to_json()?; let preconditions = match expected_etag { @@ -675,12 +709,9 @@ async fn write_checkpoint( ..Default::default() }, }; - let opts = ObjectOptions { - max_parity: true, - write_completion: WriteCompletion::TailDrained, - http_preconditions: Some(preconditions), - ..Default::default() - }; + opts.max_parity = true; + opts.write_completion = WriteCompletion::TailDrained; + opts.http_preconditions = Some(preconditions); match save_config_with_opts(Arc::clone(api), &checkpoint_path(bucket), data, &opts).await { Ok(()) => {} Err(StorageError::PreconditionFailed) => return Err(BackfillError::Conflict(bucket.to_string())), @@ -851,7 +882,14 @@ impl BackfillRunner { }); } let checkpoint = BackfillCheckpoint::new(&request, config_updated_at, &self.node, now); - let etag = write_checkpoint(&self.api, bucket, &checkpoint, stored.as_ref().map(|s| s.etag.as_str())).await?; + let etag = write_checkpoint( + &self.api, + bucket, + context.incarnation_id(), + &checkpoint, + stored.as_ref().map(|s| s.etag.as_str()), + ) + .await?; info!( event = EVENT_ODM_BACKFILL_STATE, component = LOG_COMPONENT_ECSTORE, @@ -886,30 +924,42 @@ impl BackfillRunner { } return Ok(handle.snapshot.lock().clone()); } - let _lock = self.lease_lock(bucket, get_lock_acquire_timeout()).await?; - let Some(stored) = read_checkpoint(&self.api, bucket).await? else { - return Err(BackfillError::NotFound(bucket.to_string())); - }; - if !stored.checkpoint.state.is_active() { - return Ok(stored.checkpoint); - } - let mut checkpoint = stored.checkpoint; - let now = OffsetDateTime::now_utc(); - checkpoint.state = BackfillState::Cancelled; - checkpoint.updated_at = now; - write_checkpoint(&self.api, bucket, &checkpoint, Some(&stored.etag)).await?; - info!( - event = EVENT_ODM_BACKFILL_STATE, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, - state = checkpoint.state.as_str(), - result = "cancelled", - bucket = %bucket, - job_id = %checkpoint.job_id, - owner = %checkpoint.owner.as_ref().map(|o| o.node.as_str()).unwrap_or_default(), - "On-demand migration backfill job cancelled remotely" - ); - Ok(checkpoint) + let incarnation_id = self.api.bucket_incarnation_id_from_disk(bucket).await?; + let lock = self.lease_lock(bucket, get_lock_acquire_timeout()).await?; + let api = Arc::clone(&self.api); + let bucket = bucket.to_string(); + tokio::spawn(async move { + let _lock = lock; + let fence = api.acquire_bucket_incarnation_fence(&bucket, incarnation_id).await?; + let mut opts = ObjectOptions::default(); + fence.attach_to_object_options(&mut opts); + let Some(stored) = read_checkpoint(&api, &bucket).await? else { + return Err(BackfillError::NotFound(bucket.to_string())); + }; + if !stored.checkpoint.state.is_active() { + return Ok(stored.checkpoint); + } + let mut checkpoint = stored.checkpoint; + let now = OffsetDateTime::now_utc(); + checkpoint.state = BackfillState::Cancelled; + checkpoint.updated_at = now; + write_checkpoint_while_fenced(&api, &bucket, &checkpoint, Some(&stored.etag), opts).await?; + info!( + event = EVENT_ODM_BACKFILL_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION, + state = checkpoint.state.as_str(), + result = "cancelled", + bucket = %bucket, + job_id = %checkpoint.job_id, + owner = %checkpoint.owner.as_ref().map(|o| o.node.as_str()).unwrap_or_default(), + "On-demand migration backfill job cancelled remotely" + ); + drop(fence); + Ok(checkpoint) + }) + .await + .map_err(|err| StorageError::other(format!("backfill cancellation task failed: {err}")))? } /// Latest checkpoint: the in-memory progress of a local job, else the @@ -1002,7 +1052,7 @@ impl BackfillRunner { checkpoint.state = BackfillState::Cancelled; checkpoint.updated_at = now; checkpoint.record_failure("config_changed", None, now); - write_checkpoint(&self.api, bucket, &checkpoint, Some(&stored.etag)).await?; + write_checkpoint(&self.api, bucket, context.incarnation_id(), &checkpoint, Some(&stored.etag)).await?; info!( event = EVENT_ODM_BACKFILL_STATE, component = LOG_COMPONENT_ECSTORE, @@ -1021,7 +1071,7 @@ impl BackfillRunner { node: self.node.clone(), lease_until: now + BACKFILL_LEASE, }); - let etag = write_checkpoint(&self.api, bucket, &checkpoint, Some(&stored.etag)).await?; + let etag = write_checkpoint(&self.api, bucket, context.incarnation_id(), &checkpoint, Some(&stored.etag)).await?; warn!( event = EVENT_ODM_BACKFILL_LEASE_TAKEOVER, component = LOG_COMPONENT_ECSTORE, @@ -1460,7 +1510,8 @@ impl Job { lease_until: now + BACKFILL_LEASE, }); } - let etag = write_checkpoint(&self.api, &self.bucket, &self.checkpoint, Some(&self.etag)).await?; + let etag = + write_checkpoint(&self.api, &self.bucket, self.context.incarnation_id(), &self.checkpoint, Some(&self.etag)).await?; self.etag = etag; self.keys_since_save = 0; self.last_save = Instant::now(); @@ -1500,7 +1551,10 @@ pub async fn run_backfill_recovery_loop(runner: Arc, cancel: Can #[cfg(test)] mod tests { - use super::super::storage_api::test_support::isolated_store_over_temp_disks; + use super::super::storage_api::test_support::{ + BUCKET_LIFECYCLE_LOCK_OBJECT, BucketOperations as _, PutObjectCommitBarrier, PutObjectCommitPause, + isolated_store_over_temp_disks, + }; use super::*; use crate::on_demand_migration::source_client::SourceObject; use crate::on_demand_migration::sys::PullError; @@ -1626,6 +1680,7 @@ mod tests { /// Scripted source + local store + queue with a controllable report path. struct MockContext { + incarnation_id: Mutex>, objects: Vec, page_size: usize, local: Mutex>, @@ -1654,6 +1709,7 @@ mod tests { }) .collect(); Arc::new(Self { + incarnation_id: Mutex::new(None), objects, page_size, local: Mutex::new(HashMap::new()), @@ -1688,6 +1744,10 @@ mod tests { #[async_trait] impl BackfillContext for MockContext { + fn incarnation_id(&self) -> Uuid { + self.incarnation_id.lock().expect("test bucket initialized") + } + async fn list_page(&self, prefix: Option<&str>, token: Option<&str>, max_keys: i32) -> Result { if let Some(err) = self.list_error.lock().take() { return Err(err); @@ -1780,12 +1840,17 @@ mod tests { context: Arc, ) -> (Vec, Arc, Arc) { let (dirs, store) = isolated_store_over_temp_disks().await; - // The isolated store has no bucket metadata system; the checkpoint - // only needs the bucket's directory under the metadata volume. - for dir in &dirs { - std::fs::create_dir_all(dir.path().join(RUSTFS_META_BUCKET).join(BUCKET_META_PREFIX).join(bucket)) - .expect("test bucket metadata directory"); - } + super::super::storage_api::test_support::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await; + store + .make_bucket(bucket, &Default::default()) + .await + .expect("create test bucket"); + *context.incarnation_id.lock() = Some( + store + .bucket_incarnation_id_from_disk(bucket) + .await + .expect("test bucket identity"), + ); let runner = runner_on(node, bucket, context, Arc::clone(&store)); (dirs, store, runner) } @@ -1795,6 +1860,129 @@ mod tests { BackfillRunner::new(store, node, Arc::new(contexts)) } + #[tokio::test] + async fn cancelled_checkpoint_waiter_keeps_bucket_fenced_until_commit_finishes() { + for (suffix, pause) in [ + ("before", PutObjectCommitPause::BeforeQuotaRename), + ("after", PutObjectCommitPause::AfterRenameQuorum), + ] { + let bucket = format!("backfill-cancel-tail-{suffix}"); + let context = MockContext::new(0, 1); + let (_dirs, store, _runner) = runner_with("node-a", &bucket, Arc::clone(&context)).await; + let original_incarnation = context.incarnation_id(); + let checkpoint = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", ts(1_700_000_001)); + let barrier = PutObjectCommitBarrier::install(RUSTFS_META_BUCKET, &checkpoint_path(&bucket), pause); + let writer_api = Arc::clone(&store); + let writer_bucket = bucket.clone(); + let waiter = tokio::spawn(async move { + write_checkpoint(&writer_api, &writer_bucket, original_incarnation, &checkpoint, None).await + }); + barrier.wait_until_paused().await; + waiter.abort(); + assert!(waiter.await.expect_err("caller aborted").is_cancelled()); + + let lifecycle_lock = store + .new_ns_lock(&bucket, BUCKET_LIFECYCLE_LOCK_OBJECT) + .await + .expect("lifecycle lock"); + { + let mut probe = Box::pin(lifecycle_lock.get_write_lock(Duration::from_secs(1))); + assert!( + futures::poll!(probe.as_mut()).is_pending(), + "lifecycle writer must first try to acquire the lock" + ); + assert!( + tokio::time::timeout(Duration::from_millis(100), probe.as_mut()) + .await + .is_err(), + "the checkpoint owner must retain the user bucket lifecycle read lock after caller cancellation" + ); + } + + let delete_api = Arc::clone(&store); + let delete_bucket = bucket.clone(); + let mut deletion = tokio::spawn(async move { delete_api.delete_bucket(&delete_bucket, &Default::default()).await }); + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut deletion).await.is_err(), + "DeleteBucket must wait for the checkpoint owner after its caller aborts" + ); + barrier.release(); + tokio::time::timeout(Duration::from_secs(10), deletion) + .await + .expect("commit must drain and release its lifecycle guard") + .expect("delete task") + .expect("delete original bucket"); + store + .make_bucket(&bucket, &Default::default()) + .await + .expect("recreate bucket"); + assert_ne!( + original_incarnation, + store.bucket_incarnation_id_from_disk(&bucket).await.expect("new identity") + ); + assert!( + read_checkpoint(&store, &bucket) + .await + .expect("read recreated bucket") + .is_none(), + "no old checkpoint may outlive bucket deletion" + ); + } + } + + #[tokio::test] + async fn stale_checkpoint_writer_cannot_resurrect_or_overwrite_a_recreated_bucket() { + let bucket = "backfill-incarnation"; + let context = MockContext::new(0, 1); + let (_dirs, store, _runner) = runner_with("node-a", bucket, Arc::clone(&context)).await; + let old_incarnation = context.incarnation_id(); + let old = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", ts(1_700_000_001)); + let old_etag = write_checkpoint(&store, bucket, old_incarnation, &old, None) + .await + .expect("old checkpoint"); + + store + .delete_bucket(bucket, &Default::default()) + .await + .expect("delete original bucket"); + store.make_bucket(bucket, &Default::default()).await.expect("recreate bucket"); + let current_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.expect("new identity"); + assert_ne!(old_incarnation, current_incarnation); + assert!( + read_checkpoint(&store, bucket) + .await + .expect("read after recreation") + .is_none() + ); + for expected_etag in [None, Some(old_etag.as_str())] { + let error = write_checkpoint(&store, bucket, old_incarnation, &old, expected_etag) + .await + .expect_err("stale writer rejected"); + assert!(matches!(error, BackfillError::Storage(StorageError::BucketNotFound(_)))); + } + assert!( + read_checkpoint(&store, bucket) + .await + .expect("stale writer left no checkpoint") + .is_none() + ); + + let current = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-b", ts(1_700_000_002)); + let current_etag = write_checkpoint(&store, bucket, current_incarnation, ¤t, None) + .await + .expect("current checkpoint"); + let error = write_checkpoint(&store, bucket, old_incarnation, &old, Some(¤t_etag)) + .await + .expect_err("old identity cannot overwrite a matching ETag"); + assert!(matches!(error, BackfillError::Storage(StorageError::BucketNotFound(_)))); + let stored = read_checkpoint(&store, bucket) + .await + .expect("read current checkpoint") + .expect("current checkpoint remains"); + assert_eq!(stored.etag, current_etag); + assert_eq!(stored.checkpoint, current); + } + #[tokio::test] async fn full_backfill_lists_pages_and_counts_every_key() { let bucket = "backfill-full"; @@ -2134,7 +2322,7 @@ mod tests { node: "node-a".to_string(), lease_until: now - Duration::from_secs(120), }); - let etag = write_checkpoint(&store, bucket, &crashed, None) + let etag = write_checkpoint(&store, bucket, context.incarnation_id(), &crashed, None) .await .expect("seed checkpoint"); @@ -2145,7 +2333,7 @@ mod tests { lease_until: now + Duration::from_secs(60), }); live.updated_at = now; - let etag = write_checkpoint(&store, bucket, &live, Some(&etag)) + let etag = write_checkpoint(&store, bucket, context.incarnation_id(), &live, Some(&etag)) .await .expect("live lease"); assert_eq!(runner.recover_once().await.taken_over, 0, "unexpired lease must not be taken over"); @@ -2162,7 +2350,7 @@ mod tests { lease_until: now - Duration::from_secs(1), }); expired.updated_at = now + Duration::from_millis(1); - write_checkpoint(&store, bucket, &expired, Some(&etag)) + write_checkpoint(&store, bucket, context.incarnation_id(), &expired, Some(&etag)) .await .expect("expire lease"); let stats = runner.recover_once().await; @@ -2201,7 +2389,7 @@ mod tests { crashed.continuation_token = Some("2".to_string()); crashed.failed = 1; crashed.record_failure("local_write", Some("k/00002"), crashed_at); - write_checkpoint(&store, bucket, &crashed, None) + write_checkpoint(&store, bucket, context.incarnation_id(), &crashed, None) .await .expect("seed failed page with an expired lease"); @@ -2257,7 +2445,9 @@ mod tests { // Same node name, unexpired lease: only a restart can produce this. let own = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", now); - let etag = write_checkpoint(&store, bucket, &own, None).await.expect("seed"); + let etag = write_checkpoint(&store, bucket, context.incarnation_id(), &own, None) + .await + .expect("seed"); assert_eq!(runner.recover_once().await.taken_over, 1, "own-node running job is reclaimed at once"); runner.wait_until_idle(bucket).await; let cp = read_checkpoint(&store, bucket) @@ -2275,7 +2465,7 @@ mod tests { node: "node-z".to_string(), lease_until: now - Duration::from_secs(1), }); - write_checkpoint(&store, bucket, &stale, Some(&stored.etag)) + write_checkpoint(&store, bucket, context.incarnation_id(), &stale, Some(&stored.etag)) .await .expect("seed stale"); let stats = runner.recover_once().await; diff --git a/rustfs/src/on_demand_migration/pull.rs b/rustfs/src/on_demand_migration/pull.rs index 9d52af009..9e7525642 100644 --- a/rustfs/src/on_demand_migration/pull.rs +++ b/rustfs/src/on_demand_migration/pull.rs @@ -243,6 +243,8 @@ impl PullSource for SourceClient { #[derive(Clone, Debug)] pub struct WriteBackRequest { pub bucket: String, + /// Identity captured with the source configuration, retained through cleanup. + pub bucket_incarnation_id: uuid::Uuid, pub key: String, /// Source HEAD/GET of the whole object. pub head: SourceHead, @@ -263,6 +265,7 @@ impl WriteBackRequest { let config = state.config(); Self { bucket: state.bucket().to_string(), + bucket_incarnation_id: state.incarnation_id(), key: key.to_string(), head, source_label: format!("{}:{}", config.source.provider.as_str(), config.source.bucket), @@ -357,7 +360,7 @@ pub trait OdmWriteBack: Send + Sync { parts: Vec, ) -> Result; - async fn abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), WriteBackError>; + async fn abort_multipart_upload(&self, request: &WriteBackRequest, upload_id: &str) -> Result<(), WriteBackError>; } /// Why the pump stopped feeding the write-back before EOF. @@ -662,9 +665,7 @@ async fn write_multipart( Err(err) => Err(err), }; if completed.is_err() - && let Err(abort_err) = write_back - .abort_multipart_upload(&request.bucket, &request.key, &upload_id) - .await + && let Err(abort_err) = write_back.abort_multipart_upload(request, &upload_id).await { debug!( event = EVENT_ODM_PULL_FAILED, @@ -1340,7 +1341,7 @@ mod tests { }) } - async fn abort_multipart_upload(&self, _bucket: &str, _key: &str, upload_id: &str) -> Result<(), WriteBackError> { + async fn abort_multipart_upload(&self, _request: &WriteBackRequest, upload_id: &str) -> Result<(), WriteBackError> { self.aborted.lock().push(upload_id.to_string()); Ok(()) } diff --git a/rustfs/src/on_demand_migration/sys.rs b/rustfs/src/on_demand_migration/sys.rs index d529594d4..8fbde7fa4 100644 --- a/rustfs/src/on_demand_migration/sys.rs +++ b/rustfs/src/on_demand_migration/sys.rs @@ -23,9 +23,8 @@ //! path (initial load, admin update, peer reload, refresh loop, lazy load). //! //! Change detection compares the config by value (`PartialEq`) rather than -//! by `updated_at`: the hook does not carry the timestamp, fetching it would -//! re-enter the metadata system from inside its own publish path, and a -//! byte-identical config never needs a new client anyway. +//! by `updated_at`. The bucket incarnation is part of this comparison: +//! recreating a bucket must cancel old work even with identical configuration. //! //! Client construction is async (TLS material may be read from disk), so //! the hook does not build inline: `publish` removes state synchronously and @@ -270,6 +269,7 @@ impl Drop for InflightEntryGuard<'_> { /// config change (counters excepted), removed when the config goes away. pub struct BucketOdmState { bucket: String, + incarnation_id: uuid::Uuid, config: OnDemandMigrationConfig, applied_at: OffsetDateTime, endpoint_host: String, @@ -307,6 +307,7 @@ impl BucketOdmState { async fn build( bucket: &str, config: &OnDemandMigrationConfig, + incarnation_id: uuid::Uuid, stats: Arc, write_back: Option>, ) -> Arc { @@ -323,6 +324,7 @@ impl BucketOdmState { let policy = &config.policy; Arc::new(Self { bucket: bucket.to_string(), + incarnation_id, endpoint_host: endpoint_host(&config.source), config: config.clone(), applied_at: OffsetDateTime::now_utc(), @@ -340,10 +342,18 @@ impl BucketOdmState { }) } + pub fn filter_incarnation(self: Arc, incarnation_id: uuid::Uuid) -> Option> { + (self.incarnation_id == incarnation_id && !self.is_cancelled()).then_some(self) + } + pub fn bucket(&self) -> &str { &self.bucket } + pub fn incarnation_id(&self) -> uuid::Uuid { + self.incarnation_id + } + pub fn config(&self) -> &OnDemandMigrationConfig { &self.config } @@ -742,16 +752,17 @@ impl OnDemandMigrationSys { BUCKET_CONFIG_PUBLISH_HOOK .set(Box::new(move |bucket, config_file, stored| { if config_file == BUCKET_ON_DEMAND_MIGRATION_CONFIG { - self.publish_stored(bucket, stored.map(|(bytes, _)| bytes)); + self.publish_stored(bucket, stored.map(|(bytes, _, incarnation)| (bytes, incarnation))); } })) .is_ok() } /// Corrupt persisted bytes withdraw state synchronously, just like deletion. - fn publish_stored(&'static self, bucket: &str, stored: Option<&[u8]>) { - match stored.map(OnDemandMigrationConfig::from_json).transpose() { - Ok(config) => self.publish(bucket, config.as_ref()), + fn publish_stored(&'static self, bucket: &str, stored: Option<(&[u8], uuid::Uuid)>) { + let incarnation_id = stored.map(|(_, id)| id).unwrap_or_default(); + match stored.map(|(bytes, _)| OnDemandMigrationConfig::from_json(bytes)).transpose() { + Ok(config) => self.publish_for_incarnation(bucket, incarnation_id, config.as_ref()), Err(err) => { warn!( event = EVENT_ODM_BUCKET_STATE_APPLIED, @@ -762,7 +773,7 @@ impl OnDemandMigrationSys { error = %err, "Failed to parse on-demand migration config" ); - self.publish(bucket, None); + self.publish_for_incarnation(bucket, incarnation_id, None); } } } @@ -770,14 +781,19 @@ impl OnDemandMigrationSys { /// Hook entry point: removals apply immediately, installs are spawned /// (client construction is async). Requires a Tokio runtime for the /// install path; without one the config is logged and skipped. - pub fn publish(&'static self, bucket: &str, config: Option<&OnDemandMigrationConfig>) { - let config = self.desired(config); + pub fn publish_for_incarnation( + &'static self, + bucket: &str, + incarnation_id: uuid::Uuid, + config: Option<&OnDemandMigrationConfig>, + ) { + let config = self.desired(config).filter(|_| !incarnation_id.is_nil()); let generation = self.reserve_generation(bucket, config.is_some()); let Some(config) = config else { self.remove_with_generation(bucket, generation); return; }; - if self.is_unchanged(bucket, config, generation) { + if self.is_unchanged(bucket, incarnation_id, config, generation) { return; } let Ok(handle) = tokio::runtime::Handle::try_current() else { @@ -796,32 +812,49 @@ impl OnDemandMigrationSys { let bucket = bucket.to_string(); let config = config.clone(); handle.spawn(async move { - self.apply_with_generation(&bucket, Some(&config), generation).await; + self.apply_with_generation(&bucket, incarnation_id, Some(&config), generation) + .await; }); } /// Installs, rebuilds, or removes the bucket state for `config`. /// Idempotent: the same config on an installed bucket is a no-op. + #[cfg(test)] pub async fn apply(&self, bucket: &str, config: Option<&OnDemandMigrationConfig>) -> ApplyOutcome { - let config = self.desired(config); + self.apply_for_incarnation(bucket, uuid::Uuid::from_u128(1), config).await + } + + #[cfg(test)] + pub fn publish(&'static self, bucket: &str, config: Option<&OnDemandMigrationConfig>) { + self.publish_for_incarnation(bucket, uuid::Uuid::from_u128(1), config); + } + + pub async fn apply_for_incarnation( + &self, + bucket: &str, + incarnation_id: uuid::Uuid, + config: Option<&OnDemandMigrationConfig>, + ) -> ApplyOutcome { + let config = self.desired(config).filter(|_| !incarnation_id.is_nil()); let generation = self.reserve_generation(bucket, config.is_some()); - self.apply_with_generation(bucket, config, generation).await + self.apply_with_generation(bucket, incarnation_id, config, generation).await } async fn apply_with_generation( &self, bucket: &str, + incarnation_id: uuid::Uuid, config: Option<&OnDemandMigrationConfig>, generation: u64, ) -> ApplyOutcome { let Some(config) = self.desired(config) else { return self.remove_with_generation(bucket, generation); }; - if self.is_unchanged(bucket, config, generation) { + if self.is_unchanged(bucket, incarnation_id, config, generation) { return ApplyOutcome::Unchanged; } let stats = self.state(bucket).map(|state| Arc::clone(&state.stats)).unwrap_or_default(); - let state = BucketOdmState::build(bucket, config, stats, self.write_back()).await; + let state = BucketOdmState::build(bucket, config, incarnation_id, stats, self.write_back()).await; let (outcome, previous) = { let mut buckets = self.buckets.write(); @@ -871,6 +904,7 @@ impl OnDemandMigrationSys { /// One-shot lookup: module switch, bucket state, prefix filter, /// client availability, negative cache, breaker, in that order. + #[cfg(test)] pub fn resolve(&self, bucket: &str, key: &str) -> Option { if !self.is_module_enabled() { return None; @@ -878,6 +912,13 @@ impl OnDemandMigrationSys { self.state(bucket)?.resolve_key(key) } + pub fn resolve_for_incarnation(&self, bucket: &str, key: &str, incarnation_id: uuid::Uuid) -> Option { + if !self.is_module_enabled() { + return None; + } + self.state(bucket)?.filter_incarnation(incarnation_id)?.resolve_key(key) + } + pub fn state(&self, bucket: &str) -> Option> { self.buckets.read().get(bucket).and_then(|slot| slot.state.clone()) } @@ -925,15 +966,26 @@ impl OnDemandMigrationSys { /// Claims `generation` for the bucket when the installed state already /// matches `config` and has a usable client. - fn is_unchanged(&self, bucket: &str, config: &OnDemandMigrationConfig, generation: u64) -> bool { + fn is_unchanged(&self, bucket: &str, incarnation_id: uuid::Uuid, config: &OnDemandMigrationConfig, generation: u64) -> bool { let mut buckets = self.buckets.write(); let Some(slot) = buckets.get_mut(bucket) else { return false; }; + if slot.generation > generation { + return false; + } + if slot + .state + .as_ref() + .is_some_and(|state| state.incarnation_id != incarnation_id) + && let Some(previous) = slot.state.take() + { + previous.cancel.cancel(); + } let unchanged = slot .state .as_ref() - .is_some_and(|state| state.client.is_ok() && state.config == *config); + .is_some_and(|state| state.client.is_ok() && state.incarnation_id == incarnation_id && state.config == *config); if unchanged && slot.generation < generation { slot.generation = generation; } @@ -1366,13 +1418,88 @@ mod tests { assert!(state.is_cancelled()); } + #[tokio::test] + async fn identical_config_on_recreated_bucket_cancels_old_state() { + let sys = enabled_sys(); + let cfg = config(None); + let old_id = uuid::Uuid::new_v4(); + let new_id = uuid::Uuid::new_v4(); + sys.apply_for_incarnation("recreated", old_id, Some(&cfg)).await; + let old = sys.state("recreated").expect("old state installed"); + assert!(sys.resolve_for_incarnation("recreated", "key", new_id).is_none()); + sys.apply_for_incarnation("recreated", new_id, Some(&cfg)).await; + let replacement = sys.state("recreated").expect("replacement state installed"); + assert!(old.is_cancelled()); + assert!(!Arc::ptr_eq(&old, &replacement)); + assert_eq!(replacement.incarnation_id(), new_id); + assert!(sys.resolve_for_incarnation("recreated", "key", old_id).is_none()); + assert!(sys.resolve_for_incarnation("recreated", "key", new_id).is_some()); + } + + #[tokio::test] + async fn changed_delete_marker_policy_withdraws_the_captured_lookup() { + let sys = enabled_sys(); + let incarnation = uuid::Uuid::new_v4(); + let mut cfg = config(None); + cfg.policy.respect_local_delete_marker = false; + sys.apply_for_incarnation("policy-snapshot", incarnation, Some(&cfg)).await; + let captured = sys.state("policy-snapshot").expect("policy A installed"); + assert!(!captured.config().policy.respect_local_delete_marker); + + cfg.policy.respect_local_delete_marker = true; + sys.apply_for_incarnation("policy-snapshot", incarnation, Some(&cfg)).await; + let replacement = sys.state("policy-snapshot").expect("policy B installed"); + assert!(replacement.config().policy.respect_local_delete_marker); + assert!(captured.is_cancelled()); + assert!( + captured + .filter_incarnation(incarnation) + .and_then(|state| state.resolve_key("key")) + .is_none(), + "a request that evaluated policy A cannot continue through policy B" + ); + assert!( + replacement + .clone() + .filter_incarnation(incarnation) + .and_then(|state| state.resolve_key("key")) + .is_some() + ); + assert_eq!( + replacement + .stats() + .snapshot(replacement.breaker().state()) + .source_latency + .count, + 0 + ); + } + + #[tokio::test] + async fn missing_incarnation_cannot_install_or_retain_source_state() { + let sys: &'static OnDemandMigrationSys = Box::leak(Box::new(enabled_sys())); + let cfg = config(None); + assert_eq!( + sys.apply_for_incarnation("missing", uuid::Uuid::nil(), Some(&cfg)).await, + ApplyOutcome::NotDesired + ); + sys.publish_for_incarnation("missing", uuid::Uuid::nil(), Some(&cfg)); + assert!(sys.state("missing").is_none()); + + sys.apply_for_incarnation("missing", uuid::Uuid::new_v4(), Some(&cfg)).await; + let state = sys.state("missing").expect("valid identity installed"); + sys.publish_for_incarnation("missing", uuid::Uuid::nil(), Some(&cfg)); + assert!(sys.state("missing").is_none()); + assert!(state.is_cancelled()); + } + #[tokio::test] async fn corrupt_stored_config_withdraws_runtime_state() { let sys: &'static OnDemandMigrationSys = Box::leak(Box::new(enabled_sys())); let cfg = config(None); assert_eq!(sys.apply("corrupt", Some(&cfg)).await, ApplyOutcome::Installed); let state = sys.state("corrupt").expect("state installed"); - sys.publish_stored("corrupt", Some(b"not-json")); + sys.publish_stored("corrupt", Some((b"not-json", uuid::Uuid::from_u128(1)))); assert!(sys.state("corrupt").is_none(), "corruption cannot keep an older source active"); assert!(state.is_cancelled(), "corruption cancels in-flight work"); } @@ -1395,7 +1522,11 @@ mod tests { let older = sys.reserve_generation("b", true); let newer = sys.reserve_generation("b", false); assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::NotDesired); - assert_eq!(sys.apply_with_generation("b", Some(&cfg), older).await, ApplyOutcome::Superseded); + assert_eq!( + sys.apply_with_generation("b", uuid::Uuid::from_u128(1), Some(&cfg), older) + .await, + ApplyOutcome::Superseded + ); assert!(sys.state("b").is_none(), "removal must supersede an in-flight first install"); assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Installed); @@ -1404,7 +1535,11 @@ mod tests { let newer = sys.reserve_generation("b", false); assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::Removed); assert!(installed.is_cancelled()); - assert_eq!(sys.apply_with_generation("b", Some(&cfg), older).await, ApplyOutcome::Superseded); + assert_eq!( + sys.apply_with_generation("b", uuid::Uuid::from_u128(1), Some(&cfg), older) + .await, + ApplyOutcome::Superseded + ); assert!(sys.state("b").is_none(), "the stale install is discarded"); } diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 213df046e..faa98ad72 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -288,6 +288,68 @@ async fn load_bucket_generation(fs: &FS, req: &S3Request, bucket: &str) -> load_bucket_generation_from_store(store.as_ref(), req, bucket).await } +/// A read may consult only the source admitted before its authorization. +/// Capture failures are deferred until a local miss actually needs that source. +#[derive(Clone, Debug)] +enum OdmReadGenerationGuard { + Unavailable, + Ready(BucketGenerationGuard), + Failed { code: S3ErrorCode, message: String }, +} + +impl OdmReadGenerationGuard { + fn from_result(result: S3Result) -> Self { + match result { + Ok(guard) => Self::Ready(guard), + Err(err) => Self::Failed { + code: err.code().clone(), + message: err.message().unwrap_or_else(|| err.code().as_str()).to_string(), + }, + } + } +} + +fn odm_read_source_configured(bucket: &str) -> bool { + let sys = crate::on_demand_migration::OnDemandMigrationSys::get(); + sys.is_module_enabled() && sys.state(bucket).is_some() +} + +async fn capture_odm_read_generation(fs: &FS, req: &S3Request, bucket: &str) -> OdmReadGenerationGuard { + if !odm_read_source_configured(bucket) { + return OdmReadGenerationGuard::Unavailable; + } + OdmReadGenerationGuard::from_result(load_bucket_generation(fs, req, bucket).await) +} + +/// Direct usecase callers have no access middleware; capture at their entry. +/// A server request without the access marker must never bind to a later source. +pub(crate) async fn prepare_odm_read_generation( + store: &crate::storage::storage_api::ECStore, + req: &mut S3Request, + bucket: &str, +) { + if req.extensions.get::().is_some() { + return; + } + let guard = if req.extensions.get::>().is_some() || !odm_read_source_configured(bucket) { + OdmReadGenerationGuard::Unavailable + } else { + OdmReadGenerationGuard::from_result(load_bucket_generation_from_store(store, req, bucket).await) + }; + req.extensions.insert(guard); +} + +pub(crate) fn odm_read_generation(req: &S3Request, bucket: &str) -> S3Result> { + match req.extensions.get::() { + Some(OdmReadGenerationGuard::Ready(guard)) if guard.bucket == bucket => Ok(Some(guard.incarnation_id)), + Some(OdmReadGenerationGuard::Ready(_)) => { + Err(s3_error!(InternalError, "source generation guard does not match request bucket")) + } + Some(OdmReadGenerationGuard::Failed { code, message }) => Err(S3Error::with_message(code.clone(), message.clone())), + Some(OdmReadGenerationGuard::Unavailable) | None => Ok(None), + } +} + async fn load_copy_source_bucket_generation(fs: &FS, bucket: &str) -> S3Result { let store = fs .server_ctx() @@ -2365,6 +2427,8 @@ impl S3Access for FS { /// /// This method returns `Ok(())` by default. async fn get_object(&self, req: &mut S3Request) -> S3Result<()> { + let bucket = req.input.bucket.clone(); + let source_generation = capture_odm_read_generation(self, req, &bucket).await; let req_info = ext_req_info_mut(&mut req.extensions)?; req_info.bucket = Some(req.input.bucket.clone()); req_info.object = Some(req.input.key.clone()); @@ -2372,7 +2436,9 @@ impl S3Access for FS { // GHSA-3ppv: a versioned read (?versionId=...) must authorize against // s3:GetObjectVersion, not s3:GetObject. - authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await + authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await?; + req.extensions.insert(source_generation); + Ok(()) } /// Checks whether the GetObjectAcl request has accesses to the resources. @@ -2484,6 +2550,8 @@ impl S3Access for FS { /// /// This method returns `Ok(())` by default. async fn head_object(&self, req: &mut S3Request) -> S3Result<()> { + let bucket = req.input.bucket.clone(); + let source_generation = capture_odm_read_generation(self, req, &bucket).await; let req_info = ext_req_info_mut(&mut req.extensions)?; req_info.bucket = Some(req.input.bucket.clone()); req_info.object = Some(req.input.key.clone()); @@ -2496,10 +2564,13 @@ impl S3Access for FS { if get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_CHECK).as_deref() == Some("true") { authorize_request(req, Action::S3Action(S3Action::ReplicateObjectAction)).await?; req_info_mut(req)?.replication_request_authorized = true; + req.extensions.insert(source_generation); return Ok(()); } - authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await + authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await?; + req.extensions.insert(source_generation); + Ok(()) } /// Checks whether the ListBucketAnalyticsConfigurations request has accesses to the resources. @@ -2588,10 +2659,14 @@ impl S3Access for FS { /// /// This method returns `Ok(())` by default. async fn list_objects_v2(&self, req: &mut S3Request) -> S3Result<()> { + let bucket = req.input.bucket.clone(); + let source_generation = capture_odm_read_generation(self, req, &bucket).await; let req_info = ext_req_info_mut(&mut req.extensions)?; req_info.bucket = Some(req.input.bucket.clone()); - authorize_request(req, Action::S3Action(S3Action::ListBucketAction)).await + authorize_request(req, Action::S3Action(S3Action::ListBucketAction)).await?; + req.extensions.insert(source_generation); + Ok(()) } /// Checks whether the ListParts request has accesses to the resources. @@ -3858,6 +3933,285 @@ mod tests { assert_eq!(req_info.object.as_deref(), Some("test-key")); } + #[test] + #[serial] + fn odm_read_capture_preserves_access_and_parameter_error_order() { + crate::app::gating_test_env::run_large_stack_test("odm-access-order", || async { + let context = crate::app::gating_test_env::shared_gating_ambient().await; + let server_ctx = ServerContextSlot::new(); + assert!(server_ctx.install(Arc::clone(&context))); + let fs = FS::with_server_ctx(server_ctx); + let sys = crate::on_demand_migration::OnDemandMigrationSys::get(); + let enabled_before = sys.is_module_enabled(); + let bucket = format!("odm-access-missing-{}", uuid::Uuid::new_v4()); + for enabled in [false, true] { + sys.set_module_enabled(enabled); + let mut get = build_request( + GetObjectInput { + bucket: bucket.clone(), + key: "key".into(), + part_number: Some(0), + ..Default::default() + }, + Method::GET, + ); + get.extensions.insert(ReqInfo { + cred: Some(rustfs_credentials::Credentials::default()), + is_owner: true, + ..Default::default() + }); + get.extensions.insert(fs.server_ctx().clone()); + let mut head = get.clone().map_input(|_| HeadObjectInput { + bucket: bucket.clone(), + key: "key".into(), + part_number: Some(1), + range: Some(s3s::dto::Range::Int { first: 0, last: Some(1) }), + ..Default::default() + }); + head.method = Method::HEAD; + let mut list = get.clone().map_input(|_| ListObjectsV2Input { + bucket: bucket.clone(), + max_keys: Some(-1), + ..Default::default() + }); + fs.get_object(&mut get) + .await + .expect("ordinary GET access must not require bucket identity"); + fs.head_object(&mut head) + .await + .expect("ordinary HEAD access must not require bucket identity"); + fs.list_objects_v2(&mut list) + .await + .expect("ordinary LIST access must not require bucket identity"); + assert!(matches!( + get.extensions.get::(), + Some(super::OdmReadGenerationGuard::Unavailable) + )); + assert!(matches!( + head.extensions.get::(), + Some(super::OdmReadGenerationGuard::Unavailable) + )); + assert!(matches!( + list.extensions.get::(), + Some(super::OdmReadGenerationGuard::Unavailable) + )); + let usecase = crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + assert_eq!( + usecase.execute_get_object(get).await.expect_err("bad GET part number").code(), + &S3ErrorCode::InvalidArgument + ); + assert_eq!( + usecase + .execute_head_object(head) + .await + .expect_err("range and part number conflict") + .code(), + &S3ErrorCode::InvalidArgument + ); + assert_eq!( + crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context))) + .execute_list_objects_v2(list) + .await + .expect_err("negative max keys") + .code(), + &S3ErrorCode::InvalidArgument + ); + } + sys.set_module_enabled(enabled_before); + }); + } + + #[test] + #[serial] + fn odm_read_capture_failure_is_deferred_until_source_miss() { + crate::app::gating_test_env::run_large_stack_test("odm-access-failure", || async { + let context = crate::app::gating_test_env::shared_gating_ambient().await; + let store = context.object_store(); + let server_ctx = ServerContextSlot::new(); + assert!(server_ctx.install(Arc::clone(&context))); + let fs = FS::with_server_ctx(server_ctx); + let sys = crate::on_demand_migration::OnDemandMigrationSys::get(); + let enabled_before = sys.is_module_enabled(); + sys.set_module_enabled(true); + let bucket = format!("odm-capture-failure-{}", uuid::Uuid::new_v4()); + let mut config: crate::on_demand_migration::OnDemandMigrationConfig = serde_json::from_str(r#"{"source":{"provider":"minio","endpoint":"https://source.example.com","bucket":"source","credentials":{"access_key":"test","secret_key":"test"}}}"#).expect("source config"); + config.policy.list_through = true; + sys.apply_for_incarnation(&bucket, uuid::Uuid::new_v4(), Some(&config)).await; + let mut get = build_request( + GetObjectInput { + bucket: bucket.clone(), + key: "local".into(), + ..Default::default() + }, + Method::GET, + ); + get.extensions.insert(ReqInfo { + cred: Some(rustfs_credentials::Credentials::default()), + is_owner: true, + ..Default::default() + }); + get.extensions.insert(fs.server_ctx().clone()); + let mut head = get.clone().map_input(|_| HeadObjectInput { + bucket: bucket.clone(), + key: "local".into(), + ..Default::default() + }); + head.method = Method::HEAD; + let mut list = get.clone().map_input(|_| ListObjectsV2Input { + bucket: bucket.clone(), + ..Default::default() + }); + fs.list_objects_v2(&mut list) + .await + .expect("capture failure must not preempt LIST authorization"); + fs.get_object(&mut get) + .await + .expect("capture failure must not preempt GET authorization"); + fs.head_object(&mut head) + .await + .expect("capture failure must not preempt HEAD authorization"); + assert!(matches!( + get.extensions.get::(), + Some(super::OdmReadGenerationGuard::Failed { .. }) + )); + assert!(matches!( + head.extensions.get::(), + Some(super::OdmReadGenerationGuard::Failed { .. }) + )); + store + .make_bucket( + &bucket, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("create bucket after capture"); + store + .put_object( + &bucket, + "local", + &mut crate::storage::PutObjReader::from_vec(b"local".to_vec()), + &crate::storage::ObjectOptions::default(), + ) + .await + .expect("create local hit"); + // Publishing a usable source later must not repair a failed capture. + let incarnation = store.bucket_incarnation_id(&bucket).await.expect("created identity"); + sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await; + let usecase = crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + usecase + .execute_get_object(get.clone()) + .await + .expect("a local GET hit ignores source capture failure"); + usecase + .execute_head_object(head.clone()) + .await + .expect("a local HEAD hit ignores source capture failure"); + get.input.key = "missing".into(); + head.input.key = "missing".into(); + assert_eq!( + usecase + .execute_get_object(get.clone()) + .await + .expect_err("failed capture cannot rebind on GET miss") + .code(), + &S3ErrorCode::NoSuchBucket + ); + assert_eq!( + usecase + .execute_head_object(head.clone()) + .await + .expect_err("failed capture cannot rebind on HEAD miss") + .code(), + &S3ErrorCode::NoSuchBucket + ); + assert_eq!( + crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context))) + .execute_list_objects_v2(list.clone()) + .await + .expect_err("failed capture cannot rebind on LIST") + .code(), + &S3ErrorCode::NoSuchBucket + ); + config.filter.prefix = Some("remote/".into()); + sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await; + assert_eq!( + usecase + .execute_get_object(get.clone()) + .await + .expect_err("filtered GET remains local") + .code(), + &S3ErrorCode::NoSuchKey + ); + assert_eq!( + usecase + .execute_head_object(head.clone()) + .await + .expect_err("filtered HEAD remains local") + .code(), + &S3ErrorCode::NoSuchKey + ); + list.input.prefix = Some("local/".into()); + let listing = crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context))) + .execute_list_objects_v2(list.clone()) + .await + .expect("disjoint prefix needs no source identity"); + assert_eq!(listing.output.key_count, Some(0)); + list.input.prefix = Some("remote/".into()); + list.input.max_keys = Some(0); + let listing = crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context))) + .execute_list_objects_v2(list) + .await + .expect("an empty page needs no source identity"); + assert_eq!(listing.output.key_count, Some(0)); + + config.filter.prefix = None; + config.policy.head = crate::on_demand_migration::HeadPolicy::LocalOnly; + sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await; + assert_eq!( + usecase + .execute_head_object(head.clone()) + .await + .expect_err("local-only HEAD needs no source identity") + .code(), + &S3ErrorCode::NoSuchKey + ); + config.policy.head = crate::on_demand_migration::HeadPolicy::Proxy; + sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await; + store + .delete_object( + &bucket, + "missing", + crate::storage::ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("create local delete marker"); + assert_eq!( + usecase + .execute_get_object(get) + .await + .expect_err("GET respects the delete marker before capture failure") + .code(), + &S3ErrorCode::NoSuchKey + ); + assert_eq!( + usecase + .execute_head_object(head) + .await + .expect_err("HEAD respects the delete marker before capture failure") + .code(), + &S3ErrorCode::NoSuchKey + ); + sys.remove(&bucket); + sys.set_module_enabled(enabled_before); + }); + } + #[tokio::test] #[serial] async fn put_object_access_captures_authorized_bucket_incarnation() { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index fe9a55641..e35cf812d 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -119,8 +119,9 @@ pub(crate) mod access_consumer { pub(crate) use super::super::access::{ PostObjectRequestMarker, ReqInfo, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard, authorize_internal_object_request, authorize_request, bucket_config_mutation_incarnation, has_bypass_governance_header, - load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, prepare_list_buckets_iam_authorization, - recursive_force_delete_is_authorized, replication_request_authorized, req_info_mut, req_info_ref, + load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation, + prepare_list_buckets_iam_authorization, prepare_odm_read_generation, recursive_force_delete_is_authorized, + replication_request_authorized, req_info_mut, req_info_ref, }; } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 60de2c850..9abdbd5fd 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -430,10 +430,12 @@ pub(crate) mod on_demand_migration { #[cfg(test)] pub(crate) mod test_support { + pub(crate) use crate::storage::storage_api::contract::bucket::{BUCKET_LIFECYCLE_LOCK_OBJECT, BucketOperations}; pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata::{ BUCKET_ON_DEMAND_MIGRATION_CONFIG, BucketMetadata, }; - pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::BucketMetadataSys; pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::test_support::isolated_store_over_temp_disks; + pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::{BucketMetadataSys, init_bucket_metadata_sys}; + pub(crate) use crate::storage::storage_api::ecstore_set_disk::{PutObjectCommitBarrier, PutObjectCommitPause}; } }