fix(table-catalog): return 503 when commit authority is unavailable

rustfs/src/admin/handlers/table_catalog/mod.rs

@@ -5098,6 +5098,11 @@ fn catalog_store_error(err: crate::table_catalog::TableCatalogStoreError) -> S3E
         crate::table_catalog::TableCatalogStoreError::Unsupported(message) => {
             iceberg_rest_error(ICEBERG_ERROR_UNSUPPORTED_OPERATION, StatusCode::NOT_ACCEPTABLE, message)
         }
+        crate::table_catalog::TableCatalogStoreError::Unavailable(_) => iceberg_rest_error(
+            ICEBERG_ERROR_REST,
+            StatusCode::SERVICE_UNAVAILABLE,
+            "table catalog is temporarily unavailable",
+        ),
         crate::table_catalog::TableCatalogStoreError::Internal(message) => {
             iceberg_rest_error(ICEBERG_ERROR_REST, StatusCode::INTERNAL_SERVER_ERROR, message)
         }

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -289,6 +289,26 @@ fn catalog_conflicts_use_operation_specific_iceberg_errors() {
     }
 }
 
+#[test]
+fn catalog_unavailable_errors_use_iceberg_503() {
+    let unavailable = catalog_store_error(crate::table_catalog::TableCatalogStoreError::Unavailable(
+        "failed to acquire catalog table lock: quorum required 3, achieved 1".to_string(),
+    ));
+    assert_eq!(unavailable.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_REST.into()));
+    assert_eq!(unavailable.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
+    assert_eq!(unavailable.message(), Some("table catalog is temporarily unavailable"));
+    assert!(
+        unavailable
+            .headers()
+            .is_none_or(|headers| !headers.contains_key(rustfs_utils::http::RETRY_AFTER))
+    );
+
+    let internal = catalog_store_error(crate::table_catalog::TableCatalogStoreError::Internal(
+        "commit state is unknown".to_string(),
+    ));
+    assert_eq!(internal.status_code(), Some(StatusCode::INTERNAL_SERVER_ERROR));
+}
+
 #[test]
 fn table_catalog_admin_operation_result_labels_are_stable() {
     let success: Result<(), ()> = Ok(());
@@ -4599,6 +4619,66 @@ async fn commit_publication_holds_referenced_object_locks_until_pointer_publish(
     }
 }
 
+#[tokio::test]
+async fn commit_publication_authority_unavailable_returns_503_without_catalog_advance() {
+    let store = TestTableCatalogStore::default();
+    let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    create_standard_events_table(&store, &metadata_backend, &namespace).await;
+    let before = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should exist");
+    let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse");
+    metadata_backend
+        .fail_next_write_lock(
+            crate::table_catalog::default_table_publication_lock_path(&namespace, &table),
+            crate::table_catalog::TableCatalogStoreError::Unavailable(
+                "failed to acquire catalog table lock: quorum required 3, achieved 1".to_string(),
+            ),
+        )
+        .await;
+    let commit_backend = TableCommitObjectBackend::trusted(metadata_backend);
+
+    let error = publish_table_commit(
+        &store,
+        &commit_backend,
+        false,
+        crate::table_catalog::TableCommitRequest {
+            table_bucket: "warehouse".to_string(),
+            namespace: "analytics".to_string(),
+            table: "events".to_string(),
+            commit_id: "authority-unavailable".to_string(),
+            idempotency_key: None,
+            operation: "append".to_string(),
+            expected_version_token: before.version_token.clone(),
+            expected_metadata_location: before.metadata_location.clone(),
+            new_metadata_location: "tables/table-id/metadata/00002.metadata.json".to_string(),
+            requirements: Vec::new(),
+            writer: Some("authority-contract-test".to_string()),
+        },
+    )
+    .await
+    .expect_err("missing publication authority must fail before catalog publication");
+
+    assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_REST.into()));
+    assert_eq!(error.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
+    assert_eq!(error.message(), Some("table catalog is temporarily unavailable"));
+    let after = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should remain");
+    assert_eq!(after.metadata_location, before.metadata_location);
+    assert_eq!(after.version_token, before.version_token);
+    assert_eq!(after.generation, before.generation);
+    assert!(
+        store.commits.lock().await.is_empty(),
+        "an unattempted commit must not enter recovery history"
+    );
+}
+
 #[tokio::test]
 async fn rolling_upgrade_commit_retains_legacy_data_file_guard_until_publication_completes() {
     let metadata_backend = TestTableCatalogObjectBackend::content_addressed();

rustfs/src/table_catalog/error.rs

@@ -64,6 +64,7 @@ pub(crate) enum TableCatalogStoreError {
     Conflict(String),
     Invalid(String),
     Unsupported(String),
+    Unavailable(String),
     Internal(String),
 }
 
@@ -77,6 +78,7 @@ impl fmt::Display for TableCatalogStoreError {
             Self::Conflict(message) => write!(f, "table catalog conflict: {message}"),
             Self::Invalid(message) => write!(f, "invalid table catalog entry: {message}"),
             Self::Unsupported(message) => write!(f, "unsupported table catalog operation: {message}"),
+            Self::Unavailable(message) => write!(f, "table catalog temporarily unavailable: {message}"),
             Self::Internal(message) => write!(f, "table catalog store error: {message}"),
         }
     }

