From 73e534682e93d1ed2c7b0e752ad550d969ca96b2 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 15 Jun 2026 07:06:29 +0800 Subject: [PATCH] feat(table-catalog): add catalog recovery diagnostics (#3444) * feat(table-catalog): add commit recovery observability * feat(table-catalog): harden catalog recovery diagnostics * test(table-catalog): cover recovery admin route policy * test(table-catalog): update recovery route count * fix(table-catalog): satisfy duration clippy lint * fix(obs): promote span request id field * Update local.rs Signed-off-by: houseme --------- Signed-off-by: houseme Co-authored-by: Henry Guo Co-authored-by: houseme --- rustfs/src/admin/handlers/table_catalog.rs | 242 ++++- rustfs/src/admin/route_policy.rs | 24 +- rustfs/src/admin/route_registration_test.rs | 20 + rustfs/src/table_catalog.rs | 959 +++++++++++++++++++- 4 files changed, 1162 insertions(+), 83 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 51259cada..c6bb7e371 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -23,6 +23,7 @@ use crate::table_catalog::{DEFAULT_WAREHOUSE_ID, TableCatalogStore}; use http::{HeaderMap, HeaderValue, StatusCode}; use hyper::Method; use matchit::Params; +use metrics::{counter, histogram}; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_ecstore::{ bucket::{metadata::table_catalog_path_hash, metadata_sys}, @@ -39,6 +40,7 @@ use rustfs_policy::{ use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use std::collections::{BTreeMap, HashMap}; +use std::time::{Duration as StdDuration, Instant}; use time::{Duration, OffsetDateTime}; use uuid::Uuid; @@ -69,6 +71,7 @@ const S3_SECRET_ACCESS_KEY_CONFIG_KEY: &str = "s3.secret-access-key"; const S3_SESSION_TOKEN_CONFIG_KEY: &str = "s3.session-token"; const TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT: &str = "namespaces"; const TABLE_CATALOG_TABLE_RESOURCE_ROOT: &str = "tables"; +const TABLE_CATALOG_ADMIN_OPERATION_SLOW_LOG_THRESHOLD: StdDuration = StdDuration::from_secs(2); const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "GET /v1/{prefix}/namespaces", "POST /v1/{prefix}/namespaces", @@ -107,6 +110,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export", "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", "GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", + "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", ]; @@ -135,6 +139,7 @@ static GET_TABLE_MAINTENANCE_JOB_HANDLER: GetTableMaintenanceJobHandler = GetTab static EXPORT_TABLE_CATALOG_HANDLER: ExportTableCatalogHandler = ExportTableCatalogHandler {}; static IMPORT_TABLE_CATALOG_HANDLER: ImportTableCatalogHandler = ImportTableCatalogHandler {}; static GET_TABLE_CATALOG_DIAGNOSTICS_HANDLER: GetTableCatalogDiagnosticsHandler = GetTableCatalogDiagnosticsHandler {}; +static RECOVER_TABLE_CATALOG_HANDLER: RecoverTableCatalogHandler = RecoverTableCatalogHandler {}; static ROLLBACK_TABLE_CATALOG_HANDLER: RollbackTableCatalogHandler = RollbackTableCatalogHandler {}; #[derive(Debug, Serialize)] @@ -606,6 +611,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/diagnostics").as_str(), AdminOperation(&GET_TABLE_CATALOG_DIAGNOSTICS_HANDLER), )?; + r.insert( + Method::POST, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/recovery").as_str(), + AdminOperation(&RECOVER_TABLE_CATALOG_HANDLER), + )?; r.insert( Method::POST, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/rollback").as_str(), @@ -638,6 +648,59 @@ fn empty_response(status: StatusCode) -> S3Response<(StatusCode, Body)> { S3Response::new((status, Body::default())) } +fn duration_millis_u64(duration: StdDuration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +fn table_catalog_admin_operation_result_label(result: &Result) -> &'static str { + if result.is_ok() { "success" } else { "failure" } +} + +fn record_table_catalog_admin_operation_result( + operation: &str, + warehouse: &str, + namespace: &str, + table: &str, + started: Instant, + result: &Result, +) { + let elapsed = started.elapsed(); + let result_label = table_catalog_admin_operation_result_label(result); + counter!( + "rustfs_table_catalog_admin_operations_total", + "operation" => operation.to_string(), + "result" => result_label.to_string() + ) + .increment(1); + histogram!( + "rustfs_table_catalog_admin_operation_duration_seconds", + "operation" => operation.to_string(), + "result" => result_label.to_string() + ) + .record(elapsed.as_secs_f64()); + + if result.is_err() { + tracing::warn!( + operation, + warehouse, + namespace, + table, + result = result_label, + duration_ms = duration_millis_u64(elapsed), + "table catalog admin operation failed" + ); + } else if elapsed >= TABLE_CATALOG_ADMIN_OPERATION_SLOW_LOG_THRESHOLD { + tracing::warn!( + operation, + warehouse, + namespace, + table, + duration_ms = duration_millis_u64(elapsed), + "slow table catalog admin operation" + ); + } +} + fn exists_status(exists: bool) -> StatusCode { if exists { StatusCode::NO_CONTENT @@ -2436,13 +2499,35 @@ async fn catalog_import_response( where B: crate::table_catalog::TableCatalogObjectBackend, { - ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?; - let mut entry = table_entry_from_import_request(bucket, namespace, table, request)?; - let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; - validate_metadata_table_location_in_bucket(bucket, &metadata)?; - adopt_registered_metadata_identity(&mut entry, &metadata)?; - store.register_table(entry.clone()).await.map_err(catalog_store_error)?; - Ok(load_table_response_from_entry(entry, metadata)) + let started = Instant::now(); + let result = async { + ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?; + let mut entry = table_entry_from_import_request(bucket, namespace, table, request)?; + let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; + validate_metadata_table_location_in_bucket(bucket, &metadata)?; + adopt_registered_metadata_identity(&mut entry, &metadata)?; + if let Some(existing) = store + .load_table(bucket, &namespace.public_name(), table) + .await + .map_err(catalog_store_error)? + { + if existing.table_uuid == entry.table_uuid + && existing.metadata_location == entry.metadata_location + && existing.warehouse_location == entry.warehouse_location + { + return Ok(load_table_response_from_entry(existing, metadata)); + } + return Err(s3_error!( + PreconditionFailed, + "catalog import target already exists with different table identity or metadata pointer" + )); + } + store.register_table(entry.clone()).await.map_err(catalog_store_error)?; + Ok(load_table_response_from_entry(entry, metadata)) + } + .await; + record_table_catalog_admin_operation_result("import", bucket, &namespace.public_name(), table, started, &result); + result } async fn rollback_table_response( @@ -2456,38 +2541,44 @@ async fn rollback_table_response( where S: crate::table_catalog::TableCatalogStore + ?Sized, { - let Some(current) = store - .load_table(bucket, &namespace.public_name(), table) - .await - .map_err(catalog_store_error)? - else { - return Err(s3_error!(InvalidRequest, "table not found")); - }; - let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string()) - .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?; - if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &request.metadata_location) { - return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory")); + let started = Instant::now(); + let result = async { + let Some(current) = store + .load_table(bucket, &namespace.public_name(), table) + .await + .map_err(catalog_store_error)? + else { + return Err(s3_error!(InvalidRequest, "table not found")); + }; + let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string()) + .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?; + if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &request.metadata_location) { + return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory")); + } + let current_metadata = read_table_metadata_json(metadata_backend, bucket, ¤t.metadata_location).await?; + validate_metadata_table_location_in_bucket(bucket, ¤t_metadata)?; + let target_metadata = read_table_metadata_json(metadata_backend, bucket, &request.metadata_location).await?; + validate_metadata_table_location_in_bucket(bucket, &target_metadata)?; + validate_metadata_matches_current_metadata(¤t_metadata, &target_metadata)?; + let commit_request = crate::table_catalog::TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.to_string(), + commit_id: request.commit_id.unwrap_or_else(|| Uuid::new_v4().to_string()), + idempotency_key: request.idempotency_key, + operation: "rollback".to_string(), + expected_version_token: request.version_token, + expected_metadata_location: current.metadata_location, + new_metadata_location: request.metadata_location, + requirements: Vec::new(), + writer: Some("rustfs-catalog-rollback-api".to_string()), + }; + let result = store.commit_table(commit_request).await.map_err(catalog_store_error)?; + Ok(commit_table_response_from_result(result, target_metadata)) } - let current_metadata = read_table_metadata_json(metadata_backend, bucket, ¤t.metadata_location).await?; - validate_metadata_table_location_in_bucket(bucket, ¤t_metadata)?; - let target_metadata = read_table_metadata_json(metadata_backend, bucket, &request.metadata_location).await?; - validate_metadata_table_location_in_bucket(bucket, &target_metadata)?; - validate_metadata_matches_current_metadata(¤t_metadata, &target_metadata)?; - let commit_request = crate::table_catalog::TableCommitRequest { - table_bucket: bucket.to_string(), - namespace: namespace.public_name(), - table: table.to_string(), - commit_id: request.commit_id.unwrap_or_else(|| Uuid::new_v4().to_string()), - idempotency_key: request.idempotency_key, - operation: "rollback".to_string(), - expected_version_token: request.version_token, - expected_metadata_location: current.metadata_location, - new_metadata_location: request.metadata_location, - requirements: Vec::new(), - writer: Some("rustfs-catalog-rollback-api".to_string()), - }; - let result = store.commit_table(commit_request).await.map_err(catalog_store_error)?; - Ok(commit_table_response_from_result(result, target_metadata)) + .await; + record_table_catalog_admin_operation_result("rollback", bucket, &namespace.public_name(), table, started, &result); + result } pub struct GetCatalogConfigHandler {} @@ -2866,10 +2957,13 @@ impl Operation for ExportTableCatalogHandler { let resource = TableCatalogResource::table(&warehouse, &namespace, &table); authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; let store = table_catalog_store()?; - let response = store + let started = Instant::now(); + let result = store .export_table_catalog_entry(&warehouse, &namespace.public_name(), &table) .await - .map_err(catalog_store_error)?; + .map_err(catalog_store_error); + record_table_catalog_admin_operation_result("export", &warehouse, &namespace.public_name(), &table, started, &result); + let response = result?; build_json_response(StatusCode::OK, &response) } } @@ -2910,10 +3004,42 @@ impl Operation for GetTableCatalogDiagnosticsHandler { .get_table_maintenance_config(&warehouse, &namespace.public_name(), &table) .await .map_err(catalog_store_error)?; - let response = store + let started = Instant::now(); + let result = store .diagnose_table_catalog(&warehouse, &namespace.public_name(), &table, config.retain_recent_metadata_files) .await - .map_err(catalog_store_error)?; + .map_err(catalog_store_error); + record_table_catalog_admin_operation_result( + "diagnostics", + &warehouse, + &namespace.public_name(), + &table, + started, + &result, + ); + let response = result?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct RecoverTableCatalogHandler {} + +#[async_trait::async_trait] +impl Operation for RecoverTableCatalogHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?; + let store = table_catalog_store()?; + let started = Instant::now(); + let result = store + .recover_table_commits(&warehouse, &namespace.public_name(), &table) + .await + .map_err(catalog_store_error); + record_table_catalog_admin_operation_result("recovery", &warehouse, &namespace.public_name(), &table, started, &result); + let response = result?; build_json_response(StatusCode::OK, &response) } } @@ -3003,6 +3129,20 @@ mod tests { .endpoints .contains(&"GET /{warehouse}/namespaces/{namespace}/tables/{table}/credentials") ); + assert!( + response + .endpoints + .contains(&"POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery") + ); + } + + #[test] + fn table_catalog_admin_operation_result_labels_are_stable() { + let success: Result<(), ()> = Ok(()); + let failure: Result<(), ()> = Err(()); + + assert_eq!(table_catalog_admin_operation_result_label(&success), "success"); + assert_eq!(table_catalog_admin_operation_result_label(&failure), "failure"); } #[test] @@ -3043,6 +3183,7 @@ mod tests { ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), ("GetTableCatalogDiagnosticsHandler", "AdminAction::GetTableMetadataAction"), + ("RecoverTableCatalogHandler", "AdminAction::CommitTableAction"), ("RollbackTableCatalogHandler", "AdminAction::CommitTableAction"), ] { let block = operation_block(src, handler); @@ -3075,6 +3216,7 @@ mod tests { ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), ("GetTableCatalogDiagnosticsHandler", "AdminAction::GetTableMetadataAction"), + ("RecoverTableCatalogHandler", "AdminAction::CommitTableAction"), ("RollbackTableCatalogHandler", "AdminAction::CommitTableAction"), ] { let block = operation_block(src, handler); @@ -5590,6 +5732,22 @@ mod tests { .expect("table should exist"); assert_eq!(current.properties.get("owner").map(String::as_str), Some("lakehouse")); + let imported_again = catalog_import_response( + &store, + &backend, + bucket, + &namespace, + "events", + CatalogImportRequest { + metadata_location: imported_location.clone(), + properties: BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]), + }, + true, + ) + .await + .expect("repeated catalog import should be idempotent"); + assert_eq!(imported_again.metadata_location, imported_location); + let rollback_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); backend .put_json( diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 9a2fd3827..81344b6b1 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -754,6 +754,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_METADATA, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", + COMMIT_TABLE, + RouteRiskLevel::High, + ), admin( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", @@ -905,6 +911,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_METADATA, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", + COMMIT_TABLE, + RouteRiskLevel::High, + ), admin( HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", @@ -1094,7 +1106,7 @@ mod tests { let table_specs = ADMIN_ROUTE_POLICY_SPECS .iter() .filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1")); - assert_eq!(table_specs.count(), 52); + assert_eq!(table_specs.count(), 54); assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET); assert_action(HttpMethod::Get, "/_iceberg/v1/buckets/{warehouse}", GET_TABLE_BUCKET); assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE); @@ -1183,6 +1195,16 @@ mod tests { "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", REGISTER_TABLE, ); + assert_action( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", + COMMIT_TABLE, + ); + assert_action( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", + COMMIT_TABLE, + ); assert_action( HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index c477b185b..1a3ed20df 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -382,6 +382,11 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", "/analytics/namespaces/sales/tables/orders/catalog/diagnostics", ), + table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", + "/analytics/namespaces/sales/tables/orders/catalog/recovery", + ), table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", @@ -480,6 +485,11 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", "/analytics/namespaces/sales/tables/orders/catalog/diagnostics", ), + compat_table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", + "/analytics/namespaces/sales/tables/orders/catalog/recovery", + ), compat_table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", @@ -657,6 +667,11 @@ fn test_register_routes_cover_representative_admin_paths() { Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/diagnostics"), ); + assert_route( + &router, + Method::POST, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/recovery"), + ); assert_route( &router, Method::POST, @@ -737,6 +752,11 @@ fn test_register_routes_cover_representative_admin_paths() { Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/diagnostics"), ); + assert_route( + &router, + Method::POST, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/recovery"), + ); assert_route( &router, Method::POST, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index e55568afc..d5b24c3c0 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -24,9 +24,11 @@ use std::{ collections::{BTreeMap, BTreeSet}, fmt, sync::Arc, + time::{Duration as StdDuration, Instant}, }; use http::HeaderMap; +use metrics::{counter, histogram}; use rustfs_ecstore::bucket::{ metadata::{ BUCKET_TABLE_CATALOG_META_PREFIX, BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX, BUCKET_TABLE_CONFIG, @@ -77,6 +79,7 @@ const MAINTENANCE_CONFIG_FILE: &str = "config.json"; const MAINTENANCE_JOB_ROOT: &str = "jobs"; const TABLE_CATALOG_LIST_MAX_KEYS: i32 = 1000; const TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS: i64 = 15 * 60; +const TABLE_COMMIT_SLOW_LOG_THRESHOLD: StdDuration = StdDuration::from_secs(2); #[derive(Debug, Clone, PartialEq, Eq)] pub enum CatalogIdentifierError { @@ -329,13 +332,89 @@ pub(crate) enum TableMetadataPointerStatus { InvalidJson, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableCatalogRecoveryStatus { + Healthy, + Recoverable, + ManualReviewRequired, + ReadOnlyRecommended, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableCatalogRecoveryAction { + RunCommitRecovery, + RetryCommit, + RestoreCurrentMetadataObject, + FixCurrentMetadataJson, + MoveCurrentMetadataInsideTable, + ReviewCommitLog, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub(crate) struct TableCatalogDiagnosticsReport { pub catalog: TableCatalogExport, pub current_metadata_status: TableMetadataPointerStatus, + pub recovery_status: TableCatalogRecoveryStatus, + pub recommended_actions: Vec, + pub commit_recovery: TableCommitRecoveryReport, pub orphan_metadata_candidate_locations: Vec, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableCommitRecoveryState { + Committed, + StagedBeforeTableUpdate, + FinalizationRequired, + IdempotencyIndexRepairRequired, + ManualReview, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableCommitIdempotencyIndexStatus { + NotRequired, + Missing, + Matches, + Stale, + Conflicting, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct TableCommitRecoveryEntry { + pub commit_id: String, + pub idempotency_key: Option, + pub operation: String, + pub status: CommitLogStatus, + pub recovery_state: TableCommitRecoveryState, + pub previous_metadata_location: String, + pub new_metadata_location: String, + pub expected_version_token: String, + pub new_version_token: String, + pub idempotency_index_present: bool, + pub idempotency_index_status: TableCommitIdempotencyIndexStatus, + pub reason: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct TableCommitRecoveryReport { + pub table_bucket: String, + pub namespace: String, + pub table: String, + pub table_id: String, + pub current_metadata_location: String, + pub current_version_token: String, + pub current_generation: u64, + pub commits: Vec, + pub staged_before_table_update_count: usize, + pub finalization_required_count: usize, + pub idempotency_repair_required_count: usize, + pub manual_review_count: usize, + pub finalized_count: usize, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum TableCatalogStoreError { NotFound(String), @@ -652,6 +731,15 @@ impl TableCatalogObjectPaths { ) } + pub fn commit_log_entries_prefix(&self, table_bucket: &str, table_id: &str) -> String { + format!( + "{}{}/{}/", + self.table_bucket_root_prefix(table_bucket), + COMMIT_LOG_ROOT, + table_catalog_path_hash(table_id) + ) + } + pub fn commit_idempotency_entry_path(&self, table_bucket: &str, table_id: &str, idempotency_key: &str) -> String { format!( "{}{}/{}/{}.json", @@ -846,6 +934,139 @@ where Ok(()) } + async fn table_commit_recovery_report_for_entry( + &self, + entry: &TableEntry, + finalized_count: usize, + ) -> TableCatalogStoreResult { + let commit_prefix = self.paths.commit_log_entries_prefix(&entry.table_bucket, &entry.table_id); + let mut commits = Vec::new(); + for object in self.backend.list_objects(self.catalog_bucket(), &commit_prefix).await? { + if !object.ends_with(".json") { + continue; + } + let Some(commit_log) = self.read_commit_by_path(&object).await? else { + continue; + }; + let idempotency_commit = match commit_log.idempotency_key.as_deref() { + Some(idempotency_key) => { + let idempotency_path = + self.paths + .commit_idempotency_entry_path(&entry.table_bucket, &entry.table_id, idempotency_key); + self.read_commit_by_path(&idempotency_path).await? + } + None => None, + }; + commits.push(table_commit_recovery_entry(entry, &commit_log, idempotency_commit.as_ref())); + } + commits.sort_by(|left, right| left.commit_id.cmp(&right.commit_id)); + + let finalization_required_count = commits + .iter() + .filter(|commit| matches!(commit.recovery_state, TableCommitRecoveryState::FinalizationRequired)) + .count(); + let idempotency_repair_required_count = commits + .iter() + .filter(|commit| matches!(commit.recovery_state, TableCommitRecoveryState::IdempotencyIndexRepairRequired)) + .count(); + let staged_before_table_update_count = commits + .iter() + .filter(|commit| matches!(commit.recovery_state, TableCommitRecoveryState::StagedBeforeTableUpdate)) + .count(); + let manual_review_count = commits + .iter() + .filter(|commit| matches!(commit.recovery_state, TableCommitRecoveryState::ManualReview)) + .count(); + + Ok(TableCommitRecoveryReport { + table_bucket: entry.table_bucket.clone(), + namespace: entry.namespace.clone(), + table: entry.table.clone(), + table_id: entry.table_id.clone(), + current_metadata_location: entry.metadata_location.clone(), + current_version_token: entry.version_token.clone(), + current_generation: entry.generation, + commits, + staged_before_table_update_count, + finalization_required_count, + idempotency_repair_required_count, + manual_review_count, + finalized_count, + }) + } + + pub(crate) async fn plan_table_commit_recovery( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &table_path).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + self.table_commit_recovery_report_for_entry(&entry, 0).await + } + + pub(crate) async fn recover_table_commits( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); + let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; + let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + + let commit_prefix = self.paths.commit_log_entries_prefix(table_bucket, &entry.table_id); + let mut finalized_count = 0; + for commit_path in self.backend.list_objects(self.catalog_bucket(), &commit_prefix).await? { + if !commit_path.ends_with(".json") { + continue; + } + let Some(commit_log) = self.read_commit_by_path(&commit_path).await? else { + continue; + }; + let idempotency_path = commit_log.idempotency_key.as_deref().map(|idempotency_key| { + self.paths + .commit_idempotency_entry_path(table_bucket, &entry.table_id, idempotency_key) + }); + let idempotency_commit = match idempotency_path.as_deref() { + Some(idempotency_path) => self.read_commit_by_path(idempotency_path).await?, + None => None, + }; + let recovery_entry = table_commit_recovery_entry(&entry, &commit_log, idempotency_commit.as_ref()); + if matches!( + recovery_entry.recovery_state, + TableCommitRecoveryState::FinalizationRequired | TableCommitRecoveryState::IdempotencyIndexRepairRequired + ) { + let mut committed = commit_log; + committed.status = CommitLogStatus::Committed; + self.finalize_commit_log(&commit_path, idempotency_path.as_deref(), &committed) + .await?; + finalized_count += 1; + } + } + + self.table_commit_recovery_report_for_entry(&entry, finalized_count).await + } + pub(crate) async fn get_table_maintenance_config( &self, table_bucket: &str, @@ -1059,9 +1280,15 @@ where .filter(|metadata_location| !retained.contains(metadata_location)) .collect(); + let commit_recovery = self.plan_table_commit_recovery(table_bucket, namespace, table).await?; + let (recovery_status, recommended_actions) = table_catalog_recovery_summary(¤t_metadata_status, &commit_recovery); + Ok(TableCatalogDiagnosticsReport { catalog, current_metadata_status, + recovery_status, + recommended_actions, + commit_recovery, orphan_metadata_candidate_locations, }) } @@ -1405,6 +1632,8 @@ where } async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult { + let commit_started = Instant::now(); + record_table_commit_attempt(&request.operation); let namespace = parse_namespace_for_store(&request.namespace)?; let table = parse_table_for_store(&request.table)?; let table_path = self.paths.table_entry_path(&request.table_bucket, &namespace, &table); @@ -1414,10 +1643,18 @@ where .read_table_with_etag_unlocked(&request.table_bucket, &namespace, &table) .await? else { - return Err(TableCatalogStoreError::NotFound(format!( - "table {}/{}/{}", - request.table_bucket, request.namespace, request.table - ))); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + request.table_bucket, request.namespace, request.table + ))), + ); }; let commit_path = self @@ -1435,10 +1672,18 @@ where if let Some(existing) = existing_commit.as_ref() { if !commit_log_matches_request(existing, &request, ¤t.table_id) { - return Err(TableCatalogStoreError::Conflict(format!( - "commit id already exists: {}", - request.commit_id - ))); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::Conflict(format!( + "commit id already exists: {}", + request.commit_id + ))), + ); } if matches!(existing.status, CommitLogStatus::Committed) || table_matches_committed_log(¤t, existing) { let mut committed = existing.clone(); @@ -1446,52 +1691,116 @@ where let _ = self .finalize_commit_log(&commit_path, idempotency_path.as_deref(), &committed) .await; - return Ok(TableCommitResult { - table: current, - commit_log: committed, - }); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Ok(TableCommitResult { + table: current, + commit_log: committed, + }), + ); } if !matches!(existing.status, CommitLogStatus::Staged) || !table_matches_staged_base(¤t, existing) { - return Err(TableCatalogStoreError::Conflict( - "existing commit record does not match current table state".to_string(), - )); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::Conflict( + "existing commit record does not match current table state".to_string(), + )), + ); } } if let Some(existing) = existing_idempotency_commit.as_ref() && !commit_log_matches_request(existing, &request, ¤t.table_id) { - return Err(TableCatalogStoreError::Conflict("idempotency key already exists".to_string())); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::Conflict("idempotency key already exists".to_string())), + ); } if existing_commit.is_none() && existing_idempotency_commit.is_some() { - return Err(TableCatalogStoreError::Conflict( - "idempotency key exists without a recoverable commit record".to_string(), - )); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::Conflict( + "idempotency key exists without a recoverable commit record".to_string(), + )), + ); } if current.version_token != request.expected_version_token { - return Err(TableCatalogStoreError::Conflict( - "current table version token does not match expected token".to_string(), - )); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::Conflict( + "current table version token does not match expected token".to_string(), + )), + ); } if current.metadata_location != request.expected_metadata_location { - return Err(TableCatalogStoreError::Conflict( - "current table metadata location does not match expected location".to_string(), - )); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::Conflict( + "current table metadata location does not match expected location".to_string(), + )), + ); } if !is_valid_table_metadata_location(&namespace, &table, &request.new_metadata_location) { - return Err(TableCatalogStoreError::Invalid( - "new metadata location must be inside the table metadata directory".to_string(), - )); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::Invalid( + "new metadata location must be inside the table metadata directory".to_string(), + )), + ); } let Some(new_metadata_object) = self .backend .read_object(&request.table_bucket, &request.new_metadata_location) .await? else { - return Err(TableCatalogStoreError::NotFound(format!( - "new metadata object {}", - request.new_metadata_location - ))); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(TableCatalogStoreError::NotFound(format!( + "new metadata object {}", + request.new_metadata_location + ))), + ); }; let next_warehouse_location = metadata_warehouse_location(&request.table_bucket, &request.new_metadata_location, &new_metadata_object)?; @@ -1544,13 +1853,27 @@ where .await?; } - self.write_entry_unlocked( - self.catalog_bucket(), - &table_path, - &next, - TableCatalogPutPrecondition::IfMatch(current_etag), - ) - .await?; + let cas_started = Instant::now(); + let cas_result = self + .write_entry_unlocked( + self.catalog_bucket(), + &table_path, + &next, + TableCatalogPutPrecondition::IfMatch(current_etag), + ) + .await; + record_table_commit_cas_result(&request.operation, cas_started, &cas_result); + if let Err(err) = cas_result { + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Err(err), + ); + } let mut commit_log = staged_commit_log; commit_log.status = CommitLogStatus::Committed; @@ -1560,7 +1883,15 @@ where .finalize_commit_log(&commit_path, idempotency_path.as_deref(), &commit_log) .await; - Ok(TableCommitResult { table: next, commit_log }) + table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + Ok(TableCommitResult { table: next, commit_log }), + ) } async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> { @@ -1938,6 +2269,269 @@ fn table_matches_staged_base(table: &TableEntry, commit_log: &CommitLogEntry) -> && table.version_token == commit_log.expected_version_token } +fn table_catalog_recovery_summary( + metadata_status: &TableMetadataPointerStatus, + commit_recovery: &TableCommitRecoveryReport, +) -> (TableCatalogRecoveryStatus, Vec) { + let mut actions = Vec::new(); + let metadata_status = match metadata_status { + TableMetadataPointerStatus::Valid => None, + TableMetadataPointerStatus::MissingObject => { + actions.push(TableCatalogRecoveryAction::RestoreCurrentMetadataObject); + Some(TableCatalogRecoveryStatus::ReadOnlyRecommended) + } + TableMetadataPointerStatus::InvalidJson => { + actions.push(TableCatalogRecoveryAction::FixCurrentMetadataJson); + Some(TableCatalogRecoveryStatus::ReadOnlyRecommended) + } + TableMetadataPointerStatus::InvalidLocation => { + actions.push(TableCatalogRecoveryAction::MoveCurrentMetadataInsideTable); + Some(TableCatalogRecoveryStatus::ReadOnlyRecommended) + } + }; + + if commit_recovery.manual_review_count > 0 { + actions.push(TableCatalogRecoveryAction::ReviewCommitLog); + return (metadata_status.unwrap_or(TableCatalogRecoveryStatus::ManualReviewRequired), actions); + } + if commit_recovery.finalization_required_count > 0 || commit_recovery.idempotency_repair_required_count > 0 { + actions.push(TableCatalogRecoveryAction::RunCommitRecovery); + return (metadata_status.unwrap_or(TableCatalogRecoveryStatus::Recoverable), actions); + } + if commit_recovery.staged_before_table_update_count > 0 { + actions.push(TableCatalogRecoveryAction::RetryCommit); + return (metadata_status.unwrap_or(TableCatalogRecoveryStatus::Recoverable), actions); + } + + (metadata_status.unwrap_or(TableCatalogRecoveryStatus::Healthy), actions) +} + +fn commit_logs_share_recovery_payload(left: &CommitLogEntry, right: &CommitLogEntry) -> bool { + left.version == right.version + && left.commit_id == right.commit_id + && left.idempotency_key == right.idempotency_key + && left.table_id == right.table_id + && left.operation == right.operation + && left.expected_version_token == right.expected_version_token + && left.new_version_token == right.new_version_token + && left.previous_metadata_location == right.previous_metadata_location + && left.new_metadata_location == right.new_metadata_location + && left.requirements == right.requirements + && left.writer == right.writer +} + +fn commit_idempotency_index_status( + commit_log: &CommitLogEntry, + idempotency_commit: Option<&CommitLogEntry>, +) -> TableCommitIdempotencyIndexStatus { + match (commit_log.idempotency_key.as_ref(), idempotency_commit) { + (None, _) => TableCommitIdempotencyIndexStatus::NotRequired, + (Some(_), None) => TableCommitIdempotencyIndexStatus::Missing, + (Some(_), Some(indexed)) if indexed == commit_log => TableCommitIdempotencyIndexStatus::Matches, + (Some(_), Some(indexed)) if commit_logs_share_recovery_payload(indexed, commit_log) => { + TableCommitIdempotencyIndexStatus::Stale + } + (Some(_), Some(_)) => TableCommitIdempotencyIndexStatus::Conflicting, + } +} + +fn table_commit_recovery_entry( + table: &TableEntry, + commit_log: &CommitLogEntry, + idempotency_commit: Option<&CommitLogEntry>, +) -> TableCommitRecoveryEntry { + let idempotency_index_status = commit_idempotency_index_status(commit_log, idempotency_commit); + let idempotency_index_present = matches!( + idempotency_index_status, + TableCommitIdempotencyIndexStatus::Matches + | TableCommitIdempotencyIndexStatus::Stale + | TableCommitIdempotencyIndexStatus::Conflicting + ); + let idempotency_index_repair_required = matches!( + idempotency_index_status, + TableCommitIdempotencyIndexStatus::Missing | TableCommitIdempotencyIndexStatus::Stale + ); + + let (recovery_state, reason) = if matches!(idempotency_index_status, TableCommitIdempotencyIndexStatus::Conflicting) { + ( + TableCommitRecoveryState::ManualReview, + "idempotency index points at a different commit payload".to_string(), + ) + } else if table_matches_committed_log(table, commit_log) { + if matches!(commit_log.status, CommitLogStatus::Committed) { + if idempotency_index_repair_required { + ( + TableCommitRecoveryState::IdempotencyIndexRepairRequired, + "committed table pointer is durable but idempotency index needs repair".to_string(), + ) + } else { + ( + TableCommitRecoveryState::Committed, + "commit log and current table pointer agree".to_string(), + ) + } + } else { + ( + TableCommitRecoveryState::FinalizationRequired, + "current table pointer already advanced but commit log is not finalized".to_string(), + ) + } + } else if matches!(commit_log.status, CommitLogStatus::Committed) { + if idempotency_index_repair_required { + ( + TableCommitRecoveryState::IdempotencyIndexRepairRequired, + "historical committed log needs idempotency index repair".to_string(), + ) + } else { + ( + TableCommitRecoveryState::Committed, + "commit is finalized and may be older than the current table pointer".to_string(), + ) + } + } else if table_matches_staged_base(table, commit_log) { + ( + TableCommitRecoveryState::StagedBeforeTableUpdate, + "staged commit exists but table pointer has not advanced".to_string(), + ) + } else { + ( + TableCommitRecoveryState::ManualReview, + "staged commit no longer matches the current table pointer or its expected base".to_string(), + ) + }; + + TableCommitRecoveryEntry { + commit_id: commit_log.commit_id.clone(), + idempotency_key: commit_log.idempotency_key.clone(), + operation: commit_log.operation.clone(), + status: commit_log.status.clone(), + recovery_state, + previous_metadata_location: commit_log.previous_metadata_location.clone(), + new_metadata_location: commit_log.new_metadata_location.clone(), + expected_version_token: commit_log.expected_version_token.clone(), + new_version_token: commit_log.new_version_token.clone(), + idempotency_index_present, + idempotency_index_status, + reason, + } +} + +fn record_table_commit_attempt(operation: &str) { + counter!("rustfs_table_catalog_commit_attempts_total", "operation" => operation.to_string()).increment(1); +} + +fn table_catalog_store_result_label(result: &TableCatalogStoreResult) -> &'static str { + match result { + Ok(_) => "success", + Err(TableCatalogStoreError::Conflict(_)) => "conflict", + Err(TableCatalogStoreError::Invalid(_)) => "invalid", + Err(TableCatalogStoreError::NotFound(_)) => "not_found", + Err(TableCatalogStoreError::Internal(_)) => "failure", + } +} + +fn duration_millis_u64(duration: StdDuration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +fn record_table_commit_cas_result(operation: &str, started: Instant, result: &TableCatalogStoreResult<()>) { + let elapsed = started.elapsed(); + let result_label = table_catalog_store_result_label(result); + counter!( + "rustfs_table_catalog_commit_cas_results_total", + "operation" => operation.to_string(), + "result" => result_label.to_string() + ) + .increment(1); + histogram!( + "rustfs_table_catalog_commit_cas_duration_seconds", + "operation" => operation.to_string(), + "result" => result_label.to_string() + ) + .record(elapsed.as_secs_f64()); +} + +fn record_table_commit_result( + table_bucket: &str, + namespace: &str, + table: &str, + commit_id: &str, + operation: &str, + started: Instant, + result: &TableCatalogStoreResult, +) { + let elapsed = started.elapsed(); + let result_label = table_catalog_store_result_label(result); + counter!( + "rustfs_table_catalog_commit_results_total", + "operation" => operation.to_string(), + "result" => result_label.to_string() + ) + .increment(1); + if matches!(result, Err(TableCatalogStoreError::Conflict(_))) { + counter!("rustfs_table_catalog_commit_conflicts_total", "operation" => operation.to_string()).increment(1); + } + histogram!( + "rustfs_table_catalog_commit_duration_seconds", + "operation" => operation.to_string(), + "result" => result_label.to_string() + ) + .record(elapsed.as_secs_f64()); + + match result { + Ok(commit) if elapsed >= TABLE_COMMIT_SLOW_LOG_THRESHOLD => { + tracing::warn!( + table_bucket, + namespace, + table, + commit_id, + operation, + generation = commit.table.generation, + duration_ms = duration_millis_u64(elapsed), + "slow table catalog commit" + ); + } + Ok(commit) => { + tracing::debug!( + table_bucket, + namespace, + table, + commit_id, + operation, + generation = commit.table.generation, + duration_ms = duration_millis_u64(elapsed), + "table catalog commit completed" + ); + } + Err(error) => { + tracing::warn!( + table_bucket, + namespace, + table, + commit_id, + operation, + result = result_label, + duration_ms = duration_millis_u64(elapsed), + error = %error, + "table catalog commit did not complete" + ); + } + } +} + +fn table_commit_result( + table_bucket: &str, + namespace: &str, + table: &str, + commit_id: &str, + operation: &str, + started: Instant, + result: TableCatalogStoreResult, +) -> TableCatalogStoreResult { + record_table_commit_result(table_bucket, namespace, table, commit_id, operation, started, &result); + result +} + fn http_preconditions_for_catalog_put(precondition: TableCatalogPutPrecondition) -> Option { match precondition { TableCatalogPutPrecondition::Any => None, @@ -3467,6 +4061,8 @@ mod tests { assert_eq!(report.catalog.table.metadata_location, current.clone()); assert_eq!(report.current_metadata_status, TableMetadataPointerStatus::MissingObject); + assert_eq!(report.recovery_status, TableCatalogRecoveryStatus::ReadOnlyRecommended); + assert_eq!(report.recommended_actions, vec![TableCatalogRecoveryAction::RestoreCurrentMetadataObject]); assert!(report.orphan_metadata_candidate_locations.is_empty()); } @@ -3794,6 +4390,289 @@ mod tests { assert_eq!(committed.status, CommitLogStatus::Committed); } + #[tokio::test] + async fn table_commit_recovery_reports_post_cas_staged_commit() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let new_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let commit_path = TableCatalogObjectPaths::default().commit_log_entry_path(bucket, "table-id", "commit-1"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .unwrap(); + backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).await; + backend + .fail_put_attempt(rustfs_ecstore::disk::RUSTFS_META_BUCKET, &commit_path, 2) + .await; + + let result = store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-1".to_string(), + idempotency_key: None, + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current_metadata, + new_metadata_location: new_metadata.clone(), + requirements: Vec::new(), + writer: None, + }) + .await + .unwrap(); + assert_eq!(result.commit_log.status, CommitLogStatus::Committed); + + let report = store.plan_table_commit_recovery(bucket, "sales", "orders").await.unwrap(); + + assert_eq!(report.current_metadata_location, new_metadata); + assert_eq!(report.finalization_required_count, 1); + assert_eq!(report.manual_review_count, 0); + assert_eq!(report.commits.len(), 1); + assert_eq!(report.commits[0].commit_id, "commit-1"); + assert_eq!(report.commits[0].status, CommitLogStatus::Staged); + assert_eq!(report.commits[0].recovery_state, TableCommitRecoveryState::FinalizationRequired); + } + + #[tokio::test] + async fn diagnostics_report_includes_table_commit_recovery_state() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let new_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let commit_path = TableCatalogObjectPaths::default().commit_log_entry_path(bucket, "table-id", "commit-1"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .unwrap(); + backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).await; + backend + .fail_put_attempt(rustfs_ecstore::disk::RUSTFS_META_BUCKET, &commit_path, 2) + .await; + + store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-1".to_string(), + idempotency_key: None, + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current_metadata, + new_metadata_location: new_metadata, + requirements: Vec::new(), + writer: None, + }) + .await + .unwrap(); + + let report = store.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap(); + + assert_eq!(report.recovery_status, TableCatalogRecoveryStatus::Recoverable); + assert_eq!(report.recommended_actions, vec![TableCatalogRecoveryAction::RunCommitRecovery]); + assert_eq!(report.commit_recovery.finalization_required_count, 1); + assert_eq!(report.commit_recovery.commits.len(), 1); + assert_eq!( + report.commit_recovery.commits[0].recovery_state, + TableCommitRecoveryState::FinalizationRequired + ); + } + + #[tokio::test] + async fn table_commit_recovery_finalizes_post_cas_staged_commit() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let new_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let commit_path = TableCatalogObjectPaths::default().commit_log_entry_path(bucket, "table-id", "commit-1"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .unwrap(); + backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).await; + backend + .fail_put_attempt(rustfs_ecstore::disk::RUSTFS_META_BUCKET, &commit_path, 2) + .await; + + store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-1".to_string(), + idempotency_key: None, + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current_metadata, + new_metadata_location: new_metadata, + requirements: Vec::new(), + writer: None, + }) + .await + .unwrap(); + + let report = store.recover_table_commits(bucket, "sales", "orders").await.unwrap(); + + assert_eq!(report.finalized_count, 1); + assert_eq!(report.finalization_required_count, 0); + let committed = store.get_commit_by_id(bucket, "table-id", "commit-1").await.unwrap().unwrap(); + assert_eq!(committed.status, CommitLogStatus::Committed); + } + + #[tokio::test] + async fn table_commit_recovery_reports_staged_commit_after_table_cas_failure() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let new_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let table_path = TableCatalogObjectPaths::default().table_entry_path(bucket, &namespace, &table); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .unwrap(); + backend.seed_object(bucket, ¤t_metadata, b"{}".to_vec()).await; + backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).await; + backend + .fail_put_attempt(rustfs_ecstore::disk::RUSTFS_META_BUCKET, &table_path, 2) + .await; + + let err = store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-1".to_string(), + idempotency_key: None, + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current_metadata.clone(), + new_metadata_location: new_metadata, + requirements: Vec::new(), + writer: None, + }) + .await + .unwrap_err(); + assert_matches!(err, TableCatalogStoreError::Internal(_)); + let loaded = store.load_table(bucket, "sales", "orders").await.unwrap().unwrap(); + assert_eq!(loaded.metadata_location, current_metadata); + + let report = store.plan_table_commit_recovery(bucket, "sales", "orders").await.unwrap(); + assert_eq!(report.staged_before_table_update_count, 1); + assert_eq!(report.finalization_required_count, 0); + assert_eq!(report.commits[0].recovery_state, TableCommitRecoveryState::StagedBeforeTableUpdate); + + let diagnostics = store.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap(); + assert_eq!(diagnostics.current_metadata_status, TableMetadataPointerStatus::Valid); + assert_eq!(diagnostics.recovery_status, TableCatalogRecoveryStatus::Recoverable); + assert_eq!(diagnostics.recommended_actions, vec![TableCatalogRecoveryAction::RetryCommit]); + } + + #[tokio::test] + async fn table_commit_recovery_repairs_stale_idempotency_index_after_partial_finalization() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let new_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let idempotency_path = + TableCatalogObjectPaths::default().commit_idempotency_entry_path(bucket, "table-id", "client-request-1"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .unwrap(); + backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).await; + backend + .fail_put_attempt(rustfs_ecstore::disk::RUSTFS_META_BUCKET, &idempotency_path, 2) + .await; + + let result = store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-1".to_string(), + idempotency_key: Some("client-request-1".to_string()), + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current_metadata, + new_metadata_location: new_metadata, + requirements: Vec::new(), + writer: None, + }) + .await + .unwrap(); + assert_eq!(result.commit_log.status, CommitLogStatus::Committed); + let stale_index = store + .get_commit_by_idempotency_key(bucket, "table-id", "client-request-1") + .await + .unwrap() + .unwrap(); + assert_eq!(stale_index.status, CommitLogStatus::Staged); + + let report = store.plan_table_commit_recovery(bucket, "sales", "orders").await.unwrap(); + assert_eq!(report.idempotency_repair_required_count, 1); + assert_eq!(report.manual_review_count, 0); + assert_eq!(report.commits[0].recovery_state, TableCommitRecoveryState::IdempotencyIndexRepairRequired); + assert_eq!(report.commits[0].idempotency_index_status, TableCommitIdempotencyIndexStatus::Stale); + + let repaired = store.recover_table_commits(bucket, "sales", "orders").await.unwrap(); + + assert_eq!(repaired.finalized_count, 1); + assert_eq!(repaired.idempotency_repair_required_count, 0); + let repaired_index = store + .get_commit_by_idempotency_key(bucket, "table-id", "client-request-1") + .await + .unwrap() + .unwrap(); + assert_eq!(repaired_index.status, CommitLogStatus::Committed); + } + #[tokio::test] async fn object_table_catalog_store_rejects_stale_commit_token() { let backend = TestCatalogObjectBackend::default();