feat(table-catalog): add durable strong backing (#3836)

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<ObjectInfo>;
 type CatalogListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
@@ -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(Clone, Default)]
-#[derive(Default)]
 struct StrongTableCatalogState {
+    hydrated: bool,
+    snapshot_etag: Option<String>,
     table_buckets: BTreeMap<String, TableBucketEntry>,
     namespaces: BTreeMap<StrongNamespaceKey, NamespaceEntry>,
     tables: BTreeMap<StrongResourceKey, TableEntry>,
@@ -1778,14 +1779,33 @@ struct StrongTableCatalogState {
     idempotency: BTreeMap<StrongCommitKey, CommitLogEntry>,
 }
 
-#[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<TableBucketEntry>,
+    namespaces: Vec<NamespaceEntry>,
+    tables: Vec<TableEntry>,
+    views: Vec<ViewEntry>,
+    commits: Vec<StrongCommitSnapshotRecord>,
+    idempotency: Vec<StrongCommitSnapshotRecord>,
+}
+
 #[derive(Clone)]
 pub(crate) struct StrongTableCatalogStore<B> {
     object_backend: B,
     state: Arc<tokio::sync::Mutex<StrongTableCatalogState>>,
+    // Serializes local snapshot mutations; object ETags fence independent store instances.
+    write_lock: Arc<tokio::sync::Mutex<()>>,
 }
 
-#[cfg(test)]
 impl<B> StrongTableCatalogStore<B>
 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<String>,
+    ) -> TableCatalogStoreResult<StrongTableCatalogState> {
+        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::<StrongTableCatalogSnapshot>(&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<TableCommitRecoveryReport> {
+        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<B> TableCatalogStore for StrongTableCatalogStore<B>
 where
     B: TableCatalogObjectBackend,
 {
     async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<Option<TableBucketEntry>> {
+        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,15 +2240,22 @@ where
             return Err(TableCatalogStoreError::Invalid("unsupported table bucket catalog type".to_string()));
         }
 
+        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);
-        Ok(())
+            (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 (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) {
@@ -2079,11 +2264,15 @@ where
                     entry.table_bucket, entry.namespace
                 )));
             }
+            let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state);
             state.namespaces.insert(key, entry);
-        Ok(())
+            (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<Vec<NamespaceEntry>> {
+        self.hydrate_state().await?;
         let state = self.state.lock().await;
         let mut entries = state
             .namespaces
@@ -2096,14 +2285,18 @@ where
     }
 
     async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Option<NamespaceEntry>> {
+        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 (snapshot, precondition, rollback_state) = {
             let mut state = self.state.lock().await;
             if !state.namespaces.contains_key(&key) {
                 return Err(TableCatalogStoreError::NotFound(format!(
@@ -2127,8 +2320,11 @@ where
                     namespace.public_name()
                 )));
             }
+            let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state);
             state.namespaces.remove(&key);
-        Ok(())
+            (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,11 +2332,14 @@ 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 (snapshot, precondition, rollback_state) = {
             let mut state = self.state.lock().await;
             Self::require_table_bucket_in_state(&state, &entry.table_bucket)?;
             if !state
@@ -2158,11 +2357,15 @@ where
                     entry.table_bucket, entry.namespace, entry.table
                 )));
             }
+            let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state);
             state.tables.insert(key, entry);
-        Ok(())
+            (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<Vec<TableEntry>> {
+        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<Option<TableEntry>> {
+        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<TableCommitResult> {
+        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(
+                        let snapshot = Self::snapshot_from_state_locked(&state);
-                            &request.table_bucket,
+                        Some((result, snapshot, precondition, rollback_state))
-                            &request.namespace,
+                    } else {
-                            &request.table,
+                        None
-                            &request.commit_id,
-                            &request.operation,
-                            commit_started,
-                            Ok(result),
-                        );
                     }
                 }
                 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,10 +2489,14 @@ 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 (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 {}/{}/{}",
@@ -2272,14 +2505,19 @@ where
                     table.as_str()
                 )));
             }
-        Ok(())
+            (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 (snapshot, precondition, rollback_state) = {
             let mut state = self.state.lock().await;
             Self::require_table_bucket_in_state(&state, &entry.table_bucket)?;
             if !state
@@ -2297,11 +2535,15 @@ where
                     entry.table_bucket, entry.namespace, entry.view
                 )));
             }
+            let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state);
             state.views.insert(key, entry);
-        Ok(())
+            (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<Vec<ViewEntry>> {
+        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<Option<ViewEntry>> {
+        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<ViewCommitResult> {
+        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,14 +2615,21 @@ 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 (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 {}/{}/{}",
@@ -2385,7 +2638,9 @@ where
                     view.as_str()
                 )));
             }
-        Ok(())
+            (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<Option<CommitLogEntry>> {
+        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<Option<CommitLogEntry>> {
+        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<usize>>,
+        read_attempts: BTreeMap<(String, String), usize>,
         fail_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
         fail_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
         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,10 +9394,22 @@ mod tests {
         async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
             let mut state = self.state.lock().await;
             state.read_calls += 1;
-            Ok(state
+            let key = (bucket.to_string(), object.to_string());
-                .objects
+            let attempt = {
-                .get(&(bucket.to_string(), object.to_string()))
+                let attempts = state.read_attempts.entry(key.clone()).or_default();
-                .map(|record| TableCatalogObject {
+                *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,
@@ -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::<TestCatalogObjectBackend>::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::<TestCatalogObjectBackend>::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();