feat: clean up transaction files on failed commits (#6319)

Previously, when a manifest commit failed (conflict, error, or retry
exhaustion), the `.txn` file written to `_transactions/` was left
orphaned. These files would accumulate until the GC cleanup interval (7+
days by default).

This PR adds `cleanup_transaction_file()` — a best-effort delete helper
— and calls it from all three commit failure paths
(`do_commit_new_dataset`, `do_commit_detached_transaction`,
`commit_transaction`). Failures to delete are logged as warnings and do
not surface to the caller.

In the retry loop of `commit_transaction`, the previous iteration's
transaction file is cleaned up before each retry attempt, since a new
transaction file is written on each iteration.

Fixes #6125

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: wjones127

rust/lance/src/dataset/cleanup.rs

@@ -2396,7 +2396,8 @@ mod tests {
         // deposited a data file
         assert_eq!(before_count.num_data_files, 2);
         assert_eq!(before_count.num_manifest_files, 1);
-        assert_eq!(before_count.num_tx_files, 2);
+        // Only 1 txn file: the failed commit's txn file was already cleaned up.
+        assert_eq!(before_count.num_tx_files, 1);
 
         // All of our manifests are newer than the threshold but temp files
         // should still be deleted.

rust/lance/src/io/commit.rs

@@ -86,6 +86,29 @@ pub(crate) async fn read_transaction_file(
     transaction.try_into()
 }
 
+/// Best-effort delete of a transaction file that is no longer needed.
+///
+/// Logs a warning on failure rather than propagating the error, since the
+/// primary operation has already failed and the orphaned file will eventually
+/// be removed by GC.
+async fn cleanup_transaction_file(
+    object_store: &ObjectStore,
+    base_path: &Path,
+    transaction_file: &str,
+) {
+    if transaction_file.is_empty() {
+        return;
+    }
+    let path = base_path.child(TRANSACTIONS_DIR).child(transaction_file);
+    if let Err(e) = object_store.delete(&path).await {
+        log::warn!(
+            "Failed to clean up orphaned transaction file '{}': {}",
+            transaction_file,
+            e
+        );
+    }
+}
+
 /// Write a transaction to a file and return the relative path.
 pub(crate) async fn write_transaction_file(
     object_store: &ObjectStore,
@@ -153,7 +176,7 @@ async fn do_commit_new_dataset(
                 ref_path.clone(),
                 new_base_id,
                 branch_name.clone(),
-                transaction_file,
+                transaction_file.clone(),
             );
 
             let updated_indices = if let Some(index_section_pos) = source_manifest.index_section {
@@ -252,9 +275,13 @@ async fn do_commit_new_dataset(
             Ok((manifest, manifest_location))
         }
         Err(CommitError::CommitConflict) => {
+            cleanup_transaction_file(object_store, base_path, &transaction_file).await;
             Err(crate::Error::dataset_already_exists(base_path.to_string()))
         }
-        Err(CommitError::OtherError(err)) => Err(err),
+        Err(CommitError::OtherError(err)) => {
+            cleanup_transaction_file(object_store, base_path, &transaction_file).await;
+            Err(err)
+        }
     }
 }
 
@@ -812,13 +839,15 @@ pub(crate) async fn do_commit_detached_transaction(
             }
             Err(CommitError::OtherError(err)) => {
                 // If other error, return
+                cleanup_transaction_file(object_store, &dataset.base, &transaction_file).await;
                 return Err(err);
             }
         }
     }
 
     // This should be extremely unlikely.  There should not be *that* many detached commits.  If
     // this happens then it seems more likely there is a bug in our random u64 generation.
