Fix.

src/storage/txn/txn.cpp

@@ -687,11 +687,12 @@ void Txn::Rollback() {
     LOG_TRACE(fmt::format("Txn: {} is dropped.", txn_context_ptr_->txn_id_));
 }
 
-void Txn::PutDeltaOps() {
+SharedPtr<AddDeltaEntryTask> Txn::MakeAddDeltaEntryTask() {
     if (!txn_delta_ops_entry_->operations().empty()) {
         LOG_TRACE(txn_delta_ops_entry_->ToStringSimple());
-        InfinityContext::instance().storage()->bg_processor()->Submit(MakeShared<AddDeltaEntryTask>(std::move(txn_delta_ops_entry_)));
+        return MakeShared<AddDeltaEntryTask>(std::move(txn_delta_ops_entry_));
     }
+    return nullptr;
 }
 
 void Txn::AddWalCmd(const SharedPtr<WalCmd> &cmd) {

src/storage/txn/txn.cppm

@@ -63,6 +63,7 @@ class CatalogDeltaOperation;
 class BaseTableRef;
 enum class CompactStatementType;
 struct SegmentIndexEntry;
+struct AddDeltaEntryTask;
 
 export class Txn : public EnableSharedFromThis<Txn> {
 public:
@@ -101,7 +102,7 @@ public:
 
     void Rollback();
 
-    void PutDeltaOps();
+    SharedPtr<AddDeltaEntryTask> MakeAddDeltaEntryTask();
 
     // Database OPs
     Status CreateDatabase(const SharedPtr<String> &db_name, ConflictType conflict_type, const SharedPtr<String> &comment);

src/storage/txn/txn_manager.cpp

@@ -34,6 +34,7 @@ import wal_manager;
 import defer_op;
 import infinity_context;
 import global_resource_usage;
+import bg_task;
 
 namespace infinity {
 
@@ -363,6 +364,7 @@ void TxnManager::CleanupTxn(Txn *txn, bool commit) {
                     UnrecoverableError(error_message);
                 }
             }
+            SharedPtr<AddDeltaEntryTask> add_delta_entry_task = txn->MakeAddDeltaEntryTask();
             {
                 // cleanup the txn from committing_txn and txm_map
                 auto commit_ts = txn->CommitTS();
@@ -381,13 +383,12 @@ void TxnManager::CleanupTxn(Txn *txn, bool commit) {
                     String error_message = fmt::format("Txn: {} not found in txn map", txn_id);
                     UnrecoverableError(error_message);
                 }
-
                 if (committing_txns_.empty() || committing_txns_.begin()->first > commit_ts) {
                     max_committed_ts_ = commit_ts;
                 }
-                if (commit) {
-                    txn->PutDeltaOps();
-                }
+            }
+            if (commit && add_delta_entry_task) {
+                InfinityContext::instance().storage()->bg_processor()->Submit(std::move(add_delta_entry_task));
             }
             break;
         }