fix race on enqueue tx_commit in case long synchronization (#8588)

Author: ivanmorozov333

ydb/core/tx/columnshard/blobs_action/transaction/tx_write_index.cpp

@@ -43,7 +43,7 @@ bool TTxWriteIndex::Execute(TTransactionContext& txc, const TActorContext& ctx)
         LOG_S_ERROR(TxPrefix() << " (" << changes->TypeString() << ") cannot write index blobs" << TxSuffix());
     }
 
-    Self->EnqueueProgressTx(ctx);
+    Self->EnqueueProgressTx(ctx, std::nullopt);
     return true;
 }
 

ydb/core/tx/columnshard/columnshard.cpp

@@ -61,7 +61,7 @@ void TColumnShard::SwitchToWork(const TActorContext& ctx) {
         SignalTabletActive(ctx);
         AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "initialize_shard")("step", "SignalTabletActive");
         TryRegisterMediatorTimeCast();
-        EnqueueProgressTx(ctx);
+        EnqueueProgressTx(ctx, std::nullopt);
     }
     Counters.GetCSCounters().OnIndexMetadataLimit(NOlap::IColumnEngine::GetMetadataLimit());
     EnqueueBackgroundActivities();

ydb/core/tx/columnshard/columnshard__plan_step.cpp

@@ -105,7 +105,7 @@ bool TTxPlanStep::Execute(TTransactionContext& txc, const TActorContext& ctx) {
     Self->Counters.GetTabletCounters()->IncCounter(COUNTER_PLAN_STEP_ACCEPTED);
 
     if (plannedCount > 0 || Self->ProgressTxController->HaveOutdatedTxs()) {
-        Self->EnqueueProgressTx(ctx);
+        Self->EnqueueProgressTx(ctx, std::nullopt);
     }
     return true;
 }

ydb/core/tx/columnshard/columnshard__progress_tx.cpp

@@ -48,7 +48,7 @@ class TColumnShard::TTxProgressTx: public TTransactionBase<TColumnShard> {
             TxOperator = Self->ProgressTxController->GetTxOperatorVerified(txId);
             if (auto txPrepare = TxOperator->BuildTxPrepareForProgress(Self)) {
                 AbortedThroughRemoveExpired = true;
-                Self->ProgressTxInFlight = false;
+                Self->ProgressTxInFlight = txId;
                 Self->Execute(txPrepare.release(), ctx);
                 return true;
             } else {
@@ -66,9 +66,9 @@ class TColumnShard::TTxProgressTx: public TTransactionBase<TColumnShard> {
             Self->ProgressTxController->FinishPlannedTx(txId, txc);
             Self->Counters.GetTabletCounters()->IncCounter(COUNTER_PLANNED_TX_COMPLETED);
         }
-        Self->ProgressTxInFlight = false;
+        Self->ProgressTxInFlight = std::nullopt;
         if (!!Self->ProgressTxController->GetPlannedTx()) {
-            Self->EnqueueProgressTx(ctx);
+            Self->EnqueueProgressTx(ctx, std::nullopt);
         }
         return true;
     }
@@ -93,10 +93,13 @@ class TColumnShard::TTxProgressTx: public TTransactionBase<TColumnShard> {
     }
 };
 
-void TColumnShard::EnqueueProgressTx(const TActorContext& ctx) {
+void TColumnShard::EnqueueProgressTx(const TActorContext& ctx, const std::optional<ui64> continueTxId) {
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "EnqueueProgressTx")("tablet_id", TabletID());
-    if (!ProgressTxInFlight) {
-        ProgressTxInFlight = true;
+    if (continueTxId) {
+        AFL_VERIFY(!ProgressTxInFlight || ProgressTxInFlight == continueTxId)("current", ProgressTxInFlight)("expected", continueTxId);
+    }
+    if (!ProgressTxInFlight || ProgressTxInFlight == continueTxId) {
+        ProgressTxInFlight = continueTxId.value_or(0);
         Execute(new TTxProgressTx(this), ctx);
     }
 }

ydb/core/tx/columnshard/columnshard_impl.h

@@ -468,7 +468,7 @@ class TColumnShard
     NOlap::NResourceBroker::NSubscribe::TTaskContext CompactTaskSubscription;
     NOlap::NResourceBroker::NSubscribe::TTaskContext TTLTaskSubscription;
 
-    bool ProgressTxInFlight = false;
+    std::optional<ui64> ProgressTxInFlight;
     THashMap<ui64, TInstant> ScanTxInFlight;
     THashMap<TWriteId, TLongTxWriteInfo> LongTxWrites;
     using TPartsForLTXShard = THashMap<ui32, TLongTxWriteInfo*>;
@@ -539,7 +539,7 @@ class TColumnShard
 public:
     ui64 TabletTxCounter = 0;
 
-    void EnqueueProgressTx(const TActorContext& ctx);
+    void EnqueueProgressTx(const TActorContext& ctx, const std::optional<ui64> continueTxId);
     NOlap::TSnapshot GetLastTxSnapshot() const {
         return NOlap::TSnapshot(LastPlannedStep, LastPlannedTxId);
     }

ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h

@@ -166,7 +166,7 @@ class TEvWriteCommitPrimaryTransactionOperator: public TEvWriteCommitSyncTransac
     void CheckFinished(TColumnShard& owner) {
         if (WaitShardsResultAck.empty()) {
             AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "finished");
-            owner.EnqueueProgressTx(NActors::TActivationContext::AsActorContext());
+            owner.EnqueueProgressTx(NActors::TActivationContext::AsActorContext(), GetTxId());
         }
     }
 

ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h

@@ -77,7 +77,7 @@ class TEvWriteCommitSecondaryTransactionOperator: public TEvWriteCommitSyncTrans
             op->ReceiveAck = true;
             if (!op->NeedReceiveBroken) {
                 op->TxBroken = false;
-                Self->EnqueueProgressTx(ctx);
+                Self->EnqueueProgressTx(ctx, TxId);
             }
         }
 
@@ -118,7 +118,7 @@ class TEvWriteCommitSecondaryTransactionOperator: public TEvWriteCommitSyncTrans
             if (BrokenFlag) {
                 Self->GetProgressTxController().CompleteOnCancel(TxId, ctx);
             }
-            Self->EnqueueProgressTx(ctx);
+            Self->EnqueueProgressTx(ctx, TxId);
         }
 
     public:
@@ -154,7 +154,7 @@ class TEvWriteCommitSecondaryTransactionOperator: public TEvWriteCommitSyncTrans
 
     virtual void DoOnTabletInit(TColumnShard& owner) override {
         if (TxBroken || (ReceiveAck && !NeedReceiveBroken)) {
-            owner.EnqueueProgressTx(NActors::TActivationContext::AsActorContext());
+            owner.EnqueueProgressTx(NActors::TActivationContext::AsActorContext(), GetTxId());
         } else if (!ReceiveAck) {
             SendResult(owner);
         }