+    cleanup_transaction_file(object_store, &dataset.base, &transaction_file).await;
     Err(crate::Error::commit_conflict_source(
         0,
         format!(
@@ -904,6 +933,9 @@ pub(crate) async fn commit_transaction(
     // Other transactions that may have been committed since the read_version.
     // We keep pair of (version, transaction). No other transactions to check initially
     let mut other_transactions: Vec<(u64, Arc<Transaction>)>;
+    // Track the transaction file written in the current loop iteration so we can
+    // delete it if the commit ultimately fails.
+    let mut current_transaction_file = String::new();
 
     while backoff.attempt() < num_attempts {
         // We are pessimistic here and assume there may be other transactions
@@ -931,11 +963,12 @@ pub(crate) async fn commit_transaction(
             transaction = rebase.finish(&dataset).await?;
         }
 
-        let transaction_file = if !write_config.disable_transaction_file() {
+        current_transaction_file = if !write_config.disable_transaction_file() {
             write_transaction_file(object_store, &dataset.base, &transaction).await?
         } else {
             String::new()
         };
+        let transaction_file = current_transaction_file.as_str();
 
         target_version = dataset.manifest.version + 1;
         if is_detached_version(target_version) {
@@ -952,15 +985,15 @@ pub(crate) async fn commit_transaction(
                     &dataset.base,
                     version,
                     write_config,
-                    &transaction_file,
+                    transaction_file,
                     &dataset.manifest,
                 )
                 .await?
             }
             _ => transaction.build_manifest(
                 Some(dataset.manifest.as_ref()),
                 dataset.load_indices().await?.as_ref().clone(),
-                &transaction_file,
+                transaction_file,
                 write_config,
             )?,
         };
@@ -1053,19 +1086,29 @@ pub(crate) async fn commit_transaction(
                 }
 
                 if next_attempt_i < num_attempts {
+                    // The transaction file from this attempt is now stale; clean it up
+                    // before the next attempt writes a new one (possibly rebased).
+                    cleanup_transaction_file(
+                        object_store,
+                        &dataset.base,
+                        &current_transaction_file,
+                    )
+                    .await;
                     tokio::time::sleep(backoff.next_backoff()).await;
                     continue;
                 } else {
                     break;
                 }
             }
             Err(CommitError::OtherError(err)) => {
-                // If other error, return
+                cleanup_transaction_file(object_store, &dataset.base, &current_transaction_file)
+                    .await;
                 return Err(err);
             }
         }
     }
 
+    cleanup_transaction_file(object_store, &dataset.base, &current_transaction_file).await;
     Err(crate::Error::commit_conflict_source(
         target_version,
         format!(
@@ -1092,7 +1135,7 @@ mod tests {
     use lance_linalg::distance::MetricType;
     use lance_table::format::{DataFile, DataStorageFormat};
     use lance_table::io::commit::{
-        CommitLease, CommitLock, RenameCommitHandler, UnsafeCommitHandler,
+        CommitLease, CommitLock, ManifestWriter, RenameCommitHandler, UnsafeCommitHandler,
     };
     use lance_testing::datagen::generate_random_array;
 
@@ -1700,6 +1743,77 @@ mod tests {
         assert_eq!(manifest.fragments.as_ref(), &expected_fragments);
     }
 
+    /// A CommitHandler that always fails with OtherError, used to simulate
+    /// a manifest write failure so we can verify orphaned transaction files
+    /// are cleaned up.
+    #[derive(Debug)]
+    struct FailingCommitHandler;
+
+    #[async_trait::async_trait]
+    impl CommitHandler for FailingCommitHandler {
+        async fn commit(
+            &self,
+            _manifest: &mut Manifest,
+            _indices: Option<Vec<IndexMetadata>>,
+            _base_path: &Path,
+            _object_store: &ObjectStore,
+            _manifest_writer: ManifestWriter,
+            _naming_scheme: ManifestNamingScheme,
+            _transaction: Option<lance_table::format::Transaction>,
+        ) -> std::result::Result<ManifestLocation, CommitError> {
+            Err(CommitError::OtherError(lance_core::Error::io(
+                "simulated commit failure",
+            )))
+        }
+    }
+
+    fn count_txn_files(uri: &str) -> usize {
+        let tx_dir = std::path::Path::new(uri).join("_transactions");
+        std::fs::read_dir(&tx_dir)
+            .map(|rd| rd.filter_map(|e| e.ok()).count())
+            .unwrap_or(0)
+    }
+
+    #[tokio::test]
+    async fn test_transaction_file_cleanup_on_commit_failure() {
+        let tmp = TempStrDir::default();
+        let uri = tmp.as_str();
+
+        // Create initial dataset with a normal commit handler.
+        let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+            "x",
+            DataType::Int32,
+            false,
+        )]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+        )
+        .unwrap();
+        let reader = RecordBatchIterator::new(vec![Ok(batch.clone())], schema.clone());
+        Dataset::write(reader, uri, None).await.unwrap();
+
+        let txn_files_before = count_txn_files(uri);
+
+        // Attempt to append with a commit handler that always fails.
+        let params = WriteParams {
+            mode: WriteMode::Append,
+            commit_handler: Some(Arc::new(FailingCommitHandler)),
+            ..Default::default()
+        };
+        let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
+        let result = Dataset::write(reader, uri, Some(params)).await;
+        assert!(result.is_err(), "expected commit to fail");
+
+        // The failed commit must not leave any extra transaction files behind.
+        let txn_files_after = count_txn_files(uri);
+        assert_eq!(
+            txn_files_after,
+            txn_files_before,
+            "failed commit left {extra} orphaned transaction file(s)",
+            extra = txn_files_after.saturating_sub(txn_files_before),
+        );
+    }
     /// Helper to build a simple manifest for check_column_indices tests.
     fn make_manifest_with_file(
         schema: Schema,