Claude: don't re-attempt commit unless we're sure it failed

Author: ublubu

src/storage/src/sink/iceberg.rs

@@ -777,11 +777,39 @@ async fn reload_table(
     }
 }
 
+/// What the commit operator knows about the Iceberg table in between commit attempts.
+///
+/// A commit whose response never arrived may or may not have been applied by the catalog.
+/// Sending another commit while that is unknown risks writing the same data files twice:
+/// `Transaction::commit` re-bases a transaction whose base has gone stale onto whatever the
+/// table has become, so the second attempt satisfies its preconditions and lands a duplicate
+/// snapshot instead of being rejected. The unknown therefore has to survive across retries, and
+/// can only be cleared by reading the table back.
+enum CommitState {
+    /// No commit of ours is outstanding, so this table is a sound base for the next one.
+    Known(Table),
+    /// A commit was sent and we never learned whether it landed. The table held here is the base
+    /// it was sent against, which is no longer known to be current.
+    Unresolved(Table),
+}
+
+impl CommitState {
+    /// The most recent table state. Sound as a base for a new commit only when [`Self::Known`].
+    fn into_table(self) -> Table {
+        match self {
+            CommitState::Known(table) | CommitState::Unresolved(table) => table,
+        }
+    }
+}
+
 /// Attempt a single commit of a batch of data files to an Iceberg table.
