From a464b45367748d1869fba0119fbb032ea7fb9d4d Mon Sep 17 00:00:00 2001 From: Hauser Date: Fri, 9 Oct 2026 14:08:12 +0800 Subject: [PATCH] fix(rustfs): keep recovery handler within S3 gateway (#8429) * fix(rustfs): route recovery errors through S3 gateway Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(ecstore): satisfy strict Clippy pagination checks Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(heal): avoid cloning Copy purge identity Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test: restore one-second drive request budgets * fix(ecstore): resume pending lock bucket creation Allow retries to reuse an uncommitted Object Lock bucket intent when the same incarnation remains unretired and all physical volumes are absent. Isolate the recovery entrypoint test from the shared production leader lock. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test: provision stack for deep catalog test futures Apply the existing 4 MiB test-thread stack setup to table-catalog handler/store tests and the relocated-pool GET regression in both local and CI nextest profiles. These cases overflowed libtest's 2 MiB default stack on Linux. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 Co-authored-by: overtrue --- .config/nextest.toml | 10 ++ .../bucket/lifecycle/bucket_lifecycle_ops.rs | 35 ++++- crates/ecstore/src/disk/disk_store.rs | 2 +- crates/ecstore/src/store/bucket.rs | 25 +++- crates/ecstore/src/store/init.rs | 2 +- crates/ecstore/src/store/list_objects.rs | 127 +++++++++++------- .../heal/src/heal/mrf_queue/partial_write.rs | 2 +- rustfs/src/admin/handlers/bucket_recovery.rs | 26 ++-- 8 files changed, 152 insertions(+), 77 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 80b2d8398..3881b9036 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -92,6 +92,12 @@ setup = 'ecstore-large-stack' filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))' setup = 'ecstore-base-stack' +# These handler/store fixtures and the relocated-pool GET compose deep storage +# futures in libtest threads; on Linux they overflow the default 2 MiB stack. +[[profile.default.scripts]] +filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))' +setup = 'ecstore-base-stack' + [[profile.default.scripts]] filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))' setup = 'lifecycle-large-stack' @@ -337,6 +343,10 @@ setup = 'ecstore-large-stack' filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))' setup = 'ecstore-base-stack' +[[profile.ci.scripts]] +filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))' +setup = 'ecstore-base-stack' + [[profile.ci.scripts]] filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))' setup = 'lifecycle-large-stack' diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 1d0d9ae48..06919238f 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -2939,6 +2939,20 @@ fn resolve_tier_free_version_recovery_enabled(value: Result, started: &OnceLock<()>) -> Option> { + spawn_tier_free_version_recovery_once_with_options( + api, + started, + TIER_FREE_VERSION_RECOVERY_LEADER_LOCK.to_string(), + tier_free_version_recovery_leader_retry_delay, + ) +} + +fn spawn_tier_free_version_recovery_once_with_options( + api: Arc, + started: &OnceLock<()>, + leader_lock_name: String, + retry_delay: fn() -> StdDuration, +) -> Option> { if !tier_free_version_recovery_enabled() { warn!( event = EVENT_LIFECYCLE_WORKER_STATE, @@ -2956,15 +2970,15 @@ fn spawn_tier_free_version_recovery_once(api: Arc, started: &OnceLock<( Some(tokio::spawn(async move { let cancel_token = api.ctx.background_cancel_token().unwrap_or_default(); let acquire_api = Arc::clone(&api); + let acquire_lock_name = Arc::::from(leader_lock_name); run_tier_free_version_recovery_coordinator( cancel_token, - tier_free_version_recovery_leader_retry_delay, + retry_delay, move || { let api = Arc::clone(&acquire_api); + let lock_name = Arc::clone(&acquire_lock_name); async move { - let lock = api - .new_ns_lock(RUSTFS_META_BUCKET, TIER_FREE_VERSION_RECOVERY_LEADER_LOCK) - .await?; + let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &lock_name).await?; match lock.get_write_lock_quiet(TIER_FREE_VERSION_RECOVERY_LOCK_TIMEOUT).await { Ok(guard) => Ok(Some(guard)), Err(LockError::Timeout { .. }) => Ok(None), @@ -14193,6 +14207,10 @@ mod tests { .await; } + fn fast_tier_free_version_recovery_leader_retry_delay() -> StdDuration { + StdDuration::from_millis(10) + } + #[tokio::test] #[serial] async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() { @@ -14208,8 +14226,13 @@ mod tests { seed_recoverable_free_version(&disk_paths, &bucket, object, None, None).await; let started = OnceLock::new(); - let recovery = super::spawn_tier_free_version_recovery_once(Arc::clone(&ecstore), &started) - .expect("production recovery entrypoint should start once"); + let recovery = super::spawn_tier_free_version_recovery_once_with_options( + Arc::clone(&ecstore), + &started, + format!("tier-free-version-recovery/test/{}", Uuid::new_v4()), + fast_tier_free_version_recovery_leader_retry_delay, + ) + .expect("production recovery entrypoint should start once"); let task = tokio::time::timeout(StdDuration::from_secs(30), async { recovery_rx.lock().await.recv().await }) .await .expect("production recovery entrypoint should enqueue the seeded item") diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 625bf65a9..98de238f5 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -2911,7 +2911,7 @@ mod tests { let timed = wrapper.track_disk_health_with_op( "write_all", - || std::future::pending::>(), + std::future::pending::>, SLOW_DISK_OPERATION_THRESHOLD + Duration::from_secs(1), ); tokio::pin!(timed); diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 9054fa12f..88e39627b 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -901,6 +901,7 @@ impl ECStore { } }; let confirmed_missing = existing_bucket_info.is_none(); + let mut pending_creation_metadata = None; if confirmed_missing && !opts.no_lock && !is_meta_bucketname(bucket) @@ -921,13 +922,25 @@ impl ECStore { ), other => other, })?; - return Err(StorageError::InvalidArgument( - "CreateBucket".to_string(), - "bucket".to_string(), - format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"), - )); + let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?; + let metadata_store = metadata_sys::object_store_in(&self.ctx).await?; + let pending_intent = persisted + && metadata.bucket_incarnation_id == old_incarnation + && metadata.needs_bucket_creation_commit() + && !crate::bucket::retirement::is_retired(metadata_store, bucket, old_incarnation).await?; + if pending_intent { + pending_creation_metadata = Some(metadata); + } else { + return Err(StorageError::InvalidArgument( + "CreateBucket".to_string(), + "bucket".to_string(), + format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"), + )); + } } - let existing_metadata = if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) { + let existing_metadata = if let Some(metadata) = pending_creation_metadata { + Some(metadata) + } else if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) { let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?; if !persisted { metadata = BucketMetadata::new(bucket); diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 49e02380e..0a2a855bb 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -18318,7 +18318,7 @@ mod tests { let bucket = format!("lock-create-intent-retry-{}", uuid::Uuid::new_v4()); let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket); intent.lock_enabled = true; - crate::bucket::metadata_sys::set_new_bucket_metadata_in(&ctx, intent) + crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, intent) .await .expect("persist pre-visibility lock intent"); assert!( diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 91f2aca5d..f870ecc39 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -2361,24 +2361,43 @@ fn list_objects_paginate( (objects, prefixes, is_truncated, next_marker, next_version_idmarker) } -fn list_objects_paginate_versions<'a>( - mut get_objects: Vec, - delimiter: &Option, +struct VersionedListPagination<'a, I> { + delimiter: &'a Option, max_keys: i32, disk_has_more: bool, - cache_id: Option<&str>, - marker: Option<&str>, - last_scanned_key: Option<&str>, - raw_keys: impl Iterator, -) -> (Vec, Vec, bool, Option, Option) { - filter_versions_common_prefixes_after_marker(&mut get_objects, delimiter.as_deref(), marker); - let prefix_at_page_boundary = - versioned_prefix_page_boundary_key(&get_objects, raw_keys, delimiter.as_deref(), max_keys, disk_has_more); + cache_id: Option<&'a str>, + marker: Option<&'a str>, + last_scanned_key: Option<&'a str>, + raw_keys: I, +} - let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) = - list_objects_paginate(get_objects, delimiter, max_keys, disk_has_more, cache_id, true, last_scanned_key); +fn list_objects_paginate_versions<'a, I>( + mut get_objects: Vec, + pagination: VersionedListPagination<'a, I>, +) -> (Vec, Vec, bool, Option, Option) +where + I: Iterator, +{ + filter_versions_common_prefixes_after_marker(&mut get_objects, pagination.delimiter.as_deref(), pagination.marker); + let prefix_at_page_boundary = versioned_prefix_page_boundary_key( + &get_objects, + pagination.raw_keys, + pagination.delimiter.as_deref(), + pagination.max_keys, + pagination.disk_has_more, + ); + + let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) = list_objects_paginate( + get_objects, + pagination.delimiter, + pagination.max_keys, + pagination.disk_has_more, + pagination.cache_id, + true, + pagination.last_scanned_key, + ); if is_truncated && let Some(raw_key) = prefix_at_page_boundary { - next_marker = Some(append_list_cache_id_to_marker(raw_key, cache_id)); + next_marker = Some(append_list_cache_id_to_marker(raw_key, pagination.cache_id)); next_version_idmarker = None; } @@ -4326,13 +4345,15 @@ impl ECStore { let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions( get_objects, - &delimiter, - max_keys, - disk_has_more, - next_cache_id.as_deref(), - opts.marker.as_deref(), - last_scanned_key.as_deref(), - entries.entries().iter().map(|entry| entry.name.as_str()), + VersionedListPagination { + delimiter: &delimiter, + max_keys, + disk_has_more, + cache_id: next_cache_id.as_deref(), + marker: opts.marker.as_deref(), + last_scanned_key: last_scanned_key.as_deref(), + raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()), + }, ); Ok(ListObjectVersionsInfo { @@ -5754,13 +5775,15 @@ impl Sets { let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions( get_objects, - &delimiter, - max_keys, - disk_has_more, - next_cache_id.as_deref(), - opts.marker.as_deref(), - last_scanned_key.as_deref(), - entries.entries().iter().map(|entry| entry.name.as_str()), + VersionedListPagination { + delimiter: &delimiter, + max_keys, + disk_has_more, + cache_id: next_cache_id.as_deref(), + marker: opts.marker.as_deref(), + last_scanned_key: last_scanned_key.as_deref(), + raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()), + }, ); Ok(ListObjectVersionsInfo { @@ -6568,13 +6591,15 @@ impl SetDisks { let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions( get_objects, - &delimiter, - max_keys, - disk_has_more, - next_cache_id.as_deref(), - opts.marker.as_deref(), - last_scanned_key.as_deref(), - entries.entries().iter().map(|entry| entry.name.as_str()), + VersionedListPagination { + delimiter: &delimiter, + max_keys, + disk_has_more, + cache_id: next_cache_id.as_deref(), + marker: opts.marker.as_deref(), + last_scanned_key: last_scanned_key.as_deref(), + raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()), + }, ); Ok(ListObjectVersionsInfo { @@ -8848,13 +8873,15 @@ mod test { let get_objects = fold_delimiter_page(&window, "", "-"); let (objects, prefixes, is_truncated, next_marker, _v) = super::list_objects_paginate_versions( get_objects, - &delimiter, - max_keys, - disk_has_more, - None, - marker.as_deref(), - last_scanned.as_deref(), - window.iter().map(String::as_str), + super::VersionedListPagination { + delimiter: &delimiter, + max_keys, + disk_has_more, + cache_id: None, + marker: marker.as_deref(), + last_scanned_key: last_scanned.as_deref(), + raw_keys: window.iter().map(String::as_str), + }, ); assert!( @@ -8927,13 +8954,15 @@ mod test { let get_objects = fold_delimiter_page(&window, "", "-"); let (objects, prefixes, is_truncated, next_marker, _) = super::list_objects_paginate_versions( get_objects, - &delimiter, - max_keys, - disk_has_more, - None, - marker.as_deref(), - last_scanned.as_deref(), - window.iter().map(String::as_str), + super::VersionedListPagination { + delimiter: &delimiter, + max_keys, + disk_has_more, + cache_id: None, + marker: marker.as_deref(), + last_scanned_key: last_scanned.as_deref(), + raw_keys: window.iter().map(String::as_str), + }, ); assert_eq!(objects.len() + prefixes.len(), 1, "a page must respect max-keys"); diff --git a/crates/heal/src/heal/mrf_queue/partial_write.rs b/crates/heal/src/heal/mrf_queue/partial_write.rs index 8749efe82..905664440 100644 --- a/crates/heal/src/heal/mrf_queue/partial_write.rs +++ b/crates/heal/src/heal/mrf_queue/partial_write.rs @@ -392,7 +392,7 @@ impl PartialWrites { object: anchor.object.clone(), version_id: anchor.version_id, scope: anchor.scope, - delete_marker_purge: anchor.delete_marker_purge.clone(), + delete_marker_purge: anchor.delete_marker_purge, lease: Some(anchor.lease), bucket_incarnation_id: anchor.bucket_incarnation_id, disposition: MrfVerifiedRepairDisposition::AuthoritativelyAbsent, diff --git a/rustfs/src/admin/handlers/bucket_recovery.rs b/rustfs/src/admin/handlers/bucket_recovery.rs index 4be9c6c96..2d1b5412a 100644 --- a/rustfs/src/admin/handlers/bucket_recovery.rs +++ b/rustfs/src/admin/handlers/bucket_recovery.rs @@ -17,12 +17,12 @@ use crate::admin::auth::authorize_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::current_object_store_handle; +use crate::admin::storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result}; use crate::server::ADMIN_PREFIX; use crate::site_replication::{site_replication_enabled, with_site_replication_bucket_mutation_lock}; use hyper::{Method, StatusCode}; use matchit::Params; use rustfs_policy::policy::action::{Action, AdminAction}; -use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use tracing::info; use uuid::Uuid; @@ -63,23 +63,23 @@ impl Operation for RecoverOrphanedBucketHandler { .credentials .as_ref() .map(|credentials| credentials.secret_key.clone()) - .ok_or_else(|| s3_error!(InvalidRequest, "authentication required"))?; + .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "authentication required"))?; let bucket = params .get("bucket") .filter(|bucket| !bucket.is_empty()) - .ok_or_else(|| s3_error!(InvalidRequest, "bucket name is required"))?; + .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "bucket name is required"))?; let body = req .input .store_all_limited(rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE) .await - .map_err(|_| s3_error!(InvalidRequest, "failed to read recovery request body"))?; + .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "failed to read recovery request body"))?; let request: RecoverOrphanedBucketRequest = serde_json::from_slice(&body) - .map_err(|_| s3_error!(InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?; + .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?; if request.expected_incarnation_id.is_nil() { - return Err(s3_error!(InvalidRequest, "expectedIncarnationId must be non-nil")); + return Err(s3::error(S3ErrorCode::InvalidRequest, "expectedIncarnationId must be non-nil")); } let Some(store) = current_object_store_handle() else { - return Err(s3_error!(InternalError, "object store is not initialized")); + return Err(s3::error(S3ErrorCode::InternalError, "object store is not initialized")); }; let operation_bucket = bucket.to_owned(); @@ -91,21 +91,21 @@ impl Operation for RecoverOrphanedBucketHandler { let lock_bucket = operation_bucket.clone(); with_site_replication_bucket_mutation_lock(lock_store, &lock_bucket, move || async move { if site_replication_enabled().await? { - return Err(s3_error!( - OperationAborted, - "orphaned bucket recovery is local; reconcile every site before recreating this bucket" + return Err(s3::error( + S3ErrorCode::OperationAborted, + "orphaned bucket recovery is local; reconcile every site before recreating this bucket", )); } operation_store .recover_orphaned_bucket(&operation_bucket, expected_incarnation) .await - .map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error))) + .map_err(|error| S3Error::from(crate::error::ApiError::from(error))) }) .await??; - Ok::<(), s3s::S3Error>(()) + Ok::<(), S3Error>(()) }) .await - .map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))?; + .map_err(|error| S3Error::from(crate::error::ApiError::from(error)))?; completed?; info!(