rustfs/src/table_catalog/iceberg/commit.rs

@@ -277,6 +277,7 @@ fn table_catalog_store_result_label<T>(result: &TableCatalogStoreResult<T>) -> &
             | TableCatalogStoreError::TableNotFound(_),
         ) => "not_found",
         Err(TableCatalogStoreError::Unsupported(_)) => "unsupported",
+        Err(TableCatalogStoreError::Unavailable(_)) => "unavailable",
         Err(TableCatalogStoreError::Internal(_)) => "failure",
     }
 }
@@ -382,3 +383,17 @@ pub(crate) fn table_commit_result(
     record_table_commit_result(table_bucket, namespace, table, commit_id, operation, started, &result);
     result
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn table_catalog_unavailable_result_label_is_stable() {
+        let result: TableCatalogStoreResult<()> = Err(TableCatalogStoreError::Unavailable(
+            "catalog publication authority is temporarily unavailable".to_string(),
+        ));
+
+        assert_eq!(table_catalog_store_result_label(&result), "unavailable");
+    }
+}

rustfs/src/table_catalog/store/mod.rs

@@ -28,6 +28,27 @@ pub(super) use strong::{
 };
 pub(crate) use strong::{StrongTableCatalogRuntime, StrongTableCatalogStore};
 
+pub(in crate::table_catalog) fn catalog_lock_acquisition_error(
+    operation: &str,
+    err: rustfs_lock::LockError,
+) -> TableCatalogStoreError {
+    let unavailable = matches!(
+        &err,
+        rustfs_lock::LockError::Timeout { .. }
+            | rustfs_lock::LockError::Network { .. }
+            | rustfs_lock::LockError::AlreadyLocked { .. }
+            | rustfs_lock::LockError::InsufficientNodes { .. }
+            | rustfs_lock::LockError::QuorumNotReached { .. }
+            | rustfs_lock::LockError::QueueFull { .. }
+    );
+    let message = format!("failed to {operation}: {err}");
+    if unavailable {
+        TableCatalogStoreError::Unavailable(message)
+    } else {
+        TableCatalogStoreError::Internal(message)
+    }
+}
+
 fn validate_table_bucket_entry(entry: &TableBucketEntry) -> TableCatalogStoreResult<()> {
     validate_catalog_entry_version("table bucket", entry.version)?;
     if entry.table_bucket.is_empty() {
@@ -1839,7 +1860,7 @@ where
         let guard = lock
             .get_write_lock(get_lock_acquire_timeout())
             .await
-            .map_err(|err| TableCatalogStoreError::Internal(format!("failed to acquire catalog table lock: {err}")))?;
+            .map_err(|err| catalog_lock_acquisition_error("acquire catalog table lock", err))?;
         Ok(TableCatalogLockGuard::namespace(guard))
     }
 
@@ -1852,7 +1873,7 @@ where
         let guard = lock
             .get_read_lock(get_lock_acquire_timeout())
             .await
-            .map_err(|err| TableCatalogStoreError::Internal(format!("failed to acquire catalog migration lock: {err}")))?;
+            .map_err(|err| catalog_lock_acquisition_error("acquire catalog migration lock", err))?;
         Ok(TableCatalogLockGuard::namespace(guard))
     }
 }

