diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index 3b98f27ec..9882170dc 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/rustfs/src/admin/handlers/table_catalog/mod.rs @@ -160,6 +160,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "GET /v1/{prefix}/namespaces/{namespace}", "HEAD /v1/{prefix}/namespaces/{namespace}", "DELETE /v1/{prefix}/namespaces/{namespace}", + "POST /v1/{prefix}/namespaces/{namespace}/properties", "GET /v1/{prefix}/namespaces/{namespace}/tables", "POST /v1/{prefix}/namespaces/{namespace}/tables", "POST /v1/{prefix}/namespaces/{namespace}/register", @@ -176,7 +177,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "POST /v1/{prefix}/namespaces/{namespace}/views/{view}", "DELETE /v1/{prefix}/namespaces/{namespace}/views/{view}", ]; -const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &["POST /v1/{prefix}/namespaces/{namespace}/properties"]; +const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &[]; static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {}; static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {}; diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 0d68a36dc..d32239c15 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -434,7 +434,7 @@ fn catalog_config_response_lists_standard_rest_endpoints() { Some(REST_NAMESPACE_SEPARATOR_URL_ENCODED) ); assert!( - !response + response .endpoints .contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties") ); @@ -466,6 +466,7 @@ fn catalog_config_response_reports_durable_strong_backing_override() { .contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties") ); assert!(response.endpoints.contains(&"POST /v1/{prefix}/tables/rename")); + assert_eq!(response.endpoints.as_slice(), TABLE_CATALOG_ENDPOINTS); } #[test] @@ -11273,6 +11274,183 @@ async fn namespace_helpers_call_catalog_store() { assert!(list.namespaces.is_empty()); } +#[tokio::test] +#[serial_test::serial] +async fn namespace_property_handler_updates_object_backed_catalog_and_maps_errors() { + use crate::admin::storage_api::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + + temp_env::async_with_vars( + [( + crate::table_catalog::ENV_TABLE_CATALOG_BACKING, + Some(crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT), + )], + async { + let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await; + let bucket = format!("namespace-properties-{}", Uuid::new_v4().simple()); + object_store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("table bucket should be created"); + enable_table_bucket_marker(&object_store, &bucket) + .await + .expect("table bucket marker should be enabled"); + + rustfs_iam::store::object::ObjectStore::new(object_store.clone()) + .save_iam_config( + serde_json::json!({"version": 1}), + format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX), + ) + .await + .expect("request IAM format should be seeded"); + let iam = rustfs_iam::build_iam_sys(object_store.clone()) + .await + .expect("request IAM should initialize"); + let context = Arc::new(AppContext::new( + object_store.clone(), + Arc::new(RequestIam { handle: iam }), + Arc::new(RequestKms), + )); + let root_access_key = "namespace-properties-root"; + let root_secret_key = "namespace-properties-root-secret"; + assert!(context.publish_action_credentials(rustfs_credentials::Credentials { + access_key: root_access_key.to_string(), + secret_key: root_secret_key.to_string(), + status: "on".to_string(), + ..Default::default() + })); + let slot = ServerContextSlot::new(); + assert!(slot.install(context.clone())); + + let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime( + object_store, + context.table_catalog_strong_runtime(), + ); + let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test( + backend.clone(), + crate::table_catalog::TableCatalogBackingMode::ObjectBacked, + ); + catalog + .put_table_bucket(table_bucket_entry_from_metadata_marker(&bucket)) + .await + .expect("table bucket catalog entry should be seeded"); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let entry = crate::table_catalog::NamespaceEntry { + version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, + table_bucket: bucket.clone(), + namespace: namespace.public_name(), + namespace_id: namespace.storage_id(), + state: crate::table_catalog::TableCatalogEntryState::Active, + properties: BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]), + created_at: None, + updated_at: None, + }; + catalog + .create_namespace(entry.clone()) + .await + .expect("namespace should be seeded"); + + let request = |namespace: &str, body: serde_json::Value| { + let mut extensions = http::Extensions::new(); + extensions.insert(slot.clone()); + S3Request { + input: Body::from(serde_json::to_vec(&body).expect("request body should serialize")), + method: Method::POST, + uri: format!("/iceberg/v1/{bucket}/namespaces/{namespace}/properties") + .parse() + .expect("request URI should parse"), + headers: HeaderMap::new(), + extensions, + credentials: Some(s3s::auth::Credentials { + access_key: root_access_key.to_string(), + secret_key: s3s::auth::SecretKey::from(root_secret_key.to_string()), + }), + region: None, + service: None, + trailing_headers: None, + } + }; + let mut params_router = matchit::Router::new(); + params_router + .insert("/iceberg/v1/{warehouse}/namespaces/{namespace}/properties", ()) + .expect("handler parameter route should register"); + let success_path = format!("/iceberg/v1/{bucket}/namespaces/analytics/properties"); + let params = params_router + .at(&success_path) + .expect("success handler parameters should match") + .params; + let response = RestUpdateNamespacePropertiesHandler {} + .call( + request( + "analytics", + serde_json::json!({ + "removals": ["owner", "missing"], + "updates": {"retention": "30d"} + }), + ), + params, + ) + .await + .expect("handler should update object-backed namespace properties"); + assert_eq!(response.output.0, StatusCode::OK); + let body = http_body_util::BodyExt::collect(response.output.1) + .await + .expect("response body should collect") + .to_bytes(); + assert_eq!( + serde_json::from_slice::(&body).expect("response body should decode"), + serde_json::json!({ + "updated": ["retention"], + "removed": ["owner"], + "missing": ["missing"] + }) + ); + let persisted = catalog + .get_namespace(&bucket, &namespace.public_name()) + .await + .expect("updated namespace should load") + .expect("updated namespace should remain"); + assert_eq!(persisted.properties.get("retention").map(String::as_str), Some("30d")); + assert!(!persisted.properties.contains_key("owner")); + + let missing_path = format!("/iceberg/v1/{bucket}/namespaces/missing/properties"); + let params = params_router + .at(&missing_path) + .expect("missing handler parameters should match") + .params; + let missing = RestUpdateNamespacePropertiesHandler {} + .call(request("missing", serde_json::json!({"updates": {"owner": "platform"}})), params) + .await + .expect_err("missing namespace should fail"); + assert_eq!(missing.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_NO_SUCH_NAMESPACE.into())); + assert_eq!(missing.status_code(), Some(StatusCode::NOT_FOUND)); + + let corrupt = crate::table_catalog::Namespace::parse("corrupt").expect("namespace should parse"); + let corrupt_path = crate::table_catalog::TableCatalogObjectPaths::default().namespace_entry_path(&bucket, &corrupt); + backend + .put_object( + crate::admin::storage_api::RUSTFS_META_BUCKET, + &corrupt_path, + b"{".to_vec(), + crate::table_catalog::TableCatalogPutPrecondition::Any, + ) + .await + .expect("corrupt namespace entry should be seeded"); + let corrupt_request_path = format!("/iceberg/v1/{bucket}/namespaces/corrupt/properties"); + let params = params_router + .at(&corrupt_request_path) + .expect("corrupt handler parameters should match") + .params; + let corrupt = RestUpdateNamespacePropertiesHandler {} + .call(request("corrupt", serde_json::json!({"updates": {"owner": "platform"}})), params) + .await + .expect_err("corrupt namespace should fail"); + assert_eq!(corrupt.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into())); + assert_eq!(corrupt.status_code(), Some(StatusCode::BAD_REQUEST)); + }, + ) + .await; +} + #[tokio::test] async fn table_helpers_call_catalog_store() { let store = TestTableCatalogStore::default(); diff --git a/rustfs/src/table_catalog/store/mod.rs b/rustfs/src/table_catalog/store/mod.rs index 9043db5ba..22f75765a 100644 --- a/rustfs/src/table_catalog/store/mod.rs +++ b/rustfs/src/table_catalog/store/mod.rs @@ -1171,9 +1171,7 @@ where update: NamespacePropertiesUpdate, ) -> TableCatalogStoreResult { match self { - Self::ObjectBacked(_) => Err(TableCatalogStoreError::Unsupported( - "namespace property updates require durable-strong catalog backing".to_string(), - )), + Self::ObjectBacked(store) => store.update_namespace_properties(table_bucket, namespace, update).await, Self::DurableStrong(store) => store.update_namespace_properties(table_bucket, namespace, update).await, } } diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index d696414a4..e2c7249ee 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -4636,6 +4636,66 @@ where Ok(None) } + async fn update_namespace_properties( + &self, + table_bucket: &str, + namespace: &str, + update: NamespacePropertiesUpdate, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + self.require_table_bucket(table_bucket).await?; + let _migration_guard = self.acquire_object_backed_catalog_write_permit(table_bucket).await?; + let bucket_path = self.paths.table_bucket_entry_path(table_bucket); + let _bucket_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &bucket_path).await?; + let namespace_path = self.paths.namespace_entry_path(table_bucket, &namespace); + let _namespace_guard = self + .backend + .acquire_write_lock(self.catalog_bucket(), &namespace_path) + .await?; + + let current = self + .read_entry_unlocked::(self.catalog_bucket(), &namespace_path) + .await?; + let (mut next, precondition) = match current { + Some((entry, etag)) => { + validate_namespace_entry_object(&self.paths, &namespace_path, &entry)?; + validate_namespace_properties(&entry.properties)?; + if entry.state != TableCatalogEntryState::Active { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {table_bucket}/{}", + namespace.public_name() + ))); + } + let etag = etag.ok_or_else(|| { + TableCatalogStoreError::Internal(format!("catalog namespace entry has no etag: {namespace_path}")) + })?; + (entry, TableCatalogPutPrecondition::IfMatch(etag)) + } + None => { + if !self.has_active_namespace_object(table_bucket, &namespace).await? + && !self.has_active_namespace_descendant(table_bucket, &namespace).await? + { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {table_bucket}/{}", + namespace.public_name() + ))); + } + (synthetic_namespace_entry(table_bucket, &namespace), TableCatalogPutPrecondition::IfAbsent) + } + }; + + let before = next.clone(); + let result = update.apply_to(&mut next); + validate_namespace_entry_object(&self.paths, &namespace_path, &next)?; + validate_namespace_properties(&next.properties)?; + if before == next { + return Ok(result); + } + self.write_entry_unlocked(self.catalog_bucket(), &namespace_path, &next, precondition) + .await?; + Ok(result) + } + async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult> { let parent = parse_namespace_for_store(parent)?; let prefix = format!("{}{}/", self.paths.namespace_entries_prefix(table_bucket), parent.storage_id()); diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index 5e88cc213..40f3d7c88 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -16617,13 +16617,16 @@ fn namespace_property_update_and_limits_reject_ambiguous_or_oversized_state() { } #[tokio::test] -async fn configured_object_catalog_rejects_namespace_property_update_without_mutation() { +async fn configured_object_catalog_updates_namespace_properties() { let backend = TestCatalogObjectBackend::default(); let store = ConfiguredTableCatalogStore::new_for_test(backend, TableCatalogBackingMode::ObjectBacked); let bucket = "analytics"; let namespace = Namespace::parse("sales").expect("namespace should parse"); let mut entry = test_namespace_entry(bucket, &namespace); - entry.properties = BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]); + entry.properties = BTreeMap::from([ + ("obsolete".to_string(), "true".to_string()), + ("owner".to_string(), "lakehouse".to_string()), + ]); store .put_table_bucket(test_bucket_entry(bucket)) .await @@ -16634,18 +16637,354 @@ async fn configured_object_catalog_rejects_namespace_property_update_without_mut .update_namespace_properties( bucket, "sales", - NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())])) - .expect("namespace update should validate"), + NamespacePropertiesUpdate::try_new( + vec!["obsolete".to_string(), "missing".to_string()], + BTreeMap::from([ + ("owner".to_string(), "platform".to_string()), + ("retention".to_string(), "30d".to_string()), + ]), + ) + .expect("namespace update should validate"), ) .await - .expect_err("object-backed namespace property update should be unsupported"); - assert_matches!(result, TableCatalogStoreError::Unsupported(_)); + .expect("object-backed namespace properties should update"); + assert_eq!(result.updated, vec!["owner".to_string(), "retention".to_string()]); + assert_eq!(result.removed, vec!["obsolete".to_string()]); + assert_eq!(result.missing, vec!["missing".to_string()]); let stored = store .get_namespace(bucket, "sales") .await .expect("namespace lookup should succeed") .expect("namespace should remain"); - assert_eq!(stored.properties.get("owner").map(String::as_str), Some("lakehouse")); + assert_eq!( + stored.properties, + BTreeMap::from([ + ("owner".to_string(), "platform".to_string()), + ("retention".to_string(), "30d".to_string()), + ]) + ); +} + +#[tokio::test] +async fn object_catalog_namespace_property_update_materializes_implicit_parent() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let parent = Namespace::parse("sales").expect("parent namespace should parse"); + let child = Namespace::parse("sales.daily").expect("child namespace should parse"); + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket entry should be seeded"); + let child_entry = test_namespace_entry(bucket, &child); + store + .create_namespace(child_entry.clone()) + .await + .expect("child namespace should be created"); + + let no_change = store + .update_namespace_properties( + bucket, + &parent.public_name(), + NamespacePropertiesUpdate::try_new(vec!["missing".to_string()], BTreeMap::new()) + .expect("namespace update should validate"), + ) + .await + .expect("implicit parent no-op should succeed"); + assert_eq!(no_change.missing, vec!["missing".to_string()]); + let parent_path = store.paths.namespace_entry_path(bucket, &parent); + assert!( + store + .read_entry::(store.catalog_bucket(), &parent_path) + .await + .expect("implicit parent lookup should succeed") + .is_none() + ); + + let result = store + .update_namespace_properties( + bucket, + &parent.public_name(), + NamespacePropertiesUpdate::try_new( + vec!["missing".to_string()], + BTreeMap::from([("owner".to_string(), "platform".to_string())]), + ) + .expect("namespace update should validate"), + ) + .await + .expect("implicit parent should materialize"); + + assert_eq!(result.updated, vec!["owner".to_string()]); + assert!(result.removed.is_empty()); + assert_eq!(result.missing, vec!["missing".to_string()]); + let (materialized, _) = store + .read_entry::(store.catalog_bucket(), &parent_path) + .await + .expect("materialized parent should load") + .expect("parent should have an explicit entry"); + assert_eq!(materialized.properties.get("owner").map(String::as_str), Some("platform")); + assert_eq!( + store + .get_namespace(bucket, &child.public_name()) + .await + .expect("child lookup should succeed"), + Some(child_entry) + ); +} + +#[tokio::test] +async fn object_catalog_namespace_property_update_materializes_resource_only_parents() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let table_namespace = Namespace::parse("table_only").expect("table namespace should parse"); + let view_namespace = Namespace::parse("view_only").expect("view namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let view = IdentifierSegment::parse("recent_orders").expect("view should parse"); + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket entry should be seeded"); + + let table_entry = test_table_entry( + bucket, + &table_namespace, + &table, + default_table_metadata_file_path(&table_namespace, &table, "00001.metadata.json"), + ); + backend + .seed_object( + RUSTFS_META_BUCKET, + &store.paths.table_entry_path(bucket, &table_namespace, &table), + serde_json::to_vec(&table_entry).expect("table entry should serialize"), + ) + .await; + let view_entry = test_view_entry( + bucket, + &view_namespace, + &view, + default_view_metadata_file_path(&view_namespace, &view, "00001.view.json"), + ); + backend + .seed_object( + RUSTFS_META_BUCKET, + &store.paths.view_entry_path(bucket, &view_namespace, &view), + serde_json::to_vec(&view_entry).expect("view entry should serialize"), + ) + .await; + + let table_before = store + .load_table(bucket, &table_namespace.public_name(), table.as_str()) + .await + .expect("table should load before materializing its namespace"); + let view_before = store + .load_view(bucket, &view_namespace.public_name(), view.as_str()) + .await + .expect("view should load before materializing its namespace"); + + for namespace in [&table_namespace, &view_namespace] { + let result = store + .update_namespace_properties( + bucket, + &namespace.public_name(), + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())])) + .expect("namespace update should validate"), + ) + .await + .expect("active resource should prove the implicit namespace"); + assert_eq!(result.updated, vec!["owner".to_string()]); + let materialized = store + .get_namespace(bucket, &namespace.public_name()) + .await + .expect("materialized namespace should load") + .expect("materialized namespace should exist"); + assert_eq!(materialized.properties.get("owner").map(String::as_str), Some("platform")); + } + + assert_eq!( + store + .load_table(bucket, &table_namespace.public_name(), table.as_str()) + .await + .expect("table should load after materializing its namespace"), + table_before + ); + assert_eq!( + store + .load_view(bucket, &view_namespace.public_name(), view.as_str()) + .await + .expect("view should load after materializing its namespace"), + view_before + ); +} + +#[tokio::test] +async fn object_catalog_namespace_property_update_requires_etag_and_skips_noop_writes() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket entry should be seeded"); + + let etagless = Namespace::parse("etagless").expect("namespace should parse"); + let mut etagless_entry = test_namespace_entry(bucket, &etagless); + etagless_entry.properties.insert("owner".to_string(), "lakehouse".to_string()); + store + .create_namespace(etagless_entry) + .await + .expect("etagless namespace should be seeded"); + let etagless_path = store.paths.namespace_entry_path(bucket, &etagless); + backend.omit_etag_for_object(RUSTFS_META_BUCKET, &etagless_path).await; + let etagless_puts = backend.put_attempt_count(RUSTFS_META_BUCKET, &etagless_path).await; + + assert_matches!( + store + .update_namespace_properties( + bucket, + &etagless.public_name(), + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())]),) + .expect("namespace update should validate"), + ) + .await, + Err(TableCatalogStoreError::Internal(_)) + ); + assert_eq!(backend.put_attempt_count(RUSTFS_META_BUCKET, &etagless_path).await, etagless_puts); + let unchanged = store + .get_namespace(bucket, &etagless.public_name()) + .await + .expect("etagless namespace should still load") + .expect("etagless namespace should remain"); + assert_eq!(unchanged.properties.get("owner").map(String::as_str), Some("lakehouse")); + + let no_op = Namespace::parse("no_op").expect("namespace should parse"); + let mut no_op_entry = test_namespace_entry(bucket, &no_op); + no_op_entry.properties.insert("owner".to_string(), "lakehouse".to_string()); + store + .create_namespace(no_op_entry) + .await + .expect("no-op namespace should be seeded"); + let no_op_path = store.paths.namespace_entry_path(bucket, &no_op); + backend.fail_next_put(RUSTFS_META_BUCKET, &no_op_path).await; + let puts_before_no_op = backend.put_attempt_count(RUSTFS_META_BUCKET, &no_op_path).await; + + let result = store + .update_namespace_properties( + bucket, + &no_op.public_name(), + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "lakehouse".to_string())])) + .expect("namespace update should validate"), + ) + .await + .expect("unchanged namespace properties should not write"); + assert_eq!(result.updated, vec!["owner".to_string()]); + assert_eq!(backend.put_attempt_count(RUSTFS_META_BUCKET, &no_op_path).await, puts_before_no_op); + + assert_matches!( + store + .update_namespace_properties( + bucket, + &no_op.public_name(), + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())]),) + .expect("namespace update should validate"), + ) + .await, + Err(TableCatalogStoreError::Internal(_)) + ); + let unchanged = store + .get_namespace(bucket, &no_op.public_name()) + .await + .expect("namespace should load after failed write") + .expect("namespace should remain"); + assert_eq!(unchanged.properties.get("owner").map(String::as_str), Some("lakehouse")); +} + +#[tokio::test] +async fn object_catalog_namespace_property_update_rejects_missing_inactive_and_corrupt_entries() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket entry should be seeded"); + let update = || { + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())])) + .expect("namespace update should validate") + }; + + assert_matches!( + store.update_namespace_properties(bucket, "missing", update()).await, + Err(TableCatalogStoreError::NotFound(_)) + ); + + let inactive = Namespace::parse("inactive").expect("inactive namespace should parse"); + let mut inactive_entry = test_namespace_entry(bucket, &inactive); + inactive_entry.state = TableCatalogEntryState::Deleted; + backend + .seed_object( + RUSTFS_META_BUCKET, + &store.paths.namespace_entry_path(bucket, &inactive), + serde_json::to_vec(&inactive_entry).expect("inactive namespace should encode"), + ) + .await; + assert_matches!( + store + .update_namespace_properties(bucket, &inactive.public_name(), update()) + .await, + Err(TableCatalogStoreError::NotFound(_)) + ); + + let corrupt = Namespace::parse("corrupt").expect("corrupt namespace should parse"); + backend + .seed_object(RUSTFS_META_BUCKET, &store.paths.namespace_entry_path(bucket, &corrupt), b"{".to_vec()) + .await; + assert_matches!( + store + .update_namespace_properties(bucket, &corrupt.public_name(), update()) + .await, + Err(TableCatalogStoreError::Invalid(_)) + ); + + let semantically_corrupt = Namespace::parse("semantically_corrupt").expect("corrupt namespace should parse"); + let mut semantically_corrupt_entry = test_namespace_entry(bucket, &semantically_corrupt); + semantically_corrupt_entry.properties = (0..=NAMESPACE_PROPERTIES_MAX_ENTRIES) + .map(|index| (format!("key{index}"), "value".to_string())) + .collect(); + let semantically_corrupt_path = store.paths.namespace_entry_path(bucket, &semantically_corrupt); + backend + .seed_object( + RUSTFS_META_BUCKET, + &semantically_corrupt_path, + serde_json::to_vec(&semantically_corrupt_entry).expect("corrupt namespace should encode"), + ) + .await; + let put_attempts = backend + .put_attempt_count(RUSTFS_META_BUCKET, &semantically_corrupt_path) + .await; + let repair_update = + NamespacePropertiesUpdate::try_new(vec![format!("key{NAMESPACE_PROPERTIES_MAX_ENTRIES}")], BTreeMap::new()) + .expect("repair request should validate structurally"); + + assert_matches!( + store + .update_namespace_properties(bucket, &semantically_corrupt.public_name(), repair_update,) + .await, + Err(TableCatalogStoreError::Invalid(_)) + ); + assert_eq!( + backend + .put_attempt_count(RUSTFS_META_BUCKET, &semantically_corrupt_path) + .await, + put_attempts + ); + let persisted = store + .read_entry::(RUSTFS_META_BUCKET, &semantically_corrupt_path) + .await + .expect("corrupt namespace lookup should succeed") + .expect("corrupt namespace should remain") + .0; + assert_eq!(persisted.properties.len(), NAMESPACE_PROPERTIES_MAX_ENTRIES + 1); } #[tokio::test] @@ -16984,6 +17323,59 @@ async fn object_catalog_namespace_replacement_is_fenced_by_observed_etag() { assert_eq!(stored.properties.get("owner").map(String::as_str), Some("winner")); } +#[tokio::test] +async fn object_catalog_namespace_property_update_is_fenced_by_observed_etag() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + + let namespace_path = store.paths.namespace_entry_path(bucket, &namespace); + let pause = backend.pause_next_put(RUSTFS_META_BUCKET, &namespace_path).await; + let stale_store = store.clone(); + let stale_update = tokio::spawn(async move { + stale_store + .update_namespace_properties( + bucket, + "sales", + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "stale".to_string())])) + .expect("namespace update should validate"), + ) + .await + }); + pause.wait_started().await; + + let mut winner = test_namespace_entry(bucket, &namespace); + winner.properties.insert("owner".to_string(), "winner".to_string()); + backend + .seed_object( + RUSTFS_META_BUCKET, + &namespace_path, + serde_json::to_vec(&winner).expect("winning namespace should encode"), + ) + .await; + pause.release(); + + assert_matches!( + stale_update.await.expect("stale namespace update task should finish"), + Err(TableCatalogStoreError::Conflict(_)) + ); + let stored = store + .get_namespace(bucket, &namespace.public_name()) + .await + .expect("winning namespace should load") + .expect("winning namespace should remain"); + assert_eq!(stored.properties.get("owner").map(String::as_str), Some("winner")); +} + async fn assert_direct_namespace_child_contract(store: &S, bucket: &str, cursor_prefix: &str) where S: TableCatalogStore + ?Sized,