gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 13:46:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(storage): preserve retryable publication guard errors (#7561)

Browse Source 
* 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 <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
cxymds
2026-09-09 16:40:43 +08:00
committed by GitHub
parent ed2b2cdd19
commit 51e6733a9c
2 changed files with 281 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/namespace_lock_quorum_test.rs
+121 -1
View File
@@ -25,6 +25,123 @@ const KEY: &str = "thumb/79/concurrent-overwrite.jpg";
type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
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<u8>, 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
}
rustfs/src/storage/access.rs
+160 -2
View File
@@ -1538,6 +1538,14 @@ fn table_catalog_store_for_data_plane<T>(
.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<T>(
req: &mut S3Request<T>,
table_bucket: &str,
@@ -1556,7 +1564,7 @@ async fn retain_table_data_plane_publication_guard<T>(
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::<TableDataPlanePublicationGuards>().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::<TableDataPlanePublicationGuards>()
.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");