fix: enforce pending update lifecycle

Pending updates need deterministic ownership and finalization across both standalone commits and explicit transactions. Register updates through shared ownership so transactions can retain and finalize them safely, and scope the temporary transaction binding used by standalone commits so it never escapes through TransactionContext.

Finalize no-op commits and failed applies so every update reaches a terminal state and staged files are cleaned even when the caller never commits the transaction. During Transaction::Commit retries, defer eager finalization while updates are being reapplied; otherwise a retryable validation error would finalize the transaction and destroy staged state before RetryRunner can retry. Restore the retry lifecycle marker with RAII when commit exits or throws.

Convert snapshot update factories and test helpers to shared_ptr to satisfy the ownership contract.

Tests cover detached temporary transactions, cleared standalone bindings, rejection of unshared standalone updates, no-op and apply-failure finalization, staged-file cleanup, standalone retry reapplication, and lifecycle restoration after commit exceptions.

Author: zhjwpku

src/iceberg/test/fast_append_test.cc

@@ -315,6 +315,26 @@ TEST_F(FastAppendTest, FinalizeIgnoresCleanupDeleteFailure) {
               IsOk());
 }
 
+TEST_F(FastAppendTest, TransactionApplyFailureCleansUpStagedFiles) {
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto fast_append, txn->NewFastAppend());
+  std::vector<std::string> deleted_paths;
+  fast_append->DeleteWith([&](const std::string& path) {
+    deleted_paths.push_back(path);
+    return file_io_->DeleteFile(path);
+  });
+  fast_append->AppendFile(file_a_);
+
+  EXPECT_THAT(static_cast<SnapshotUpdate&>(*fast_append).Apply(), IsOk());
+  fast_append->AppendFile(nullptr);
+
+  EXPECT_THAT(fast_append->Commit(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(deleted_paths, ::testing::SizeIs(2U));
+  EXPECT_THAT(txn->Commit(),
+              ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+                               HasErrorMessage("Transaction already finalized")));
+}
+
 TEST_F(FastAppendTest, RetryCopiesAppendManifestAgain) {
   table_->metadata()->format_version = 1;
   const auto path = table_location_ + "/metadata/input.avro";

src/iceberg/test/merging_snapshot_update_test.cc

@@ -81,11 +81,11 @@ class MergingSnapshotCapturingReporter final : public MetricsReporter {
 /// \brief Concrete subclass of MergingSnapshotUpdate for testing.
 class TestMergeAppend : public MergingSnapshotUpdate {
  public:
-  static Result<std::unique_ptr<TestMergeAppend>> Make(std::string table_name,
+  static Result<std::shared_ptr<TestMergeAppend>> Make(std::string table_name,
                                                        std::shared_ptr<Table> table) {
     ICEBERG_ASSIGN_OR_RAISE(
         auto ctx, TransactionContext::Make(std::move(table), TransactionKind::kUpdate));
-    return std::unique_ptr<TestMergeAppend>(
+    return std::shared_ptr<TestMergeAppend>(
         new TestMergeAppend(std::move(table_name), std::move(ctx)));
   }
 
@@ -231,11 +231,11 @@ class TestMergeAppend : public MergingSnapshotUpdate {
 
 class TestOverwriteUpdate : public MergingSnapshotUpdate {
  public:
-  static Result<std::unique_ptr<TestOverwriteUpdate>> Make(std::string table_name,
+  static Result<std::shared_ptr<TestOverwriteUpdate>> Make(std::string table_name,
                                                            std::shared_ptr<Table> table) {
     ICEBERG_ASSIGN_OR_RAISE(
         auto ctx, TransactionContext::Make(std::move(table), TransactionKind::kUpdate));
-    return std::unique_ptr<TestOverwriteUpdate>(
+    return std::shared_ptr<TestOverwriteUpdate>(
         new TestOverwriteUpdate(std::move(table_name), std::move(ctx)));
   }
 
@@ -335,11 +335,11 @@ class MergingSnapshotUpdateTest : public MinimalUpdateTestBase {
     return f;
   }
 
-  Result<std::unique_ptr<TestMergeAppend>> NewMergeAppend() {
+  Result<std::shared_ptr<TestMergeAppend>> NewMergeAppend() {
     return TestMergeAppend::Make(TableName(), table_);
   }
 
-  Result<std::unique_ptr<TestOverwriteUpdate>> NewOverwriteUpdate() {
+  Result<std::shared_ptr<TestOverwriteUpdate>> NewOverwriteUpdate() {
     return TestOverwriteUpdate::Make(TableName(), table_);
   }
 
@@ -1101,7 +1101,7 @@ class MergingSnapshotUpdateV1Test : public UpdateTestBase {
     return f;
   }
 
-  Result<std::unique_ptr<TestMergeAppend>> NewMergeAppend() {
+  Result<std::shared_ptr<TestMergeAppend>> NewMergeAppend() {
     return TestMergeAppend::Make(TableName(), table_);
   }
 

src/iceberg/test/replace_partitions_test.cc

@@ -216,7 +216,7 @@ class ReplacePartitionsTest : public UpdateTestBase {
     return paths;
   }
 
-  Result<std::unique_ptr<ReplacePartitions>> NewReplace() {
+  Result<std::shared_ptr<ReplacePartitions>> NewReplace() {
     ICEBERG_ASSIGN_OR_RAISE(auto ctx,
                             TransactionContext::Make(table_, TransactionKind::kUpdate));
     return ReplacePartitions::Make(TableName(), std::move(ctx));

src/iceberg/test/transaction_test.cc

@@ -19,20 +19,38 @@
 
 #include "iceberg/transaction.h"
 
+#include <format>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
 #include "iceberg/expression/expressions.h"
 #include "iceberg/expression/term.h"
 #include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
 #include "iceberg/test/update_test_base.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
+#include "iceberg/update/fast_append.h"
+#include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/update_properties.h"
 #include "iceberg/update/update_schema.h"
 #include "iceberg/update/update_sort_order.h"
 
 namespace iceberg {
 
+class TestPendingUpdate final : public PendingUpdate {
+ public:
+  explicit TestPendingUpdate(std::shared_ptr<TransactionContext> ctx)
+      : PendingUpdate(std::move(ctx)) {}
+
+  Kind kind() const override { return Kind::kUpdateProperties; }
+  bool IsRetryable() const override { return true; }
+};
+
 class TransactionTest : public UpdateTestBase {};
 
 TEST_F(TransactionTest, CreateTransaction) {
@@ -46,6 +64,58 @@ TEST_F(TransactionTest, CommitEmptyTransaction) {
   EXPECT_THAT(txn->Commit(), IsOk());
 }
 
+TEST_F(TransactionTest, TemporaryTransactionDoesNotAttachToContext) {
+  ICEBERG_UNWRAP_OR_FAIL(auto ctx,
+                         TransactionContext::Make(table_, TransactionKind::kUpdate));
+  ASSERT_FALSE(ctx->transaction.has_value());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, Transaction::Make(ctx));
+
+  EXPECT_NE(txn, nullptr);
+  EXPECT_FALSE(ctx->transaction.has_value());
+}
+
+TEST_F(TransactionTest, StandaloneCommitRequiresSharedOwnership) {
+  ICEBERG_UNWRAP_OR_FAIL(auto ctx,
+                         TransactionContext::Make(table_, TransactionKind::kUpdate));
+  auto update = std::make_unique<TestPendingUpdate>(ctx);
+
+  EXPECT_THAT(update->Commit(),
+              ::testing::AllOf(
+                  IsError(ErrorKind::kInvalidArgument),
+                  HasErrorMessage("PendingUpdate must be owned by std::shared_ptr")));
+  EXPECT_FALSE(ctx->transaction.has_value());
+}
+
+TEST_F(TransactionTest, StandaloneCommitClearsTemporaryTransactionBinding) {
+  ICEBERG_UNWRAP_OR_FAIL(auto ctx,
+                         TransactionContext::Make(table_, TransactionKind::kUpdate));
+  ICEBERG_UNWRAP_OR_FAIL(auto update, UpdateProperties::Make(ctx));
+  update->Set("standalone.property", "standalone.value");
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_FALSE(ctx->transaction.has_value());
+}
+
+TEST_F(TransactionTest, CommitNoOpUpdate) {
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewSetSnapshot());
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(txn->Commit(), IsOk());
+}
+
+TEST_F(TransactionTest, ApplyFailureFinalizesTransaction) {
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewFastAppend());
+  update->AppendFile(nullptr);
+
+  EXPECT_THAT(update->Commit(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(txn->Commit(),
+              ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+                               HasErrorMessage("Transaction already finalized")));
+}
+
 TEST_F(TransactionTest, CommitTransactionWithPropertyUpdate) {
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
   ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties());
@@ -151,6 +221,39 @@ TEST_F(TransactionRetryTest, CommitRetrySucceedsAfterConflict) {
   EXPECT_EQ(update_call_count, 2);
 }
 
+TEST_F(TransactionRetryTest, StandaloneCommitRetryReappliesUpdate) {
+  std::vector<size_t> update_counts;
+  ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [this, &update_counts](const TableIdentifier&,
+                                 const std::vector<std::unique_ptr<TableRequirement>>&,
+                                 const std::vector<std::unique_ptr<TableUpdate>>& updates)
+              -> Result<std::shared_ptr<Table>> {
+            update_counts.push_back(updates.size());
+            if (update_counts.size() == 1) {
+              return CommitFailed("conflict on first attempt");
+            }
+            return Table::Make(mock_table_->name(), mock_table_->metadata(),
+                               std::string(mock_table_->metadata_file_location()),
+                               mock_table_->io(), mock_catalog_);
+          });
+  EXPECT_CALL(*mock_catalog_, LoadTable(::testing::_))
+      .WillOnce([this](const TableIdentifier&) -> Result<std::shared_ptr<Table>> {
+        auto builder = TableMetadataBuilder::BuildFrom(mock_table_->metadata().get());
+        ICEBERG_ASSIGN_OR_RAISE(auto metadata, builder->Build());
+        return Table::Make(
+            mock_table_->name(), std::shared_ptr<TableMetadata>(std::move(metadata)),
+            std::format("{}.refreshed", mock_table_->metadata_file_location()),
+            mock_table_->io(), mock_catalog_);
+      });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, mock_table_->NewUpdateProperties());
+  update->Set("retry.test", "value");
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(update_counts, ::testing::ElementsAre(1U, 1U));
+}
+
 TEST_F(TransactionRetryTest, CommitRetryExhausted) {
   int update_call_count = 0;
   ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
@@ -195,6 +298,34 @@ TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) {
   EXPECT_EQ(update_call_count, 1);  // Should not retry
 }
 
+TEST_F(TransactionRetryTest, CommitExceptionRestoresLifecycleState) {
+  int update_call_count = 0;
+  ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [&update_call_count](const TableIdentifier&,
+                               const std::vector<std::unique_ptr<TableRequirement>>&,
+                               const std::vector<std::unique_ptr<TableUpdate>>&)
+              -> Result<std::shared_ptr<Table>> {
+            ++update_call_count;
+            throw std::runtime_error("injected catalog failure");
+          });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, mock_table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto properties, txn->NewUpdateProperties());
+  properties->Set("exception.test", "value");
+  EXPECT_THAT(properties->Commit(), IsOk());
+
+  EXPECT_THROW(std::ignore = txn->Commit(), std::runtime_error);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
+  append->AppendFile(nullptr);
+  EXPECT_THAT(append->Commit(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(txn->Commit(),
+              ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+                               HasErrorMessage("Transaction already finalized")));
+  EXPECT_EQ(update_call_count, 1);
+}
+
 TEST_F(TransactionRetryTest, CreateTransactionDoesNotRetry) {
   int update_call_count = 0;
   ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))

src/iceberg/transaction.cc

@@ -57,6 +57,21 @@
 #include "iceberg/util/retry_util.h"
 
 namespace iceberg {
+namespace {
+
+class ScopedTrue {
+ public:
+  explicit ScopedTrue(bool& value) : value_(value) { value_ = true; }
+  ~ScopedTrue() { value_ = false; }
+
+  ScopedTrue(const ScopedTrue&) = delete;
+  ScopedTrue& operator=(const ScopedTrue&) = delete;
+
+ private:
+  bool& value_;
+};
+
+}  // namespace
 
 // ---------------------------------------------------------------------------
 // TransactionContext
@@ -122,9 +137,7 @@ Result<std::shared_ptr<Transaction>> Transaction::Make(std::shared_ptr<Table> ta
 Result<std::shared_ptr<Transaction>> Transaction::Make(
     std::shared_ptr<TransactionContext> ctx) {
   ICEBERG_PRECHECK(ctx != nullptr, "TransactionContext cannot be null");
-  auto txn = std::shared_ptr<Transaction>(new Transaction(ctx));
-  ctx->transaction = std::weak_ptr<Transaction>(txn);
-  return txn;
+  return std::shared_ptr<Transaction>(new Transaction(std::move(ctx)));
 }
 
 const std::shared_ptr<Table>& Transaction::table() const { return ctx_->table; }
@@ -138,6 +151,8 @@ std::string Transaction::MetadataFileLocation(std::string_view filename) const {
 }
 
 Status Transaction::AddUpdate(const std::shared_ptr<PendingUpdate>& update) {
+  ICEBERG_CHECK(!committed_, "Cannot add update to a committed transaction");
+  ICEBERG_CHECK(!finalized_, "Cannot add update to a finalized transaction");
   ICEBERG_CHECK(last_update_committed_,
                 "Cannot add update when previous update is not committed");
 
@@ -147,6 +162,9 @@ Status Transaction::AddUpdate(const std::shared_ptr<PendingUpdate>& update) {
 }
 
 Status Transaction::Apply(PendingUpdate& update) {
+  ICEBERG_CHECK(!committed_, "Cannot apply update to a committed transaction");
+  ICEBERG_CHECK(!finalized_, "Cannot apply update to a finalized transaction");
+
   switch (update.kind()) {
     case PendingUpdate::Kind::kExpireSnapshots:
       ICEBERG_RETURN_UNEXPECTED(
@@ -358,40 +376,37 @@ Status Transaction::ApplyUpdatePartitionStatistics(UpdatePartitionStatistics& up
 
 Result<std::shared_ptr<Table>> Transaction::Commit() {
   ICEBERG_CHECK(!committed_, "Transaction already committed");
+  ICEBERG_CHECK(!finalized_, "Transaction already finalized");
   ICEBERG_CHECK(last_update_committed_,
                 "Cannot commit transaction when previous update is not committed");
 
   const auto& updates = ctx_->metadata_builder->changes();
-  if (updates.empty()) {
-    committed_ = true;
-    return ctx_->table;
+  Result<std::shared_ptr<Table>> commit_result = ctx_->table;
+  if (!updates.empty()) {
+    const auto& props = ctx_->table->properties();
+    int32_t num_retries =
+        CanRetry() ? static_cast<int32_t>(props.Get(TableProperties::kCommitNumRetries))
+                   : 0;
+    int32_t min_wait_ms = props.Get(TableProperties::kCommitMinRetryWaitMs);
+    int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs);
+    int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs);
+
+    bool is_first_attempt = true;
+    ScopedTrue committing(committing_);
+    commit_result =
+        MakeCommitRetryRunner(num_retries, min_wait_ms, max_wait_ms, total_timeout_ms)
+            .Run([this, &is_first_attempt]() -> Result<std::shared_ptr<Table>> {
+              auto result = CommitOnce(is_first_attempt);
+              is_first_attempt = false;
+              return result;
+            });
   }
 
-  const auto& props = ctx_->table->properties();
-  int32_t num_retries =
-      CanRetry() ? static_cast<int32_t>(props.Get(TableProperties::kCommitNumRetries))
-                 : 0;
-  int32_t min_wait_ms = props.Get(TableProperties::kCommitMinRetryWaitMs);
-  int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs);
-  int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs);
-
-  bool is_first_attempt = true;
-  auto commit_result =
-      MakeCommitRetryRunner(num_retries, min_wait_ms, max_wait_ms, total_timeout_ms)
-          .Run([this, &is_first_attempt]() -> Result<std::shared_ptr<Table>> {
-            auto result = CommitOnce(is_first_attempt);
-            is_first_attempt = false;
-            return result;
-          });
-
   Result<const TableMetadata*> finalize_result =
       commit_result.has_value()
           ? Result<const TableMetadata*>(commit_result.value()->metadata().get())
           : std::unexpected(commit_result.error());
-
-  for (const auto& update : pending_updates_) {
-    std::ignore = update->Finalize(finalize_result);
-  }
+  FinalizeUpdates(finalize_result);
 
   ICEBERG_RETURN_UNEXPECTED(commit_result);
 
@@ -402,6 +417,16 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
   return ctx_->table;
 }
 
+void Transaction::FinalizeUpdates(const Result<const TableMetadata*>& commit_result) {
+  if (finalized_) {
+    return;
+  }
+  finalized_ = true;
+  for (const auto& update : pending_updates_) {
+    std::ignore = update->Finalize(commit_result);
+  }
+}
+
 Result<std::shared_ptr<Table>> Transaction::CommitOnce(bool is_first_attempt) {
   std::vector<std::unique_ptr<TableRequirement>> requirements;