Clean plan queue item (ydb-platform#3849)

Author: ivanmorozov333

ydb/core/tx/columnshard/columnshard__progress_tx.cpp

@@ -30,6 +30,7 @@ class TColumnShard::TTxProgressTx : public TTransactionBase<TColumnShard> {
         // Process a single transaction at the front of the queue
         auto plannedItem = Self->ProgressTxController->StartPlannedTx();
         if (!!plannedItem) {
+            PlannedQueueItem.emplace(plannedItem->PlanStep, plannedItem->TxId);
             ui64 step = plannedItem->PlanStep;
             ui64 txId = plannedItem->TxId;
             LastCompletedTx = NOlap::TSnapshot(step, txId);
@@ -57,6 +58,9 @@ class TColumnShard::TTxProgressTx : public TTransactionBase<TColumnShard> {
         if (TxOperator) {
             TxOperator->Complete(*Self, ctx);
         }
+        if (PlannedQueueItem) {
+            Self->GetProgressTxController().CompleteRunningTx(*PlannedQueueItem);
+        }
         if (LastCompletedTx) {
             Self->LastCompletedTx = std::max(*LastCompletedTx, Self->LastCompletedTx);
         }
@@ -67,6 +71,7 @@ class TColumnShard::TTxProgressTx : public TTransactionBase<TColumnShard> {
     TTxController::ITransactionOperatior::TPtr TxOperator;
     const ui32 TabletTxNo;
     std::optional<NOlap::TSnapshot> LastCompletedTx;
+    std::optional<TTxController::TPlanQueueItem> PlannedQueueItem;
 };
 
 void TColumnShard::EnqueueProgressTx(const TActorContext& ctx) {

ydb/core/tx/columnshard/transactions/tx_controller.cpp

@@ -195,7 +195,7 @@ void TTxController::FinishPlannedTx(const ui64 txId, NTabletFlatExecutor::TTrans
 }
 
 void TTxController::CompleteRunningTx(const TPlanQueueItem& txItem) {
-    RunningQueue.erase(txItem);
+    AFL_VERIFY(RunningQueue.erase(txItem))("info", txItem.DebugString());
 }
 
 std::optional<TTxController::TPlanQueueItem> TTxController::GetPlannedTx() const {

ydb/core/tx/columnshard/transactions/tx_controller.h

@@ -24,6 +24,10 @@ class TTxController {
         inline bool operator<(const TPlanQueueItem& rhs) const {
             return Step < rhs.Step || (Step == rhs.Step && TxId < rhs.TxId);
         }
+
+        TString DebugString() const {
+            return TStringBuilder() << "step=" << Step << ";txId=" << TxId << ";";
+        }
     };
 
     struct TBasicTxInfo {