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!(