From 51e6733a9cf53692fd6184d838ffc203b3fc1c12 Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 9 Sep 2026 16:40:43 +0800 Subject: [PATCH] fix(storage): preserve retryable publication guard errors (#7561) * fix(storage): preserve retryable publication guard errors * fix(scanner): satisfy scoped cache clippy lint Use bool::then_some for the scoped cold bucket reuse proof construction after merging the latest release branch. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: houseme Co-authored-by: zhi22915 --- .../src/namespace_lock_quorum_test.rs | 122 ++++++++++++- rustfs/src/storage/access.rs | 162 +++++++++++++++++- 2 files changed, 281 insertions(+), 3 deletions(-) diff --git a/crates/e2e_test/src/namespace_lock_quorum_test.rs b/crates/e2e_test/src/namespace_lock_quorum_test.rs index 0af609ecc..47b180cd7 100644 --- a/crates/e2e_test/src/namespace_lock_quorum_test.rs +++ b/crates/e2e_test/src/namespace_lock_quorum_test.rs @@ -25,6 +25,123 @@ const KEY: &str = "thumb/79/concurrent-overwrite.jpg"; type TestResult = Result<(), Box>; +async fn assert_degraded_cluster_publication_guard_errors_are_retryable() -> TestResult { + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:2"); + cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); + cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false"); + cluster.set_env("RUST_LOG", "warn"); + cluster.start().await?; + cluster.create_test_bucket(BUCKET).await?; + let clients: Vec<_> = cluster + .create_all_clients()? + .into_iter() + .map(|client| { + Client::from_conf( + client + .config() + .to_builder() + .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1)) + .build(), + ) + }) + .collect(); + + for alive in (1..=4).rev() { + if alive < 4 { + cluster.stop_node(alive)?; + } + for (node, client) in clients.iter().take(alive).enumerate() { + let key = format!("publication-put-{alive}-{node}"); + let put = client + .put_object() + .bucket(BUCKET) + .key(&key) + .body(Bytes::from_static(b"publication guard regression").into()) + .send() + .await; + if alive >= 3 { + put?; + } else { + let err = put.expect_err("PUT must reject writes without a write quorum"); + assert_eq!( + err.raw_response().map(|response| response.status().as_u16()), + Some(503), + "PUT with {alive} nodes alive, requested through node {node}: {err:?}" + ); + assert_eq!( + err.as_service_error().and_then(|error| error.meta().code()), + Some("ServiceUnavailable"), + "PUT with {alive} nodes alive, requested through node {node}: {err:?}" + ); + } + + let multipart_key = format!("publication-multipart-{alive}-{node}"); + let multipart = client + .create_multipart_upload() + .bucket(BUCKET) + .key(&multipart_key) + .send() + .await; + if alive >= 3 { + let upload = multipart?; + let upload_id = upload + .upload_id() + .expect("successful multipart initialization must return an upload ID"); + client + .abort_multipart_upload() + .bucket(BUCKET) + .key(&multipart_key) + .upload_id(upload_id) + .send() + .await?; + } else { + let err = multipart.expect_err("multipart initialization must reject writes without a write quorum"); + assert_eq!( + err.raw_response().map(|response| response.status().as_u16()), + Some(503), + "CreateMultipartUpload with {alive} nodes alive, requested through node {node}: {err:?}" + ); + assert_eq!( + err.as_service_error().and_then(|error| error.meta().code()), + Some("ServiceUnavailable"), + "CreateMultipartUpload with {alive} nodes alive, requested through node {node}: {err:?}" + ); + } + } + } + + cluster.stop(); + cluster.start().await?; + for client in &clients { + for alive in [1, 3, 4] { + for node in 0..alive { + let key = format!("publication-put-{alive}-{node}"); + let get = client.get_object().bucket(BUCKET).key(key).send().await; + if alive >= 3 { + assert_eq!( + get?.body.collect().await?.into_bytes().as_ref(), + b"publication guard regression", + "acknowledged writes must survive restart" + ); + } else { + let err = get.expect_err("a rejected publication guard must not publish an object"); + assert_eq!(err.as_service_error().and_then(|error| error.meta().code()), Some("NoSuchKey")); + } + } + } + let uploads = client.list_multipart_uploads().bucket(BUCKET).send().await?; + assert!( + uploads + .uploads() + .iter() + .all(|upload| upload.key() != Some("publication-multipart-1-0")), + "a rejected publication guard must not publish a multipart upload" + ); + } + Ok(()) +} + async fn put_object(client: Client, payload: Vec, writer_id: usize) -> Result<(), String> { client .put_object() @@ -125,6 +242,8 @@ async fn test_concurrent_cluster_overwrites_do_not_fail_namespace_lock_quorum() /// Before the fix, `map_namespace_lock_error` wrapped lock timeout/conflict errors as /// `StorageError::other(...)` → `StorageError::Io(...)`, which fell through to /// `S3ErrorCode::InternalError` (500) in the error mapping. +/// Also checks PUT and multipart initialization when node failures prevent +/// acquiring a table publication guard. #[tokio::test] async fn test_concurrent_put_same_key_never_returns_500() -> TestResult { crate::common::init_logging(); @@ -227,5 +346,6 @@ async fn test_concurrent_put_same_key_never_returns_500() -> TestResult { ); clients[0].delete_object().bucket(BUCKET).key(KEY).send().await?; - Ok(()) + cluster.stop(); + assert_degraded_cluster_publication_guard_errors_are_retryable().await } diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index a2a95b017..405be6281 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -1538,6 +1538,14 @@ fn table_catalog_store_for_data_plane( .map_err(|err| s3_error!(InternalError, "failed to configure table catalog backing: {}", err)) } +fn table_publication_guard_error(err: crate::table_catalog::TableCatalogStoreError) -> S3Error { + let code = match &err { + crate::table_catalog::TableCatalogStoreError::Unavailable(_) => S3ErrorCode::ServiceUnavailable, + _ => S3ErrorCode::InternalError, + }; + S3Error::with_message(code, format!("failed to acquire table publication guard: {err}")) +} + async fn retain_table_data_plane_publication_guard( req: &mut S3Request, table_bucket: &str, @@ -1556,7 +1564,7 @@ async fn retain_table_data_plane_publication_guard( let backend = table_catalog_backend_for_data_plane(req)?; let guard = crate::table_catalog::TableCatalogObjectBackend::acquire_read_lock(&backend, table_bucket, lock_object) .await - .map_err(|err| s3_error!(InternalError, "failed to acquire table publication guard: {}", err))?; + .map_err(table_publication_guard_error)?; let mut state = retained.state.lock(); state.keys.insert(key); state.guards.push(Box::new(guard)); @@ -3106,7 +3114,8 @@ mod tests { merge_request_object_tag_conditions, owner_can_bypass_policy_deny, post_object_authorize_action, put_bucket_policy_authorize_action, request_context_from_req, request_object_store, retention_write_requested, secondary_tag_hint_action, table_data_plane_admin_action, table_data_plane_content_mutation, - table_data_plane_resource_for_request, validate_post_object_success_controls, versioned_read_action, + table_data_plane_resource_for_request, table_publication_guard_error, validate_post_object_success_controls, + versioned_read_action, }; use crate::error::ApiError; use crate::storage::storage_api::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions}; @@ -3268,6 +3277,155 @@ mod tests { } } + #[test] + fn table_publication_guard_unavailable_errors_are_retryable() { + use crate::table_catalog::TableCatalogStoreError; + + for message in [ + "failed to acquire catalog migration lock: Quorum not reached: required 2, achieved 1", + "failed to acquire catalog migration lock: lock acquisition timed out after 5s", + "peer unavailable", + "", + ] { + let error = TableCatalogStoreError::Unavailable(message.to_string()); + let expected_message = format!("failed to acquire table publication guard: {error}"); + let err = table_publication_guard_error(error); + assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable, "{message}"); + assert_eq!( + err.status_code().or_else(|| err.code().status_code()), + Some(http::StatusCode::SERVICE_UNAVAILABLE) + ); + assert_eq!(err.message(), Some(expected_message.as_str())); + } + } + + #[test] + fn table_publication_guard_non_retryable_errors_stay_internal() { + use crate::table_catalog::TableCatalogStoreError; + + let message = "temporarily unavailable: timeout: quorum not reached".to_string(); + for error in [ + TableCatalogStoreError::Internal(message.clone()), + TableCatalogStoreError::Invalid(message.clone()), + TableCatalogStoreError::Unsupported(message.clone()), + TableCatalogStoreError::NotFound(message.clone()), + TableCatalogStoreError::NamespaceNotFound(message.clone()), + TableCatalogStoreError::TableNotFound(message.clone()), + TableCatalogStoreError::AlreadyExists(message.clone()), + TableCatalogStoreError::Conflict(message), + ] { + let expected_message = format!("failed to acquire table publication guard: {error}"); + let err = table_publication_guard_error(error); + assert_eq!(err.code(), &S3ErrorCode::InternalError); + assert_eq!( + err.status_code().or_else(|| err.code().status_code()), + Some(http::StatusCode::INTERNAL_SERVER_ERROR) + ); + assert_eq!(err.message(), Some(expected_message.as_str())); + } + } + + #[tokio::test] + #[serial] + async fn table_publication_guard_timeout_blocks_put_and_multipart_authorization() { + use crate::storage::storage_api::contract::namespace::NamespaceLocking as _; + use std::time::Duration; + + let store = crate::app::gating_test_env::shared_gating_ecstore().await; + let server_ctx = ServerContextSlot::new(); + assert!(server_ctx.install(Arc::new(AppContext::new(Arc::clone(&store), Arc::new(UnreadyIam), Arc::new(TestKms))))); + let fs = FS::with_server_ctx(server_ctx); + let bucket = format!("publication-timeout-{}", uuid::Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("test bucket should be created"); + let policy_json = format!( + r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","Principal":{{"AWS":"*"}},"Action":["s3:PutObject"],"Resource":["arn:aws:s3:::{bucket}/*"]}}]}}"# + ); + let mut metadata = (*crate::storage::get_bucket_metadata(&bucket) + .await + .expect("test bucket metadata should be cached")) + .clone(); + metadata.policy_config = Some(serde_json::from_str(&policy_json).expect("test policy should parse")); + metadata.policy_config_json = policy_json.into_bytes(); + crate::storage::storage_api::set_bucket_metadata(bucket.clone(), metadata) + .await + .expect("test policy should be published"); + let lock_object = crate::table_catalog::default_table_bucket_publication_lock_path(); + let lock = store + .new_ns_lock(&bucket, &lock_object) + .await + .expect("publication lock should be created"); + let mut writer = lock + .get_write_lock(Duration::from_secs(1)) + .await + .expect("publication writer should hold the fence"); + let mut put_req = build_request( + PutObjectInput::builder() + .bucket(bucket.clone()) + .key("object".to_string()) + .build() + .expect("PUT input should build"), + Method::PUT, + ); + let mut multipart_req = build_request( + CreateMultipartUploadInput::builder() + .bucket(bucket.clone()) + .key("multipart-object".to_string()) + .build() + .expect("multipart input should build"), + Method::POST, + ); + ensure_req_info(&mut put_req); + ensure_req_info(&mut multipart_req); + put_req.extensions.insert(fs.server_ctx().clone()); + multipart_req.extensions.insert(fs.server_ctx().clone()); + + for (operation, err) in [ + ( + "PutObject", + fs.put_object(&mut put_req) + .await + .expect_err("PUT must wait for the publication writer"), + ), + ( + "CreateMultipartUpload", + fs.create_multipart_upload(&mut multipart_req) + .await + .expect_err("multipart initialization must wait for the publication writer"), + ), + ] { + assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable, "{operation}: {err}"); + } + for extensions in [&put_req.extensions, &multipart_req.extensions] { + assert!(extensions.get::().is_none()); + } + + assert!(writer.release()); + for _ in 0..2 { + fs.put_object(&mut put_req) + .await + .expect("PUT may retry after publication finishes"); + fs.create_multipart_upload(&mut multipart_req) + .await + .expect("multipart initialization may retry after publication finishes"); + for extensions in [&put_req.extensions, &multipart_req.extensions] { + let retained = extensions + .get::() + .expect("successful admission retains the guard"); + let state = retained.state.lock(); + assert!(state.keys.contains(&(bucket.clone(), lock_object.clone()))); + assert_eq!(state.guards.len(), 1, "repeated admission must reuse its retained guard"); + } + } + drop(put_req); + drop(multipart_req); + lock.get_write_lock(Duration::from_secs(1)) + .await + .expect("dropping the request must release its publication guard"); + } + #[test] fn table_data_plane_mutations_fence_before_table_bucket_marker_lookup() { let source = include_str!("access.rs");