From 1c4a279909c625bc2ec057e1033fa5bed8efab8d Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 25 Jun 2026 12:22:30 +0800 Subject: [PATCH] feat(table-catalog): add durable strong backing (#3836) --- rustfs/src/table_catalog.rs | 794 ++++++++++++++++++++++++++++++------ 1 file changed, 672 insertions(+), 122 deletions(-) diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index b05fcda6f..3e574b76e 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -115,6 +115,9 @@ const ICEBERG_MAX_REF_AGE_MS_PROPERTY: &str = "history.expire.max-ref-age-ms"; const ICEBERG_REF_MIN_SNAPSHOTS_TO_KEEP_FIELD: &str = "min-snapshots-to-keep"; const ICEBERG_REF_MAX_SNAPSHOT_AGE_MS_FIELD: &str = "max-snapshot-age-ms"; const ICEBERG_REF_MAX_REF_AGE_MS_FIELD: &str = "max-ref-age-ms"; +const STRONG_TABLE_CATALOG_SNAPSHOT_VERSION: u16 = 1; +const STRONG_TABLE_CATALOG_BACKING_ROOT: &str = "strong-backing"; +const STRONG_TABLE_CATALOG_SNAPSHOT_FILE: &str = "snapshot.json"; type CatalogListObjectsV2Info = StorageListObjectsV2Info; type CatalogListObjectVersionsInfo = StorageListObjectVersionsInfo; @@ -1760,16 +1763,14 @@ fn table_catalog_backing_manifest( } } -#[cfg(test)] type StrongNamespaceKey = (String, String); -#[cfg(test)] type StrongResourceKey = (String, String, String); -#[cfg(test)] type StrongCommitKey = (String, String, String); -#[cfg(test)] -#[derive(Default)] +#[derive(Clone, Default)] struct StrongTableCatalogState { + hydrated: bool, + snapshot_etag: Option, table_buckets: BTreeMap, namespaces: BTreeMap, tables: BTreeMap, @@ -1778,14 +1779,33 @@ struct StrongTableCatalogState { idempotency: BTreeMap, } -#[cfg(test)] +#[derive(Debug, Clone, Serialize, Deserialize)] +struct StrongCommitSnapshotRecord { + table_bucket: String, + table_id: String, + lookup_key: String, + commit: CommitLogEntry, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct StrongTableCatalogSnapshot { + version: u16, + table_buckets: Vec, + namespaces: Vec, + tables: Vec, + views: Vec, + commits: Vec, + idempotency: Vec, +} + #[derive(Clone)] pub(crate) struct StrongTableCatalogStore { object_backend: B, state: Arc>, + // Serializes local snapshot mutations; object ETags fence independent store instances. + write_lock: Arc>, } -#[cfg(test)] impl StrongTableCatalogStore where B: TableCatalogObjectBackend, @@ -1794,6 +1814,7 @@ where Self { object_backend, state: Arc::new(tokio::sync::Mutex::new(StrongTableCatalogState::default())), + write_lock: Arc::new(tokio::sync::Mutex::new(())), } } @@ -1813,6 +1834,160 @@ where (table_bucket.to_string(), table_id.to_string(), idempotency_key.to_string()) } + fn snapshot_object_path() -> String { + format!("{INTERNAL_CATALOG_ROOT}/{STRONG_TABLE_CATALOG_BACKING_ROOT}/{STRONG_TABLE_CATALOG_SNAPSHOT_FILE}") + } + + fn snapshot_from_state_locked(state: &StrongTableCatalogState) -> StrongTableCatalogSnapshot { + StrongTableCatalogSnapshot { + version: STRONG_TABLE_CATALOG_SNAPSHOT_VERSION, + table_buckets: state.table_buckets.values().cloned().collect(), + namespaces: state.namespaces.values().cloned().collect(), + tables: state.tables.values().cloned().collect(), + views: state.views.values().cloned().collect(), + commits: state + .commits + .iter() + .map(|((table_bucket, table_id, lookup_key), commit)| StrongCommitSnapshotRecord { + table_bucket: table_bucket.clone(), + table_id: table_id.clone(), + lookup_key: lookup_key.clone(), + commit: commit.clone(), + }) + .collect(), + idempotency: state + .idempotency + .iter() + .map(|((table_bucket, table_id, lookup_key), commit)| StrongCommitSnapshotRecord { + table_bucket: table_bucket.clone(), + table_id: table_id.clone(), + lookup_key: lookup_key.clone(), + commit: commit.clone(), + }) + .collect(), + } + } + + fn state_from_snapshot( + snapshot: StrongTableCatalogSnapshot, + snapshot_etag: Option, + ) -> TableCatalogStoreResult { + if snapshot.version != STRONG_TABLE_CATALOG_SNAPSHOT_VERSION { + return Err(TableCatalogStoreError::Invalid(format!( + "unsupported strong catalog snapshot version: {}", + snapshot.version + ))); + } + + let mut state = StrongTableCatalogState { + hydrated: true, + snapshot_etag, + ..StrongTableCatalogState::default() + }; + for entry in snapshot.table_buckets { + state.table_buckets.insert(entry.table_bucket.clone(), entry); + } + for entry in snapshot.namespaces { + let namespace = parse_namespace_for_store(&entry.namespace)?; + state + .namespaces + .insert(Self::namespace_key(&entry.table_bucket, &namespace), entry); + } + for entry in snapshot.tables { + let namespace = parse_namespace_for_store(&entry.namespace)?; + let table = parse_table_for_store(&entry.table)?; + state + .tables + .insert(Self::table_key(&entry.table_bucket, &namespace, &table), entry); + } + for entry in snapshot.views { + let namespace = parse_namespace_for_store(&entry.namespace)?; + let view = parse_table_for_store(&entry.view)?; + state + .views + .insert(Self::table_key(&entry.table_bucket, &namespace, &view), entry); + } + for record in snapshot.commits { + state.commits.insert( + Self::commit_key(&record.table_bucket, &record.table_id, &record.lookup_key), + record.commit, + ); + } + for record in snapshot.idempotency { + state.idempotency.insert( + Self::idempotency_key(&record.table_bucket, &record.table_id, &record.lookup_key), + record.commit, + ); + } + Ok(state) + } + + fn snapshot_write_precondition_locked(state: &StrongTableCatalogState) -> TableCatalogPutPrecondition { + state + .snapshot_etag + .as_ref() + .map_or(TableCatalogPutPrecondition::IfAbsent, |etag| { + TableCatalogPutPrecondition::IfMatch(etag.clone()) + }) + } + + fn snapshot_write_context_locked(state: &StrongTableCatalogState) -> (TableCatalogPutPrecondition, StrongTableCatalogState) { + (Self::snapshot_write_precondition_locked(state), state.clone()) + } + + async fn hydrate_state(&self) -> TableCatalogStoreResult<()> { + self.reload_state_from_durable().await + } + + async fn reload_state_from_durable(&self) -> TableCatalogStoreResult<()> { + let snapshot_object = self + .object_backend + .read_object(RUSTFS_META_BUCKET, &Self::snapshot_object_path()) + .await?; + let mut state = self.state.lock().await; + if let Some(snapshot_object) = snapshot_object { + let snapshot = serde_json::from_slice::(&snapshot_object.data) + .map_err(|err| TableCatalogStoreError::Internal(format!("failed to decode strong catalog snapshot: {err}")))?; + *state = Self::state_from_snapshot(snapshot, snapshot_object.etag)?; + } else { + *state = StrongTableCatalogState { + hydrated: true, + ..StrongTableCatalogState::default() + }; + } + Ok(()) + } + + async fn persist_snapshot( + &self, + snapshot: StrongTableCatalogSnapshot, + precondition: TableCatalogPutPrecondition, + ) -> TableCatalogStoreResult<()> { + let data = serde_json::to_vec(&snapshot) + .map_err(|err| TableCatalogStoreError::Internal(format!("failed to encode strong catalog snapshot: {err}")))?; + self.object_backend + .put_object(RUSTFS_META_BUCKET, &Self::snapshot_object_path(), data, precondition) + .await + } + + async fn finalize_snapshot_write( + &self, + snapshot: StrongTableCatalogSnapshot, + precondition: TableCatalogPutPrecondition, + rollback_state: StrongTableCatalogState, + ) -> TableCatalogStoreResult<()> { + match self.persist_snapshot(snapshot, precondition).await { + Ok(()) => self.reload_state_from_durable().await, + Err(err) => { + if self.reload_state_from_durable().await.is_err() { + let mut state = self.state.lock().await; + *state = rollback_state; + } + Err(err) + } + } + } + fn require_table_bucket_in_state(state: &StrongTableCatalogState, table_bucket: &str) -> TableCatalogStoreResult<()> { if !state.table_buckets.contains_key(table_bucket) { return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); @@ -2026,6 +2201,7 @@ where namespace: &str, table: &str, ) -> TableCatalogStoreResult { + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let table = parse_table_for_store(table)?; let key = Self::table_key(table_bucket, &namespace, &table); @@ -2042,18 +2218,20 @@ where } } -#[cfg(test)] #[async_trait::async_trait] impl TableCatalogStore for StrongTableCatalogStore where B: TableCatalogObjectBackend, { async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let state = self.state.lock().await; Ok(state.table_buckets.get(table_bucket).cloned()) } async fn put_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; validate_catalog_entry_version("table bucket", entry.version)?; if entry.table_bucket.is_empty() { return Err(TableCatalogStoreError::Invalid("table bucket name cannot be empty".to_string())); @@ -2062,28 +2240,39 @@ where return Err(TableCatalogStoreError::Invalid("unsupported table bucket catalog type".to_string())); } - let mut state = self.state.lock().await; - state.table_buckets.insert(entry.table_bucket.clone(), entry); - Ok(()) + let (snapshot, precondition, rollback_state) = { + let mut state = self.state.lock().await; + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + state.table_buckets.insert(entry.table_bucket.clone(), entry); + (Self::snapshot_from_state_locked(&state), precondition, rollback_state) + }; + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await } async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; validate_catalog_entry_version("namespace", entry.version)?; let namespace = parse_namespace_for_store(&entry.namespace)?; let key = Self::namespace_key(&entry.table_bucket, &namespace); - let mut state = self.state.lock().await; - Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; - if state.namespaces.contains_key(&key) { - return Err(TableCatalogStoreError::Conflict(format!( - "catalog object already exists: namespace {}/{}", - entry.table_bucket, entry.namespace - ))); - } - state.namespaces.insert(key, entry); - Ok(()) + let (snapshot, precondition, rollback_state) = { + let mut state = self.state.lock().await; + Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; + if state.namespaces.contains_key(&key) { + return Err(TableCatalogStoreError::Conflict(format!( + "catalog object already exists: namespace {}/{}", + entry.table_bucket, entry.namespace + ))); + } + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + state.namespaces.insert(key, entry); + (Self::snapshot_from_state_locked(&state), precondition, rollback_state) + }; + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await } async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let state = self.state.lock().await; let mut entries = state .namespaces @@ -2096,39 +2285,46 @@ where } async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let state = self.state.lock().await; Ok(state.namespaces.get(&Self::namespace_key(table_bucket, &namespace)).cloned()) } async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()> { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let key = Self::namespace_key(table_bucket, &namespace); - let mut state = self.state.lock().await; - if !state.namespaces.contains_key(&key) { - return Err(TableCatalogStoreError::NotFound(format!( - "namespace {}/{}", - table_bucket, - namespace.public_name() - ))); - } - if state - .tables - .keys() - .any(|(bucket, namespace_name, _)| bucket == table_bucket && namespace_name == &namespace.public_name()) - || state - .views + let (snapshot, precondition, rollback_state) = { + let mut state = self.state.lock().await; + if !state.namespaces.contains_key(&key) { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {}/{}", + table_bucket, + namespace.public_name() + ))); + } + if state + .tables .keys() .any(|(bucket, namespace_name, _)| bucket == table_bucket && namespace_name == &namespace.public_name()) - { - return Err(TableCatalogStoreError::Conflict(format!( - "namespace {}/{} is not empty", - table_bucket, - namespace.public_name() - ))); - } - state.namespaces.remove(&key); - Ok(()) + || state + .views + .keys() + .any(|(bucket, namespace_name, _)| bucket == table_bucket && namespace_name == &namespace.public_name()) + { + return Err(TableCatalogStoreError::Conflict(format!( + "namespace {}/{} is not empty", + table_bucket, + namespace.public_name() + ))); + } + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + state.namespaces.remove(&key); + (Self::snapshot_from_state_locked(&state), precondition, rollback_state) + }; + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await } async fn create_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()> { @@ -2136,33 +2332,40 @@ where } async fn register_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()> { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; validate_catalog_entry_version("table", entry.version)?; let namespace = parse_namespace_for_store(&entry.namespace)?; let table = parse_table_for_store(&entry.table)?; table_warehouse_object_prefix(&entry)?; let key = Self::table_key(&entry.table_bucket, &namespace, &table); - let mut state = self.state.lock().await; - Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; - if !state - .namespaces - .contains_key(&Self::namespace_key(&entry.table_bucket, &namespace)) - { - return Err(TableCatalogStoreError::NotFound(format!( - "namespace {}/{}", - entry.table_bucket, entry.namespace - ))); - } - if state.tables.contains_key(&key) { - return Err(TableCatalogStoreError::Conflict(format!( - "catalog object already exists: table {}/{}/{}", - entry.table_bucket, entry.namespace, entry.table - ))); - } - state.tables.insert(key, entry); - Ok(()) + let (snapshot, precondition, rollback_state) = { + let mut state = self.state.lock().await; + Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; + if !state + .namespaces + .contains_key(&Self::namespace_key(&entry.table_bucket, &namespace)) + { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {}/{}", + entry.table_bucket, entry.namespace + ))); + } + if state.tables.contains_key(&key) { + return Err(TableCatalogStoreError::Conflict(format!( + "catalog object already exists: table {}/{}/{}", + entry.table_bucket, entry.namespace, entry.table + ))); + } + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + state.tables.insert(key, entry); + (Self::snapshot_from_state_locked(&state), precondition, rollback_state) + }; + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await } async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let state = self.state.lock().await; let mut entries = state @@ -2176,6 +2379,7 @@ where } async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let table = parse_table_for_store(table)?; let state = self.state.lock().await; @@ -2183,27 +2387,25 @@ where } async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; 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 key = Self::table_key(&request.table_bucket, &namespace, &table); - { + let committed_existing_result = { let mut state = self.state.lock().await; let current = Self::validate_new_table_commit_locked(&state, &key, &request, &namespace, &table); match current { Ok(current) => { + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); if let Some(result) = Self::committed_existing_result_locked(&mut state, &request, current) { - return table_commit_result( - &request.table_bucket, - &request.namespace, - &request.table, - &request.commit_id, - &request.operation, - commit_started, - Ok(result), - ); + let snapshot = Self::snapshot_from_state_locked(&state); + Some((result, snapshot, precondition, rollback_state)) + } else { + None } } Err(error) => { @@ -2218,6 +2420,21 @@ where ); } } + }; + if let Some((result, snapshot, precondition, rollback_state)) = committed_existing_result { + let result = self + .finalize_snapshot_write(snapshot, precondition, rollback_state) + .await + .map(|_| result); + return table_commit_result( + &request.table_bucket, + &request.namespace, + &request.table, + &request.commit_id, + &request.operation, + commit_started, + result, + ); } let Some(new_metadata_object) = self @@ -2242,9 +2459,21 @@ where metadata_warehouse_location(&request.table_bucket, &request.new_metadata_location, &new_metadata_object)?; let cas_started = Instant::now(); - let result = { + let (result, snapshot, precondition, rollback_state) = { let mut state = self.state.lock().await; - Self::apply_commit_locked(&mut state, &request, &namespace, &table, next_warehouse_location) + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + let result = Self::apply_commit_locked(&mut state, &request, &namespace, &table, next_warehouse_location); + let snapshot = result.as_ref().ok().map(|_| Self::snapshot_from_state_locked(&state)); + let precondition = result.as_ref().ok().map(|_| precondition); + let rollback_state = result.as_ref().ok().map(|_| rollback_state); + (result, snapshot, precondition, rollback_state) + }; + let result = match (result, snapshot, precondition, rollback_state) { + (Ok(result), Some(snapshot), Some(precondition), Some(rollback_state)) => self + .finalize_snapshot_write(snapshot, precondition, rollback_state) + .await + .map(|_| result), + (result, _, _, _) => result, }; let cas_result = result.as_ref().map(|_| ()).map_err(Clone::clone); record_table_commit_cas_result(&request.operation, cas_started, &cas_result); @@ -2260,48 +2489,61 @@ where } async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let table = parse_table_for_store(table)?; let key = Self::table_key(table_bucket, &namespace, &table); - let mut state = self.state.lock().await; - if state.tables.remove(&key).is_none() { - return Err(TableCatalogStoreError::NotFound(format!( - "table {}/{}/{}", - table_bucket, - namespace.public_name(), - table.as_str() - ))); - } - Ok(()) + let (snapshot, precondition, rollback_state) = { + let mut state = self.state.lock().await; + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + if state.tables.remove(&key).is_none() { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + } + (Self::snapshot_from_state_locked(&state), precondition, rollback_state) + }; + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await } async fn create_view(&self, entry: ViewEntry) -> TableCatalogStoreResult<()> { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; validate_catalog_entry_version("view", entry.version)?; let namespace = parse_namespace_for_store(&entry.namespace)?; let view = parse_table_for_store(&entry.view)?; let key = Self::table_key(&entry.table_bucket, &namespace, &view); - let mut state = self.state.lock().await; - Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; - if !state - .namespaces - .contains_key(&Self::namespace_key(&entry.table_bucket, &namespace)) - { - return Err(TableCatalogStoreError::NotFound(format!( - "namespace {}/{}", - entry.table_bucket, entry.namespace - ))); - } - if state.views.contains_key(&key) { - return Err(TableCatalogStoreError::Conflict(format!( - "catalog object already exists: view {}/{}/{}", - entry.table_bucket, entry.namespace, entry.view - ))); - } - state.views.insert(key, entry); - Ok(()) + let (snapshot, precondition, rollback_state) = { + let mut state = self.state.lock().await; + Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; + if !state + .namespaces + .contains_key(&Self::namespace_key(&entry.table_bucket, &namespace)) + { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {}/{}", + entry.table_bucket, entry.namespace + ))); + } + if state.views.contains_key(&key) { + return Err(TableCatalogStoreError::Conflict(format!( + "catalog object already exists: view {}/{}/{}", + entry.table_bucket, entry.namespace, entry.view + ))); + } + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + state.views.insert(key, entry); + (Self::snapshot_from_state_locked(&state), precondition, rollback_state) + }; + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await } async fn list_views(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let state = self.state.lock().await; let mut entries = state @@ -2315,6 +2557,7 @@ where } async fn load_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let view = parse_table_for_store(view)?; let state = self.state.lock().await; @@ -2322,6 +2565,8 @@ where } async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; let namespace = parse_namespace_for_store(&request.namespace)?; let view = parse_table_for_store(&request.view)?; if !is_valid_view_metadata_location(&namespace, &view, &request.new_metadata_location) { @@ -2344,6 +2589,7 @@ where let key = Self::table_key(&request.table_bucket, &namespace, &view); let mut state = self.state.lock().await; + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); let Some(current) = state.views.get(&key).cloned() else { return Err(TableCatalogStoreError::NotFound(format!( "view {}/{}/{}", @@ -2369,23 +2615,32 @@ where next.version_token = format!("token-{}", Uuid::new_v4()); next.generation = next.generation.saturating_add(1); state.views.insert(key, next.clone()); + let snapshot = Self::snapshot_from_state_locked(&state); + drop(state); + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await?; Ok(ViewCommitResult { view: next }) } async fn drop_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<()> { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let view = parse_table_for_store(view)?; let key = Self::table_key(table_bucket, &namespace, &view); - let mut state = self.state.lock().await; - if state.views.remove(&key).is_none() { - return Err(TableCatalogStoreError::NotFound(format!( - "view {}/{}/{}", - table_bucket, - namespace.public_name(), - view.as_str() - ))); - } - Ok(()) + let (snapshot, precondition, rollback_state) = { + let mut state = self.state.lock().await; + let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); + if state.views.remove(&key).is_none() { + return Err(TableCatalogStoreError::NotFound(format!( + "view {}/{}/{}", + table_bucket, + namespace.public_name(), + view.as_str() + ))); + } + (Self::snapshot_from_state_locked(&state), precondition, rollback_state) + }; + self.finalize_snapshot_write(snapshot, precondition, rollback_state).await } async fn get_commit_by_id( @@ -2394,6 +2649,7 @@ where table_id: &str, commit_id: &str, ) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let state = self.state.lock().await; Ok(state .commits @@ -2407,6 +2663,7 @@ where table_id: &str, idempotency_key: &str, ) -> TableCatalogStoreResult> { + self.hydrate_state().await?; let state = self.state.lock().await; Ok(state .idempotency @@ -8805,6 +9062,8 @@ mod tests { #[derive(Default)] struct TestCatalogObjectState { objects: BTreeMap<(String, String), TestCatalogObjectRecord>, + fail_read_attempts: BTreeMap<(String, String), BTreeSet>, + read_attempts: BTreeMap<(String, String), usize>, fail_put_attempts: BTreeMap<(String, String), BTreeSet>, fail_delete_attempts: BTreeMap<(String, String), BTreeSet>, put_attempts: BTreeMap<(String, String), usize>, @@ -8866,6 +9125,20 @@ mod tests { state.read_calls = 0; state.list_calls = 0; } + + async fn fail_next_read(&self, bucket: &str, object: &str) { + let mut state = self.state.lock().await; + let key = (bucket.to_string(), object.to_string()); + let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1; + state.fail_read_attempts.entry(key).or_default().insert(next_attempt); + } + + async fn fail_next_put(&self, bucket: &str, object: &str) { + let mut state = self.state.lock().await; + let key = (bucket.to_string(), object.to_string()); + let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1; + state.fail_put_attempts.entry(key).or_default().insert(next_attempt); + } } impl TestCatalogObjectState { @@ -9121,14 +9394,26 @@ mod tests { async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult> { let mut state = self.state.lock().await; state.read_calls += 1; - Ok(state - .objects - .get(&(bucket.to_string(), object.to_string())) - .map(|record| TableCatalogObject { - data: record.data.clone(), - etag: Some(record.etag.clone()), - mod_time: record.mod_time, - })) + let key = (bucket.to_string(), object.to_string()); + let attempt = { + let attempts = state.read_attempts.entry(key.clone()).or_default(); + *attempts += 1; + *attempts + }; + if state + .fail_read_attempts + .get(&key) + .is_some_and(|attempts| attempts.contains(&attempt)) + { + return Err(TableCatalogStoreError::Internal(format!( + "injected read failure for {object} attempt {attempt}" + ))); + } + Ok(state.objects.get(&key).map(|record| TableCatalogObject { + data: record.data.clone(), + etag: Some(record.etag.clone()), + mod_time: record.mod_time, + })) } async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult { @@ -12699,6 +12984,271 @@ mod tests { assert_eq!(recovery.manual_review_count, 0); } + #[tokio::test] + async fn strong_catalog_backing_replays_durable_commit_state_after_restart() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + 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"); + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .expect("table should be created"); + backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).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".to_string()), + 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: Some("pyiceberg/test".to_string()), + }) + .await + .expect("commit should succeed"); + + let restarted = StrongTableCatalogStore::new(backend.clone()); + let loaded = restarted + .load_table(bucket, "sales", "orders") + .await + .expect("table load should succeed") + .expect("table should replay from durable state"); + assert_eq!(loaded.metadata_location, result.table.metadata_location); + assert_eq!(loaded.version_token, result.table.version_token); + assert_eq!( + restarted + .get_commit_by_id(bucket, "table-id", "commit-1") + .await + .expect("commit lookup should succeed") + .expect("commit log should replay from durable state") + .status, + CommitLogStatus::Committed + ); + assert_eq!( + restarted + .get_commit_by_idempotency_key(bucket, "table-id", "client-request") + .await + .expect("idempotency lookup should succeed") + .expect("idempotency index should replay from durable state") + .status, + CommitLogStatus::Committed + ); + + let recovery = restarted + .plan_table_commit_recovery(bucket, "sales", "orders") + .await + .expect("recovery report should replay from durable state"); + assert_eq!(recovery.finalized_count, 1); + assert_eq!(recovery.finalization_required_count, 0); + assert_eq!(recovery.idempotency_repair_required_count, 0); + assert_eq!(recovery.manual_review_count, 0); + } + + #[tokio::test] + async fn strong_catalog_backing_rejects_stale_snapshot_cas_after_concurrent_restart() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let first_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let stale_metadata = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + store + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .expect("table should be created"); + backend.seed_object(bucket, &first_metadata, b"{}".to_vec()).await; + backend.seed_object(bucket, &stale_metadata, b"{}".to_vec()).await; + + let stale_store = StrongTableCatalogStore::new(backend.clone()); + stale_store + .load_table(bucket, "sales", "orders") + .await + .expect("stale store should hydrate") + .expect("table should exist"); + + let first_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.clone(), + new_metadata_location: first_metadata.clone(), + requirements: Vec::new(), + writer: Some("pyiceberg/test".to_string()), + }) + .await + .expect("first commit should succeed"); + + let err = stale_store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-2".to_string(), + idempotency_key: Some("client-request-2".to_string()), + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current_metadata, + new_metadata_location: stale_metadata, + requirements: Vec::new(), + writer: Some("pyiceberg/test".to_string()), + }) + .await + .expect_err("stale snapshot CAS should fail"); + + assert_matches!(err, TableCatalogStoreError::Conflict(_)); + let loaded = stale_store + .load_table(bucket, "sales", "orders") + .await + .expect("stale store should reload after CAS conflict") + .expect("table should still exist"); + assert_eq!(loaded.metadata_location, first_result.table.metadata_location); + assert_eq!(loaded.version_token, first_result.table.version_token); + assert!( + stale_store + .get_commit_by_id(bucket, "table-id", "commit-2") + .await + .expect("commit lookup should succeed") + .is_none() + ); + assert!( + stale_store + .get_commit_by_idempotency_key(bucket, "table-id", "client-request-2") + .await + .expect("idempotency lookup should succeed") + .is_none() + ); + } + + #[tokio::test] + async fn strong_catalog_backing_refreshes_hydrated_reads_after_independent_commit() { + let backend = TestCatalogObjectBackend::default(); + let writer = StrongTableCatalogStore::new(backend.clone()); + let reader = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + 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"); + + writer + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + writer + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + writer + .create_table(test_table_entry(bucket, &namespace, &table, current_metadata.clone())) + .await + .expect("table should be created"); + backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).await; + + let loaded_before_commit = reader + .load_table(bucket, "sales", "orders") + .await + .expect("reader should hydrate") + .expect("table should exist"); + assert_eq!(loaded_before_commit.metadata_location, current_metadata); + + let result = writer + .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: Some("pyiceberg/test".to_string()), + }) + .await + .expect("writer commit should succeed"); + + let loaded_after_commit = reader + .load_table(bucket, "sales", "orders") + .await + .expect("reader should refresh durable state") + .expect("table should still exist"); + assert_eq!(loaded_after_commit.metadata_location, result.table.metadata_location); + assert_eq!(loaded_after_commit.version_token, result.table.version_token); + } + + #[tokio::test] + async fn strong_catalog_backing_rolls_back_cache_when_persist_and_reload_fail() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let snapshot_path = StrongTableCatalogStore::::snapshot_object_path(); + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + backend.fail_next_put(RUSTFS_META_BUCKET, &snapshot_path).await; + backend.fail_next_read(RUSTFS_META_BUCKET, &snapshot_path).await; + + let err = store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect_err("namespace write should fail"); + + assert_matches!(err, TableCatalogStoreError::Internal(_)); + let state = store.state.lock().await; + assert!( + !state + .namespaces + .contains_key(&StrongTableCatalogStore::::namespace_key(bucket, &namespace)) + ); + drop(state); + + assert!( + store + .list_namespaces(bucket) + .await + .expect("durable state should reload after transient failure") + .is_empty() + ); + } + #[tokio::test] async fn strong_catalog_backing_commit_conflict_keeps_pointer_and_wal_unchanged() { let backend = TestCatalogObjectBackend::default();