From be98c1f86a69751f575ed867ea8119185e7a77a4 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 3 Jun 2026 21:16:59 +0800 Subject: [PATCH] feat(table): add catalog boundary primitives (#3173) Co-authored-by: Henry Guo --- crates/ecstore/src/bucket/metadata.rs | 50 +- crates/ecstore/src/store/bucket.rs | 51 +- rustfs/src/app/multipart_usecase.rs | 15 + rustfs/src/app/object_usecase.rs | 24 + rustfs/src/lib.rs | 1 + rustfs/src/storage/ecfs.rs | 15 + rustfs/src/table_catalog.rs | 859 ++++++++++++++++++++++++++ 7 files changed, 1011 insertions(+), 4 deletions(-) create mode 100644 rustfs/src/table_catalog.rs diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index c82903bb9..f2f9d8c2b 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -244,6 +244,8 @@ pub const BUCKET_ACCELERATE_CONFIG: &str = "accelerate.xml"; pub const BUCKET_REQUEST_PAYMENT_CONFIG: &str = "request-payment.xml"; pub const BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG: &str = "public-access-block.xml"; pub const BUCKET_ACL_CONFIG: &str = "bucket-acl.json"; +pub const BUCKET_TABLE_CONFIG: &str = "table-bucket.json"; +pub const BUCKET_TABLE_RESERVED_PREFIX: &str = ".rustfs-table"; #[derive(Debug, Clone)] pub struct BucketMetadata { @@ -268,6 +270,7 @@ pub struct BucketMetadata { pub request_payment_config_xml: Vec, pub public_access_block_config_xml: Vec, pub bucket_acl_config_json: Vec, + pub table_bucket_config_json: Vec, pub policy_config_updated_at: OffsetDateTime, pub object_lock_config_updated_at: OffsetDateTime, @@ -287,6 +290,7 @@ pub struct BucketMetadata { pub request_payment_config_updated_at: OffsetDateTime, pub public_access_block_config_updated_at: OffsetDateTime, pub bucket_acl_config_updated_at: OffsetDateTime, + pub table_bucket_config_updated_at: OffsetDateTime, pub new_field_updated_at: OffsetDateTime, @@ -334,6 +338,7 @@ impl Default for BucketMetadata { request_payment_config_xml: Default::default(), public_access_block_config_xml: Default::default(), bucket_acl_config_json: Default::default(), + table_bucket_config_json: Default::default(), policy_config_updated_at: OffsetDateTime::UNIX_EPOCH, object_lock_config_updated_at: OffsetDateTime::UNIX_EPOCH, encryption_config_updated_at: OffsetDateTime::UNIX_EPOCH, @@ -352,6 +357,7 @@ impl Default for BucketMetadata { request_payment_config_updated_at: OffsetDateTime::UNIX_EPOCH, public_access_block_config_updated_at: OffsetDateTime::UNIX_EPOCH, bucket_acl_config_updated_at: OffsetDateTime::UNIX_EPOCH, + table_bucket_config_updated_at: OffsetDateTime::UNIX_EPOCH, new_field_updated_at: OffsetDateTime::UNIX_EPOCH, policy_config: Default::default(), notification_config: Default::default(), @@ -397,6 +403,10 @@ impl BucketMetadata { self.lock_enabled || (self.versioning_config.as_ref().is_some_and(|v| v.enabled())) } + pub fn table_bucket_enabled(&self) -> bool { + !self.table_bucket_config_json.is_empty() + } + /// Decode from msgp bytes. Field order follows MinIO BucketMetadata for compatibility. pub fn decode_from(&mut self, rd: &mut R) -> Result<()> { let mut fields = rmp::decode::read_map_len(rd)?; @@ -447,6 +457,7 @@ impl BucketMetadata { self.public_access_block_config_xml = read_msgp_bin(rd)? } "BucketAclConfigJSON" | "BucketAclConfigJson" => self.bucket_acl_config_json = read_msgp_bin(rd)?, + "TableBucketConfigJSON" | "TableBucketConfigJson" => self.table_bucket_config_json = read_msgp_bin(rd)?, "CorsConfigUpdatedAt" => self.cors_config_updated_at = read_msgp_time_value(rd)?, "LoggingConfigUpdatedAt" => self.logging_config_updated_at = read_msgp_time_value(rd)?, "WebsiteConfigUpdatedAt" => self.website_config_updated_at = read_msgp_time_value(rd)?, @@ -454,6 +465,7 @@ impl BucketMetadata { "RequestPaymentConfigUpdatedAt" => self.request_payment_config_updated_at = read_msgp_time_value(rd)?, "PublicAccessBlockConfigUpdatedAt" => self.public_access_block_config_updated_at = read_msgp_time_value(rd)?, "BucketAclConfigUpdatedAt" => self.bucket_acl_config_updated_at = read_msgp_time_value(rd)?, + "TableBucketConfigUpdatedAt" => self.table_bucket_config_updated_at = read_msgp_time_value(rd)?, other => { tracing::debug!(field = %other, "BucketMetadata decode_from: skipping unknown field"); skip_msgp_value(rd)?; @@ -466,8 +478,8 @@ impl BucketMetadata { /// Encode to msgp bytes. Field order follows MinIO BucketMetadata for compatibility. pub fn encode_to(&self, wr: &mut W) -> Result<()> { - // Map size: MinIO fields (25) + RustFS extensions (14) - let map_len: u32 = 39; + // Map size: MinIO fields (25) + RustFS extensions (16) + let map_len: u32 = 41; rmp::encode::write_map_len(wr, map_len)?; // MinIO field order (same as Go struct) @@ -523,6 +535,7 @@ impl BucketMetadata { write_bin_field(wr, "RequestPaymentConfigXML", &self.request_payment_config_xml)?; write_bin_field(wr, "PublicAccessBlockConfigXML", &self.public_access_block_config_xml)?; write_bin_field(wr, "BucketAclConfigJSON", &self.bucket_acl_config_json)?; + write_bin_field(wr, "TableBucketConfigJSON", &self.table_bucket_config_json)?; rmp::encode::write_str(wr, "CorsConfigUpdatedAt")?; write_msgp_time(wr, self.cors_config_updated_at)?; rmp::encode::write_str(wr, "LoggingConfigUpdatedAt")?; @@ -537,6 +550,8 @@ impl BucketMetadata { write_msgp_time(wr, self.public_access_block_config_updated_at)?; rmp::encode::write_str(wr, "BucketAclConfigUpdatedAt")?; write_msgp_time(wr, self.bucket_acl_config_updated_at)?; + rmp::encode::write_str(wr, "TableBucketConfigUpdatedAt")?; + write_msgp_time(wr, self.table_bucket_config_updated_at)?; Ok(()) } @@ -632,6 +647,9 @@ impl BucketMetadata { if self.bucket_acl_config_updated_at == OffsetDateTime::UNIX_EPOCH { self.bucket_acl_config_updated_at = self.created } + if self.table_bucket_config_updated_at == OffsetDateTime::UNIX_EPOCH { + self.table_bucket_config_updated_at = self.created + } } pub fn update_config(&mut self, config_file: &str, data: Vec) -> Result { @@ -709,6 +727,10 @@ impl BucketMetadata { self.bucket_acl_config_json = data; self.bucket_acl_config_updated_at = updated; } + BUCKET_TABLE_CONFIG => { + self.table_bucket_config_json = data; + self.table_bucket_config_updated_at = updated; + } _ => return Err(Error::other(format!("config file not found : {config_file}"))), } @@ -1028,6 +1050,10 @@ mod test { bm.bucket_acl_config_json = bucket_acl.as_bytes().to_vec(); bm.bucket_acl_config_updated_at = OffsetDateTime::now_utc(); + let table_bucket_marker = r#"{"enabled":true}"#; + bm.table_bucket_config_json = table_bucket_marker.as_bytes().to_vec(); + bm.table_bucket_config_updated_at = OffsetDateTime::now_utc(); + // Test serialization let buf = bm.marshal_msg().unwrap(); assert!(!buf.is_empty(), "Serialized buffer should not be empty"); @@ -1049,6 +1075,7 @@ mod test { assert_eq!(bm.quota_config_json, deserialized_bm.quota_config_json); assert_eq!(bm.public_access_block_config_xml, deserialized_bm.public_access_block_config_xml); assert_eq!(bm.bucket_acl_config_json, deserialized_bm.bucket_acl_config_json); + assert_eq!(bm.table_bucket_config_json, deserialized_bm.table_bucket_config_json); assert_eq!(bm.object_lock_config_xml, deserialized_bm.object_lock_config_xml); assert_eq!(bm.notification_config_xml, deserialized_bm.notification_config_xml); assert_eq!(bm.replication_config_xml, deserialized_bm.replication_config_xml); @@ -1100,6 +1127,11 @@ mod test { bm.bucket_targets_config_meta_updated_at.unix_timestamp(), deserialized_bm.bucket_targets_config_meta_updated_at.unix_timestamp() ); + assert_eq!( + bm.table_bucket_config_updated_at.unix_timestamp(), + deserialized_bm.table_bucket_config_updated_at.unix_timestamp() + ); + assert!(deserialized_bm.table_bucket_enabled()); // Test that the serialized data contains expected content let buf_str = String::from_utf8_lossy(&buf); @@ -1116,6 +1148,20 @@ mod test { println!(" - Serialized buffer size: {} bytes", buf.len()); } + #[test] + fn table_bucket_marker_tracks_config_presence() { + let mut bm = BucketMetadata::new("table-bucket"); + assert!(!bm.table_bucket_enabled()); + + bm.update_config(BUCKET_TABLE_CONFIG, br#"{"enabled":true}"#.to_vec()) + .unwrap(); + assert!(bm.table_bucket_enabled()); + assert!(!bm.table_bucket_config_json.is_empty()); + + bm.update_config(BUCKET_TABLE_CONFIG, Vec::new()).unwrap(); + assert!(!bm.table_bucket_enabled()); + } + /// After policy deletion (policy_config_json cleared), parse_policy_config sets policy_config to None. #[test] fn test_parse_policy_config_clears_cache_when_json_empty() { diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 13bb0b7b6..a49843bd8 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -13,13 +13,48 @@ // limitations under the License. use super::*; -use crate::bucket::utils::is_meta_bucketname; +use crate::bucket::{metadata::BUCKET_TABLE_RESERVED_PREFIX, utils::is_meta_bucketname}; use crate::set_disk::get_lock_acquire_timeout; fn should_override_created_from_metadata(created: OffsetDateTime) -> bool { created != OffsetDateTime::UNIX_EPOCH } +fn validate_table_bucket_delete_allowed( + bucket: &str, + table_bucket_enabled: bool, + table_catalog_metadata_exists: bool, +) -> Result<()> { + if table_bucket_enabled && table_catalog_metadata_exists { + return Err(StorageError::BucketNotEmpty(bucket.to_string())); + } + + Ok(()) +} + +async fn table_catalog_metadata_exists(bucket: &str) -> bool { + let local_disks = all_local_disk().await; + for disk in local_disks.iter() { + let catalog_path = disk.path().join(bucket).join(BUCKET_TABLE_RESERVED_PREFIX); + if has_xlmeta_files(&catalog_path).await { + return true; + } + } + + false +} + +async fn validate_table_bucket_delete_guard(bucket: &str) -> Result<()> { + let table_bucket_enabled = metadata_sys::get(bucket) + .await + .is_ok_and(|metadata| metadata.table_bucket_enabled()); + if table_bucket_enabled { + validate_table_bucket_delete_allowed(bucket, true, table_catalog_metadata_exists(bucket).await)?; + } + + Ok(()) +} + impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { @@ -167,6 +202,8 @@ impl ECStore { return Err(to_object_err(storage_err, vec![bucket])); } + validate_table_bucket_delete_guard(bucket).await?; + // Check bucket is empty before deletion (per S3 API spec) // If bucket is not empty (contains actual objects with xl.meta files) and force // is not set, return BucketNotEmpty error. @@ -203,7 +240,8 @@ impl ECStore { #[cfg(test)] mod tests { - use super::should_override_created_from_metadata; + use super::{should_override_created_from_metadata, validate_table_bucket_delete_allowed}; + use crate::error::StorageError; use time::OffsetDateTime; #[test] @@ -216,4 +254,13 @@ mod tests { let created = OffsetDateTime::from_unix_timestamp(1704067200).expect("valid timestamp"); assert!(should_override_created_from_metadata(created)); } + + #[test] + fn table_bucket_delete_guard_rejects_remaining_catalog_metadata() { + let err = validate_table_bucket_delete_allowed("table-bucket", true, true).unwrap_err(); + + assert!(matches!(err, StorageError::BucketNotEmpty(bucket) if bucket == "table-bucket")); + assert!(validate_table_bucket_delete_allowed("table-bucket", true, false).is_ok()); + assert!(validate_table_bucket_delete_allowed("regular-bucket", false, true).is_ok()); + } } diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 68c41a6b8..6075f4aa8 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -30,6 +30,7 @@ use crate::storage::s3_api::multipart::{ }; use crate::storage::sse::{build_ssec_read_headers, encryption_material_to_metadata, map_get_object_reader_error}; use crate::storage::*; +use crate::table_catalog; use bytes::Bytes; use futures::StreamExt; use http::{HeaderMap, Uri}; @@ -141,6 +142,12 @@ fn normalize_complete_multipart_parts(parts: Vec) -> S3Result S3Result<()> { + table_catalog::validate_bucket_object_mutation(bucket, key) + .await + .map_err(|_| s3_error!(InvalidRequest, "{}", table_catalog::RESERVED_CATALOG_OBJECT_MESSAGE)) +} + fn has_complete_multipart_object_lock_headers(headers: &HeaderMap) -> bool { headers.contains_key(X_AMZ_OBJECT_LOCK_MODE) || headers.contains_key(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) @@ -275,6 +282,8 @@ impl DefaultMultipartUsecase { .. } = input; + validate_table_catalog_object_mutation(&bucket, &key).await?; + if if_match.is_some() || if_none_match.is_some() { let Some(store) = new_object_layer_fn() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); @@ -534,6 +543,8 @@ impl DefaultMultipartUsecase { return Err(s3_error!(InvalidStorageClass)); } + validate_table_catalog_object_mutation(&bucket, &key).await?; + let Some(store) = new_object_layer_fn() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -677,6 +688,8 @@ impl DefaultMultipartUsecase { let part_id = parse_upload_part_number(part_number)?; + validate_table_catalog_object_mutation(&bucket, &key).await?; + let mut size = content_length; let mut body_stream = body.ok_or_else(|| s3_error!(IncompleteBody))?; @@ -1015,6 +1028,8 @@ impl DefaultMultipartUsecase { let part_id = parse_upload_part_number(part_number)?; + validate_table_catalog_object_mutation(&bucket, &key).await?; + let Some(store) = new_object_layer_fn() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index c1f1dcc4e..cb55cb4eb 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -34,6 +34,7 @@ use crate::storage::s3_api::multipart::parse_list_parts_params; use crate::storage::sse::{SSEType, build_ssec_read_headers, encryption_material_to_metadata, map_get_object_reader_error}; use crate::storage::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper}; use crate::storage::*; +use crate::table_catalog; use bytes::Bytes; use futures::StreamExt; use http::{HeaderMap, HeaderValue, StatusCode}; @@ -151,6 +152,12 @@ fn request_uses_aws_chunked(headers: &HeaderMap) -> bool { has_aws_chunked("content-encoding") || has_aws_chunked("transfer-encoding") } +async fn validate_table_catalog_object_mutation(bucket: &str, key: &str) -> S3Result<()> { + table_catalog::validate_bucket_object_mutation(bucket, key) + .await + .map_err(|_| s3_error!(InvalidRequest, "{}", table_catalog::RESERVED_CATALOG_OBJECT_MESSAGE)) +} + struct DeadlockRequestGuard { deadlock_detector: Arc, request_id: String, @@ -1751,6 +1758,7 @@ impl DefaultObjectUsecase { // Validate object key validate_object_key(&key, request_method_name)?; + validate_table_catalog_object_mutation(&bucket, &key).await?; if let Some(size) = content_length { self.check_bucket_quota(&bucket, quota_operation, size as u64).await?; @@ -2701,6 +2709,7 @@ impl DefaultObjectUsecase { // Validate both source and destination keys validate_object_key(&src_key, "COPY (source)")?; validate_object_key(&key, "COPY (dest)")?; + validate_table_catalog_object_mutation(&bucket, &key).await?; // AWS S3 allows self-copy when metadata directive is REPLACE (used to update metadata in-place). // Reject only when the directive is not REPLACE. @@ -3083,6 +3092,16 @@ impl DefaultObjectUsecase { } } + if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await { + delete_results[idx].error = Some(Error { + code: Some("InvalidRequest".to_string()), + key: Some(obj_id.key.clone()), + message: Some(err.to_string()), + version_id: version_id.clone(), + }); + continue; + } + let mut object = ObjectToDelete { object_name: obj_id.key.clone(), version_id: version_uuid, @@ -3322,6 +3341,7 @@ impl DefaultObjectUsecase { // Validate object key validate_object_key(&key, "DELETE")?; + validate_table_catalog_object_mutation(&bucket, &key).await?; let replica = req .headers @@ -3877,6 +3897,8 @@ impl DefaultObjectUsecase { .. } = req.input.clone(); + validate_table_catalog_object_mutation(&bucket, &object).await?; + let rreq = rreq.ok_or_else(|| { S3Error::with_message(S3ErrorCode::Custom("ErrValidRestoreObject".into()), "restore request is required") })?; @@ -4170,6 +4192,7 @@ impl DefaultObjectUsecase { return Err(s3_error!(UnexpectedContent)); } validate_object_key(&key, "PUT")?; + validate_table_catalog_object_mutation(&bucket, &key).await?; self.check_bucket_quota(&bucket, QuotaOperation::PutObject, size as u64) .await?; @@ -4276,6 +4299,7 @@ impl DefaultObjectUsecase { return Err(err); } }; + validate_table_catalog_object_mutation(&bucket, &fpath).await?; let mut auth_req = S3Request { input: PutObjectInput::default(), diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index 050eacf19..020f45c48 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -69,6 +69,7 @@ pub mod protocols; pub mod server; pub mod startup_fs_guard; pub mod storage; +pub(crate) mod table_catalog; pub mod tls; pub mod update; pub mod version; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index c781c05e8..3d3c89c72 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -21,6 +21,7 @@ use crate::storage::helper::OperationHelper; use crate::storage::options::get_opts; use crate::storage::s3_api::acl; use crate::storage::{parse_object_lock_legal_hold, parse_object_lock_retention, validate_bucket_object_lock_enabled}; +use crate::table_catalog; use http::StatusCode; use metrics::{counter, histogram}; use rustfs_ecstore::{ @@ -154,6 +155,12 @@ pub(crate) fn parse_object_version_id(version_id: Option) -> S3Result S3Result<()> { + table_catalog::validate_bucket_object_mutation(bucket, key) + .await + .map_err(|_| s3_error!(InvalidRequest, "{}", table_catalog::RESERVED_CATALOG_OBJECT_MESSAGE)) +} + const MAXIMUM_RETENTION_DAYS: i32 = 36_500; const MAXIMUM_RETENTION_YEARS: i32 = 100; @@ -383,6 +390,8 @@ impl S3 for FS { .. } = req.input.clone(); + validate_table_catalog_object_mutation(&bucket, &object).await?; + let Some(store) = new_object_layer_fn() else { error!("Store not initialized"); return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); @@ -1176,6 +1185,8 @@ impl S3 for FS { .. } = req.input.clone(); + validate_table_catalog_object_mutation(&bucket, &key).await?; + let Some(store) = new_object_layer_fn() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1298,6 +1309,8 @@ impl S3 for FS { .. } = req.input.clone(); + validate_table_catalog_object_mutation(&bucket, &key).await?; + let Some(store) = new_object_layer_fn() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1372,6 +1385,8 @@ impl S3 for FS { .. } = req.input.clone(); + validate_table_catalog_object_mutation(&bucket, &object).await?; + crate::storage::s3_api::tagging::validate_object_tag_set(&tagging.tag_set)?; let Some(store) = new_object_layer_fn() else { diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs new file mode 100644 index 000000000..5b6275bd5 --- /dev/null +++ b/rustfs/src/table_catalog.rs @@ -0,0 +1,859 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Internal table catalog primitives for the Iceberg REST Catalog framework. +//! +//! This module intentionally does not expose HTTP handlers or mutate existing +//! S3 object behavior. It defines the stable internal boundary that later +//! catalog routes and object guards can share. + +#![allow(dead_code)] + +use std::fmt; + +use rustfs_ecstore::bucket::{ + metadata::{BUCKET_TABLE_CONFIG, BUCKET_TABLE_RESERVED_PREFIX}, + metadata_sys, +}; +use serde::{Deserialize, Serialize}; + +pub(crate) const TABLE_BUCKET_MARKER_CONFIG: &str = BUCKET_TABLE_CONFIG; +pub(crate) const RESERVED_CATALOG_OBJECT_MESSAGE: &str = "Object key is reserved for the table catalog"; +pub(crate) const TABLE_BUCKET_CATALOG_TYPE: &str = "iceberg-rest"; +pub(crate) const TABLE_BUCKET_CONFIG_VERSION: u16 = 1; +pub(crate) const DEFAULT_WAREHOUSE_ID: &str = "default"; +pub(crate) const TABLE_NAMESPACE_MARKER_VERSION: u16 = 1; +pub(crate) const TABLE_RESOURCE_MARKER_VERSION: u16 = 1; +pub(crate) const TABLE_METADATA_POINTER_VERSION: u16 = 1; +pub(crate) const TABLE_METADATA_FILE_NAME_MAX_LEN: usize = 128; +pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX; +const WAREHOUSE_ROOT: &str = "warehouses"; +const NAMESPACE_ROOT: &str = "namespaces"; +const TABLE_ROOT: &str = "tables"; +const NAMESPACE_MARKER_FILE: &str = "namespace.json"; +const TABLE_MARKER_FILE: &str = "table.json"; +const CURRENT_POINTER_FILE: &str = "current.json"; +const LIFECYCLE_FILE: &str = "lifecycle.json"; +const METADATA_DIR: &str = "metadata"; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CatalogIdentifierError { + Empty, + TooLong { max: usize }, + InvalidCharacter, + InvalidBoundary, + Ambiguous, +} + +impl fmt::Display for CatalogIdentifierError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Empty => f.write_str("catalog identifier segment is empty"), + Self::TooLong { max } => write!(f, "catalog identifier segment exceeds {max} characters"), + Self::InvalidCharacter => f.write_str("catalog identifier segment contains invalid characters"), + Self::InvalidBoundary => { + f.write_str("catalog identifier segment must start and end with a lowercase letter or digit") + } + Self::Ambiguous => f.write_str("catalog identifier segment is ambiguous"), + } + } +} + +impl std::error::Error for CatalogIdentifierError {} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TableObjectMutationError { + ReservedCatalogObject, +} + +impl fmt::Display for TableObjectMutationError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ReservedCatalogObject => f.write_str("object key is reserved for the table catalog"), + } + } +} + +impl std::error::Error for TableObjectMutationError {} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct TableBucketMarker { + pub version: u16, + pub catalog_type: &'static str, + pub reserved_prefix: &'static str, +} + +impl Default for TableBucketMarker { + fn default() -> Self { + Self { + version: TABLE_BUCKET_CONFIG_VERSION, + catalog_type: TABLE_BUCKET_CATALOG_TYPE, + reserved_prefix: TABLE_RESERVED_PREFIX, + } + } +} + +pub(crate) fn table_bucket_marker_json() -> Result, serde_json::Error> { + serde_json::to_vec(&TableBucketMarker::default()) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct NamespaceMarker { + pub version: u16, + pub namespace: String, +} + +impl NamespaceMarker { + pub fn new(namespace: &Namespace) -> Self { + Self { + version: TABLE_NAMESPACE_MARKER_VERSION, + namespace: namespace.public_name(), + } + } +} + +pub(crate) fn namespace_marker_json(namespace: &Namespace) -> Result, serde_json::Error> { + serde_json::to_vec(&NamespaceMarker::new(namespace)) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct TableMarker { + pub version: u16, + pub namespace: String, + pub name: String, + pub metadata_location: Option, +} + +impl TableMarker { + pub fn new(namespace: &Namespace, table: &IdentifierSegment) -> Self { + Self { + version: TABLE_RESOURCE_MARKER_VERSION, + namespace: namespace.public_name(), + name: table.as_str().to_string(), + metadata_location: None, + } + } +} + +pub(crate) fn table_marker_json(namespace: &Namespace, table: &IdentifierSegment) -> Result, serde_json::Error> { + serde_json::to_vec(&TableMarker::new(namespace, table)) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct TableMetadataPointer { + pub version: u16, + pub metadata_location: String, +} + +impl TableMetadataPointer { + pub fn new(metadata_location: String) -> Self { + Self { + version: TABLE_METADATA_POINTER_VERSION, + metadata_location, + } + } +} + +pub(crate) fn table_metadata_pointer_json(metadata_location: String) -> Result, serde_json::Error> { + serde_json::to_vec(&TableMetadataPointer::new(metadata_location)) +} + +pub(crate) fn parse_table_metadata_pointer(data: &[u8]) -> Result { + serde_json::from_slice(data) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct IdentifierSegment(String); + +impl IdentifierSegment { + pub const MAX_LEN: usize = 64; + + pub fn parse(value: impl Into) -> Result { + let value = value.into(); + validate_identifier_segment(&value)?; + Ok(Self(value)) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Namespace { + segments: Vec, +} + +impl Namespace { + pub fn parse(value: &str) -> Result { + if value.is_empty() { + return Err(CatalogIdentifierError::Empty); + } + + let mut segments = Vec::new(); + for segment in value.split('.') { + segments.push(IdentifierSegment::parse(segment.to_string())?); + } + + Ok(Self { segments }) + } + + pub fn segments(&self) -> &[IdentifierSegment] { + &self.segments + } + + pub fn storage_id(&self) -> String { + self.segments + .iter() + .map(IdentifierSegment::as_str) + .collect::>() + .join("/") + } + + pub fn public_name(&self) -> String { + self.segments + .iter() + .map(IdentifierSegment::as_str) + .collect::>() + .join(".") + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TableIdentifier { + warehouse: IdentifierSegment, + namespace: Namespace, + name: IdentifierSegment, +} + +impl TableIdentifier { + pub fn new(warehouse: IdentifierSegment, namespace: Namespace, name: IdentifierSegment) -> Self { + Self { + warehouse, + namespace, + name, + } + } + + pub fn warehouse(&self) -> &IdentifierSegment { + &self.warehouse + } + + pub fn namespace(&self) -> &Namespace { + &self.namespace + } + + pub fn name(&self) -> &IdentifierSegment { + &self.name + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TablePathResolver { + reserved_prefix: &'static str, +} + +impl Default for TablePathResolver { + fn default() -> Self { + Self { + reserved_prefix: TABLE_RESERVED_PREFIX, + } + } +} + +impl TablePathResolver { + pub fn current_pointer_path(&self, table: &TableIdentifier) -> String { + format!("{}/{}", self.table_root(table), CURRENT_POINTER_FILE) + } + + pub fn metadata_dir_path(&self, table: &TableIdentifier) -> String { + format!("{}/{}", self.table_root(table), METADATA_DIR) + } + + pub fn metadata_file_path(&self, table: &TableIdentifier, metadata_file_name: &str) -> String { + format!("{}/{}", self.metadata_dir_path(table), metadata_file_name) + } + + fn table_root(&self, table: &TableIdentifier) -> String { + format!( + "{}/{}/{}/{}/{}/{}/{}", + self.reserved_prefix, + WAREHOUSE_ROOT, + table.warehouse().as_str(), + NAMESPACE_ROOT, + table.namespace().storage_id(), + TABLE_ROOT, + table.name().as_str() + ) + } +} + +pub fn is_reserved_table_object_key(object_key: &str) -> bool { + object_key == TABLE_RESERVED_PREFIX + || object_key + .strip_prefix(TABLE_RESERVED_PREFIX) + .is_some_and(|rest| rest.starts_with('/')) +} + +pub(crate) fn default_namespace_root_prefix() -> String { + format!( + "{}/{}/{}/{}/", + TABLE_RESERVED_PREFIX, WAREHOUSE_ROOT, DEFAULT_WAREHOUSE_ID, NAMESPACE_ROOT + ) +} + +pub(crate) fn default_namespace_marker_path(namespace: &Namespace) -> String { + format!("{}{}/{}", default_namespace_root_prefix(), namespace.storage_id(), NAMESPACE_MARKER_FILE) +} + +pub(crate) fn default_table_root_prefix(namespace: &Namespace) -> String { + format!("{}{}/{}/", default_namespace_root_prefix(), namespace.storage_id(), TABLE_ROOT) +} + +pub(crate) fn default_table_marker_path(namespace: &Namespace, table: &IdentifierSegment) -> String { + format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), TABLE_MARKER_FILE) +} + +pub(crate) fn default_table_metadata_dir_path(namespace: &Namespace, table: &IdentifierSegment) -> String { + format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), METADATA_DIR) +} + +pub(crate) fn default_table_metadata_file_path( + namespace: &Namespace, + table: &IdentifierSegment, + metadata_file_name: &str, +) -> String { + format!("{}/{}", default_table_metadata_dir_path(namespace, table), metadata_file_name) +} + +pub(crate) fn default_table_current_pointer_path(namespace: &Namespace, table: &IdentifierSegment) -> String { + format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), CURRENT_POINTER_FILE) +} + +pub(crate) fn default_table_lifecycle_path(namespace: &Namespace, table: &IdentifierSegment) -> String { + format!("{}{}/{}", default_table_root_prefix(namespace), table.as_str(), LIFECYCLE_FILE) +} + +pub(crate) fn namespace_name_from_marker_path(object_key: &str) -> Option { + let prefix = default_namespace_root_prefix(); + let suffix = format!("/{NAMESPACE_MARKER_FILE}"); + + object_key + .strip_prefix(prefix.as_str()) + .and_then(|value| value.strip_suffix(suffix.as_str())) + .filter(|value| !value.is_empty()) + .map(|value| value.replace('/', ".")) +} + +pub(crate) fn table_name_from_marker_path(namespace: &Namespace, object_key: &str) -> Option { + let prefix = default_table_root_prefix(namespace); + let suffix = format!("/{TABLE_MARKER_FILE}"); + + object_key + .strip_prefix(prefix.as_str()) + .and_then(|value| value.strip_suffix(suffix.as_str())) + .filter(|value| !value.is_empty() && !value.contains('/')) + .map(ToString::to_string) +} + +pub(crate) fn metadata_location_from_metadata_file_path( + namespace: &Namespace, + table: &IdentifierSegment, + object_key: &str, +) -> Option { + let prefix = format!("{}/", default_table_metadata_dir_path(namespace, table)); + + object_key + .strip_prefix(prefix.as_str()) + .filter(|value| is_valid_table_metadata_file_name(value)) + .map(|_| object_key.to_string()) +} + +pub(crate) fn is_valid_table_metadata_location( + namespace: &Namespace, + table: &IdentifierSegment, + metadata_location: &str, +) -> bool { + if metadata_location.is_empty() { + return false; + } + + let metadata_prefix = format!("{}/", default_table_metadata_dir_path(namespace, table)); + metadata_location + .strip_prefix(&metadata_prefix) + .is_some_and(is_valid_table_metadata_file_name) +} + +pub(crate) fn is_valid_table_metadata_file_name(metadata_file_name: &str) -> bool { + if metadata_file_name.is_empty() + || metadata_file_name.len() > TABLE_METADATA_FILE_NAME_MAX_LEN + || !metadata_file_name.ends_with(".json") + || metadata_file_name.contains("..") + || metadata_file_name.contains('%') + || metadata_file_name.contains('/') + || metadata_file_name.contains('\\') + || metadata_file_name.bytes().any(|byte| byte.is_ascii_control()) + { + return false; + } + + let bytes = metadata_file_name.as_bytes(); + if !is_lower_ascii_alnum(bytes[0]) || !is_lower_ascii_alnum(bytes[bytes.len() - 1]) { + return false; + } + + bytes + .iter() + .all(|byte| is_lower_ascii_alnum(*byte) || matches!(*byte, b'.' | b'_' | b'-')) +} + +pub fn validate_object_mutation(table_bucket_enabled: bool, object_key: &str) -> Result<(), TableObjectMutationError> { + if table_bucket_enabled && is_reserved_table_object_key(object_key) { + return Err(TableObjectMutationError::ReservedCatalogObject); + } + + Ok(()) +} + +pub(crate) async fn validate_bucket_object_mutation(bucket: &str, object_key: &str) -> Result<(), TableObjectMutationError> { + let table_bucket_enabled = metadata_sys::get(bucket) + .await + .is_ok_and(|metadata| metadata.table_bucket_enabled()); + + validate_object_mutation(table_bucket_enabled, object_key) +} + +fn validate_identifier_segment(value: &str) -> Result<(), CatalogIdentifierError> { + if value.is_empty() { + return Err(CatalogIdentifierError::Empty); + } + + if value.len() > IdentifierSegment::MAX_LEN { + return Err(CatalogIdentifierError::TooLong { + max: IdentifierSegment::MAX_LEN, + }); + } + + if matches!(value, "." | "..") || value.contains('%') || value.contains('/') || value.contains('\\') { + return Err(CatalogIdentifierError::Ambiguous); + } + + let bytes = value.as_bytes(); + if !is_lower_ascii_alnum(bytes[0]) || !is_lower_ascii_alnum(bytes[bytes.len() - 1]) { + return Err(CatalogIdentifierError::InvalidBoundary); + } + + if bytes + .iter() + .any(|byte| !is_lower_ascii_alnum(*byte) && !matches!(*byte, b'_' | b'-')) + { + return Err(CatalogIdentifierError::InvalidCharacter); + } + + Ok(()) +} + +fn is_lower_ascii_alnum(value: u8) -> bool { + value.is_ascii_lowercase() || value.is_ascii_digit() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reserved_table_object_key_matches_exact_prefix_and_children_only() { + assert!(is_reserved_table_object_key(".rustfs-table")); + assert!(is_reserved_table_object_key(".rustfs-table/")); + assert!(is_reserved_table_object_key(".rustfs-table/metadata/current.json")); + + assert!(!is_reserved_table_object_key("")); + assert!(!is_reserved_table_object_key(".rustfs-table-other")); + assert!(!is_reserved_table_object_key("prefix/.rustfs-table/object")); + assert!(!is_reserved_table_object_key("user/.rustfs-table")); + } + + #[test] + fn object_mutation_guard_only_blocks_reserved_prefix_for_table_buckets() { + assert!(validate_object_mutation(false, ".rustfs-table/current.json").is_ok()); + assert_eq!( + validate_object_mutation(true, ".rustfs-table/current.json").unwrap_err(), + TableObjectMutationError::ReservedCatalogObject + ); + assert!(validate_object_mutation(true, ".rustfs-table-other/current.json").is_ok()); + } + + #[tokio::test] + async fn bucket_object_mutation_guard_allows_when_bucket_metadata_is_unavailable() { + assert!( + validate_bucket_object_mutation("missing-bucket", ".rustfs-table/current.json") + .await + .is_ok() + ); + } + + #[test] + fn table_bucket_marker_json_uses_stable_catalog_defaults() { + let marker = serde_json::to_value(TableBucketMarker::default()).unwrap(); + + assert_eq!(marker["version"], TABLE_BUCKET_CONFIG_VERSION); + assert_eq!(marker["catalog_type"], TABLE_BUCKET_CATALOG_TYPE); + assert_eq!(marker["reserved_prefix"], TABLE_RESERVED_PREFIX); + assert!(!table_bucket_marker_json().unwrap().is_empty()); + } + + #[test] + fn namespace_marker_path_stays_under_default_reserved_boundary() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + + assert_eq!( + default_namespace_marker_path(&namespace), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/namespace.json" + ); + assert_eq!(namespace.public_name(), "analytics.daily_events"); + } + + #[test] + fn namespace_marker_path_extracts_public_name() { + assert_eq!( + namespace_name_from_marker_path(".rustfs-table/warehouses/default/namespaces/analytics/daily_events/namespace.json"), + Some("analytics.daily_events".to_string()) + ); + assert_eq!( + namespace_name_from_marker_path( + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/current.json" + ), + None + ); + assert_eq!( + namespace_name_from_marker_path(".rustfs-table/warehouses/other/namespaces/analytics/daily_events/namespace.json"), + None + ); + } + + #[test] + fn namespace_marker_json_uses_stable_catalog_defaults() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + let marker = serde_json::to_value(NamespaceMarker::new(&namespace)).unwrap(); + + assert_eq!(marker["version"], TABLE_NAMESPACE_MARKER_VERSION); + assert_eq!(marker["namespace"], "analytics.daily_events"); + assert!(!namespace_marker_json(&namespace).unwrap().is_empty()); + } + + #[test] + fn table_marker_path_stays_under_namespace_reserved_boundary() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + let table = IdentifierSegment::parse("events").unwrap(); + + assert_eq!( + default_table_root_prefix(&namespace), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/" + ); + assert_eq!( + default_table_marker_path(&namespace, &table), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/table.json" + ); + } + + #[test] + fn table_marker_path_extracts_table_name() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + + assert_eq!( + table_name_from_marker_path( + &namespace, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/table.json" + ), + Some("events".to_string()) + ); + assert_eq!( + table_name_from_marker_path( + &namespace, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/current.json" + ), + None + ); + assert_eq!( + table_name_from_marker_path( + &namespace, + ".rustfs-table/warehouses/default/namespaces/analytics/other/tables/events/table.json" + ), + None + ); + } + + #[test] + fn table_marker_json_uses_stable_catalog_defaults() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + let table = IdentifierSegment::parse("events").unwrap(); + let marker = serde_json::to_value(TableMarker::new(&namespace, &table)).unwrap(); + + assert_eq!(marker["version"], TABLE_RESOURCE_MARKER_VERSION); + assert_eq!(marker["namespace"], "analytics.daily_events"); + assert_eq!(marker["name"], "events"); + assert!(marker["metadata_location"].is_null()); + assert!(!table_marker_json(&namespace, &table).unwrap().is_empty()); + } + + #[test] + fn table_current_pointer_path_stays_under_table_boundary() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + let table = IdentifierSegment::parse("events").unwrap(); + + assert_eq!( + default_table_metadata_dir_path(&namespace, &table), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata" + ); + assert_eq!( + default_table_current_pointer_path(&namespace, &table), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/current.json" + ); + assert_eq!( + default_table_lifecycle_path(&namespace, &table), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/lifecycle.json" + ); + } + + #[test] + fn table_metadata_file_path_stays_under_metadata_boundary() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + let table = IdentifierSegment::parse("events").unwrap(); + let table_identifier = + TableIdentifier::new(IdentifierSegment::parse(DEFAULT_WAREHOUSE_ID).unwrap(), namespace.clone(), table.clone()); + + assert_eq!( + default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.metadata.json" + ); + assert_eq!( + TablePathResolver::default().metadata_file_path(&table_identifier, "00001.metadata.json"), + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.metadata.json" + ); + } + + #[test] + fn table_metadata_file_name_validation_rejects_unsafe_names() { + assert!(is_valid_table_metadata_file_name("00001.metadata.json")); + assert!(is_valid_table_metadata_file_name("v1-4f2c_metadata.json")); + + assert!(!is_valid_table_metadata_file_name("")); + assert!(!is_valid_table_metadata_file_name(".metadata.json")); + assert!(!is_valid_table_metadata_file_name("00001.metadata")); + assert!(!is_valid_table_metadata_file_name("00001.JSON")); + assert!(!is_valid_table_metadata_file_name("../current.json")); + assert!(!is_valid_table_metadata_file_name("nested/00001.json")); + assert!(!is_valid_table_metadata_file_name("nested%2f00001.json")); + assert!(!is_valid_table_metadata_file_name("00001\\metadata.json")); + assert!(!is_valid_table_metadata_file_name("00001\nmetadata.json")); + } + + #[test] + fn table_metadata_location_validation_stays_inside_metadata_dir() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + let table = IdentifierSegment::parse("events").unwrap(); + + assert!(is_valid_table_metadata_location( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.metadata.json" + )); + assert!(!is_valid_table_metadata_location(&namespace, &table, "")); + assert!(!is_valid_table_metadata_location( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/current.json" + )); + assert!(!is_valid_table_metadata_location( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/other/metadata/00001.json" + )); + assert!(!is_valid_table_metadata_location( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/../current.json" + )); + assert!(!is_valid_table_metadata_location( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/nested/00001.json" + )); + } + + #[test] + fn metadata_location_from_metadata_file_path_extracts_table_metadata_only() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + let table = IdentifierSegment::parse("events").unwrap(); + + assert_eq!( + metadata_location_from_metadata_file_path( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.metadata.json" + ), + Some( + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.metadata.json" + .to_string() + ) + ); + assert_eq!( + metadata_location_from_metadata_file_path( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/current.json" + ), + None + ); + assert_eq!( + metadata_location_from_metadata_file_path( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/nested/00001.metadata.json" + ), + None + ); + assert_eq!( + metadata_location_from_metadata_file_path( + &namespace, + &table, + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/other/metadata/00001.metadata.json" + ), + None + ); + } + + #[test] + fn table_metadata_pointer_json_round_trips() { + let location = + ".rustfs-table/warehouses/default/namespaces/analytics/daily_events/tables/events/metadata/00001.json".to_string(); + let data = table_metadata_pointer_json(location.clone()).unwrap(); + let pointer = parse_table_metadata_pointer(&data).unwrap(); + + assert_eq!(pointer.version, TABLE_METADATA_POINTER_VERSION); + assert_eq!(pointer.metadata_location, location); + } + + #[test] + fn object_mutation_entrypoints_call_reserved_prefix_guard() { + let source = include_str!("app/object_usecase.rs"); + for expected in [ + "validate_object_key(&key, request_method_name)?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;", + "validate_object_key(&key, \"COPY (dest)\")?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;", + "if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await", + "validate_object_key(&key, \"DELETE\")?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;", + "validate_table_catalog_object_mutation(&bucket, &object).await?;", + "validate_object_key(&key, \"PUT\")?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;", + "validate_table_catalog_object_mutation(&bucket, &fpath).await?;", + ] { + assert!(source.contains(expected), "missing object mutation guard: {expected}"); + } + } + + #[test] + fn multipart_mutation_entrypoints_call_reserved_prefix_guard() { + let source = include_str!("app/multipart_usecase.rs"); + + assert_eq!( + source + .matches("validate_table_catalog_object_mutation(&bucket, &key).await?;") + .count(), + 4 + ); + } + + #[test] + fn object_metadata_mutation_entrypoints_call_reserved_prefix_guard() { + let source = include_str!("storage/ecfs.rs"); + + assert_eq!( + source + .matches("validate_table_catalog_object_mutation(&bucket, &object).await?;") + .count(), + 2 + ); + assert_eq!( + source + .matches("validate_table_catalog_object_mutation(&bucket, &key).await?;") + .count(), + 2 + ); + } + + #[test] + fn identifier_segment_accepts_conservative_catalog_names() { + for value in [ + "a", + "a1", + "a-b", + "a_b", + "abc123", + "a23456789012345678901234567890123456789012345678901234567890123", + ] { + assert_eq!(IdentifierSegment::parse(value).unwrap().as_str(), value); + } + } + + #[test] + fn identifier_segment_rejects_ambiguous_or_unsafe_names() { + for value in [ + "", + ".", + "..", + "Upper", + "has.dot", + "has/slash", + "has\\slash", + "has%2fslash", + "-leading", + "trailing-", + "_leading", + "trailing_", + "has space", + "name\nbreak", + ] { + assert!(IdentifierSegment::parse(value).is_err(), "value should be rejected: {value:?}"); + } + + let too_long = "a".repeat(IdentifierSegment::MAX_LEN + 1); + assert!(IdentifierSegment::parse(too_long).is_err()); + } + + #[test] + fn namespace_uses_dot_syntax_for_public_identity_and_slash_for_storage() { + let namespace = Namespace::parse("analytics.daily_events").unwrap(); + + assert_eq!(namespace.segments().len(), 2); + assert_eq!(namespace.storage_id(), "analytics/daily_events"); + } + + #[test] + fn resolver_builds_paths_under_reserved_table_boundary() { + let table = TableIdentifier::new( + IdentifierSegment::parse("warehouse1").unwrap(), + Namespace::parse("analytics.daily").unwrap(), + IdentifierSegment::parse("events").unwrap(), + ); + let resolver = TablePathResolver::default(); + + assert_eq!( + resolver.current_pointer_path(&table), + ".rustfs-table/warehouses/warehouse1/namespaces/analytics/daily/tables/events/current.json" + ); + assert_eq!( + resolver.metadata_dir_path(&table), + ".rustfs-table/warehouses/warehouse1/namespaces/analytics/daily/tables/events/metadata" + ); + } +}