rustfs/src/table_catalog/test_support.rs

@@ -355,6 +355,7 @@ pub(crate) struct TestCatalogObjectBackend {
     pub(crate) corrupt_put_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
     pub(crate) missing_read_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
     pub(crate) fail_read_object_path: Arc<tokio::sync::Mutex<Option<String>>>,
+    pub(crate) fail_write_lock_path: Arc<tokio::sync::Mutex<Option<(String, TableCatalogStoreError)>>>,
     pub(crate) lock_attempts: Arc<tokio::sync::Mutex<Vec<(String, String)>>>,
     pub(crate) reject_reads_while_write_locked: bool,
     /// Content-addressed (sha256) etags instead of the store fake's counter.
@@ -939,6 +940,17 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
                 .entry((bucket.to_string(), object.to_string()))
                 .or_default() += 1;
         }
+        let injected_error = {
+            let mut failure = self.fail_write_lock_path.lock().await;
+            if failure.as_ref().is_some_and(|(path, _)| path == object) {
+                failure.take().map(|(_, error)| error)
+            } else {
+                None
+            }
+        };
+        if let Some(error) = injected_error {
+            return Err(error);
+        }
         let lock = {
             let mut locks = self.locks.lock().await;
             locks
@@ -1053,6 +1065,10 @@ impl TestCatalogObjectBackend {
         .await
         .expect("lock acquisition attempts should be observable");
     }
+
+    pub(crate) async fn fail_next_write_lock(&self, object: impl Into<String>, error: TableCatalogStoreError) {
+        *self.fail_write_lock_path.lock().await = Some((object.into(), error));
+    }
 }
 
 #[derive(Clone, Default)]

rustfs/src/table_catalog/tests.rs

@@ -20,6 +20,45 @@ use std::sync::Arc;
 
 const TABLE_CATALOG_TEST_TIMEOUT: StdDuration = StdDuration::from_secs(30);
 
+#[test]
+fn catalog_lock_authority_failures_are_typed_as_unavailable() {
+    for error in [
+        rustfs_lock::LockError::timeout("table-publication", StdDuration::from_secs(5)),
+        rustfs_lock::LockError::Network {
+            message: "peer unavailable".to_string(),
+            source: Box::new(std::io::Error::other("peer unavailable")),
+        },
+        rustfs_lock::LockError::AlreadyLocked {
+            resource: "table-publication".to_string(),
+            owner: "another-node".to_string(),
+        },
+        rustfs_lock::LockError::InsufficientNodes {
+            required: 3,
+            available: 1,
+        },
+        rustfs_lock::LockError::QuorumNotReached {
+            required: 3,
+            achieved: 1,
+        },
+        rustfs_lock::LockError::QueueFull {
+            message: "lock queue is full".to_string(),
+        },
+    ] {
+        assert_matches!(
+            super::store::catalog_lock_acquisition_error("acquire catalog table lock", error),
+            TableCatalogStoreError::Unavailable(_)
+        );
+    }
+
+    assert_matches!(
+        super::store::catalog_lock_acquisition_error(
+            "acquire catalog table lock",
+            rustfs_lock::LockError::configuration("invalid lock configuration"),
+        ),
+        TableCatalogStoreError::Internal(_)
+    );
+}
+
 #[test]
 fn reserved_table_object_key_matches_exact_prefix_and_children_only() {
     assert!(is_reserved_table_object_key(".rustfs-table"));