fix(storage): preserve retryable publication guard errors

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

@@ -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");