-/// On conflict or failure, reloads the table and returns a retryable error.
+///
+/// If a previous attempt left the outcome unknown, first reload the table to establish whether
+/// that attempt landed, and commit again only once it is established that it did not.
+/// On conflict, reloads the table and returns a retryable error.
 /// On success, returns the updated table state.
 async fn try_commit_batch(
-    mut table: Table,
+    state: CommitState,
     snapshot_properties: Vec<(String, String)>,
     data_files: Vec<DataFile>,
     delete_files: Vec<DataFile>,
@@ -793,7 +821,49 @@ async fn try_commit_batch(
     batch_lower: &Antichain<Timestamp>,
     batch_upper: &Antichain<Timestamp>,
     metrics: &IcebergSinkMetrics,
-) -> (Table, RetryResult<(), anyhow::Error>) {
+) -> (CommitState, RetryResult<(), anyhow::Error>) {
+    let table = match state {
+        CommitState::Known(table) => table,
+        CommitState::Unresolved(stale) => {
+            let reloaded = match reload_table(
+                catalog,
+                conn_namespace.to_string(),
+                conn_table.to_string(),
+                stale.clone(),
+            )
+            .await
+            {
+                Ok(reloaded) => reloaded,
+                // Still unknown. Spend the retry on another read, never on another commit.
+                Err(e) => {
+                    return (
+                        CommitState::Unresolved(stale),
+                        RetryResult::RetryableErr(anyhow!(e)),
+                    );
+                }
+            };
+
+            let mut snapshots: Vec<_> = reloaded.metadata().snapshots().cloned().collect();
+            match retrieve_upper_from_snapshots(&mut snapshots) {
+                // Our own commit for this batch is already on the table. It landed and we never
+                // saw the response.
+                Ok(Some((last_frontier, last_version)))
+                    if last_version == sink_version && last_frontier == *frontier =>
+                {
+                    return (CommitState::Known(reloaded), RetryResult::Ok(()));
+                }
+                // It is not there, so it never landed and committing again is safe.
+                Ok(_) => reloaded,
+                Err(e) => {
+                    return (
+                        CommitState::Unresolved(stale),
+                        RetryResult::RetryableErr(anyhow!(e)),
+                    );
+                }
+            }
+        }
+    };
+
     let tx = Transaction::new(&table);
     let mut action = tx
         .row_delta()
@@ -812,21 +882,24 @@ async fn try_commit_batch(
     {
         Ok(tx) => tx,
         Err(e) => {
-            match reload_table(
+            let reloaded = match reload_table(
                 catalog,
                 conn_namespace.to_string(),
                 conn_table.to_string(),
                 table.clone(),
             )
             .await
             {
-                Ok(reloaded) => table = reloaded,
+                Ok(reloaded) => reloaded,
                 Err(reload_err) => {
-                    return (table, RetryResult::RetryableErr(anyhow!(reload_err)));
+                    return (
+                        CommitState::Known(table),
+                        RetryResult::RetryableErr(anyhow!(reload_err)),
+                    );
                 }
-            }
+            };
             return (
-                table,
+                CommitState::Known(reloaded),
                 RetryResult::RetryableErr(anyhow!(
                     "Failed to apply data file addition to iceberg table transaction: {}",
                     e
@@ -839,17 +912,20 @@ async fn try_commit_batch(
     match new_table {
         Err(e) if matches!(e.kind(), ErrorKind::CatalogCommitConflicts) => {
             metrics.commit_conflicts.inc();
-            match reload_table(
+            let table = match reload_table(
                 catalog,
                 conn_namespace.to_string(),
                 conn_table.to_string(),
                 table.clone(),
             )
             .await
             {
-                Ok(reloaded) => table = reloaded,
+                Ok(reloaded) => reloaded,
                 Err(e) => {
-                    return (table, RetryResult::RetryableErr(anyhow!(e)));
+                    return (
+                        CommitState::Known(table),
+                        RetryResult::RetryableErr(anyhow!(e)),
+                    );
                 }
             };
 
@@ -858,15 +934,18 @@ async fn try_commit_batch(
             let last = match last {
                 Ok(val) => val,
                 Err(e) => {
-                    return (table, RetryResult::RetryableErr(anyhow!(e)));
+                    return (
+                        CommitState::Known(table),
+                        RetryResult::RetryableErr(anyhow!(e)),
+                    );
                 }
             };
 
             // Check if another writer has advanced the frontier beyond ours (fencing check)
             if let Some((last_frontier, last_version)) = last {
                 if last_version > sink_version {
                     return (
-                        table,
+                        CommitState::Known(table),
                         RetryResult::FatalErr(anyhow!(
                             "Iceberg table '{}' has been modified by another writer \
                              with version {}. Current sink version: {}. \
@@ -879,7 +958,7 @@ async fn try_commit_batch(
                 }
                 if PartialOrder::less_equal(frontier, &last_frontier) {
                     return (
-                        table,
+                        CommitState::Known(table),
                         RetryResult::FatalErr(anyhow!(
                             "Iceberg table '{}' has been modified by another writer. \
                              Current frontier: {:?}, last frontier: {:?}.",
@@ -892,7 +971,7 @@ async fn try_commit_batch(
             }
 
             (
-                table,
+                CommitState::Known(table),
                 RetryResult::RetryableErr(anyhow!(
                     "Commit conflict detected when committing batch [{}, {}) \
                      to Iceberg table '{}.{}'. Retrying...",
@@ -903,11 +982,23 @@ async fn try_commit_batch(
                 )),
             )
         }
+        // The catalog may have applied this commit before the response was lost. Only a read of
+        // the table can say which, so record that the outcome is unknown and resolve it on the
+        // next attempt, before any further commit is sent.
+        Err(e) if matches!(e.kind(), ErrorKind::Unexpected) => {
+            metrics.commit_failures.inc();
+            (
+                CommitState::Unresolved(table),
+                RetryResult::RetryableErr(anyhow!(e)),
+            )
+        }
+        // Everything else is deterministic: the same batch against the same table fails the same
+        // way however often it is retried.
         Err(e) => {
             metrics.commit_failures.inc();
-            (table, RetryResult::RetryableErr(anyhow!(e)))
+            (CommitState::Known(table), RetryResult::FatalErr(anyhow!(e)))
         }
-        Ok(new_table) => (new_table, RetryResult::Ok(())),
+        Ok(new_table) => (CommitState::Known(new_table), RetryResult::Ok(())),
     }
 }
 
@@ -2538,9 +2629,9 @@ fn commit_to_iceberg<'scope>(
                         ("mz-sink-version".to_string(), sink_version.to_string()),
                     ];
 
-                    let (table_state, commit_result) = Retry::default()
+                    let (commit_state, commit_result) = Retry::default()
                         .max_tries(5)
-                        .retry_async_with_state(table, |_, table| {
+                        .retry_async_with_state(CommitState::Known(table), |_, commit_state| {
                             let snapshot_properties = snapshot_properties.clone();
                             let data_files = data_files.clone();
                             let delete_files = delete_files.clone();
@@ -2553,7 +2644,7 @@ fn commit_to_iceberg<'scope>(
                             let batch_upper = batch.1.clone();
                             async move {
                                 try_commit_batch(
-                                    table,
+                                    commit_state,
                                     snapshot_properties,
                                     data_files,
                                     delete_files,
@@ -2576,7 +2667,10 @@ fn commit_to_iceberg<'scope>(
                             connection.namespace, connection.table
                         )
                     });
-                    table = table_state;
+                    // An unresolved state only survives the retry loop when the commit failed, in
+                    // which case `commit_result?` below tears the sink down without reusing this
+                    // table as a base.
+                    table = commit_state.into_table();
                     let duration = instant.elapsed();
                     metrics
                         .commit_duration_seconds