prevent problem with same request in one transaction pack

don't apply actions for backup and sharing in case same request

Author: ivanmorozov333

ydb/core/tx/columnshard/bg_tasks/manager/manager.cpp

@@ -87,4 +87,8 @@ std::unique_ptr<NKikimr::NTabletFlatExecutor::ITransaction> TSessionsManager::Tx
     return std::make_unique<TTxRemoveSession>(className, identifier, Adapter, Storage);
 }
 
+bool TSessionsManager::HasTask(const TTask& task) const {
+    return Storage->GetSession(task.GetDescriptionContainer().GetClassName(), task.GetIdentifier());
+}
+
 }

ydb/core/tx/columnshard/bg_tasks/manager/manager.h

@@ -39,6 +39,8 @@ class TSessionsManager {
     [[nodiscard]] std::unique_ptr<NTabletFlatExecutor::ITransaction> TxApplyControlFromProto(const NKikimrTxBackgroundProto::TSessionControlContainer& controlProto);
     [[nodiscard]] std::unique_ptr<NTabletFlatExecutor::ITransaction> TxApplyControl(const TSessionControlContainer& control);
 
+    bool HasTask(const TTask& task) const;
+
     void Start() {
         AFL_VERIFY(!Finished);
         AFL_VERIFY(!Started);

ydb/core/tx/columnshard/columnshard__propose_transaction.cpp

@@ -69,15 +69,17 @@ class TTxProposeTransaction : public NTabletFlatExecutor::TTransactionBase<TColu
         AFL_VERIFY(!!TxOperator);
         const ui64 txId = record.GetTxId();
         NActors::TLogContextGuard lGuard = NActors::TLogContextBuilder::Build()("tablet_id", Self->TabletID())("tx_id", txId)("this", (ui64)this);
-        if (Ev->Sender != TxOperator->GetTxInfo().Source) {
-            return;
-        }
         if (TxOperator->IsFail()) {
             TxOperator->SendReply(*Self, ctx);
-        } else if (TxOperator->IsAsync()) {
-            Self->GetProgressTxController().StartProposeOnComplete(txId, ctx);
         } else {
-            Self->GetProgressTxController().FinishProposeOnComplete(txId, ctx);
+            if (!Self->GetProgressTxController()->IsActualOperator(TxOperator)) {
+                return;
+            }
+            if (TxOperator->IsAsync()) {
+                Self->GetProgressTxController().StartProposeOnComplete(txId, ctx);
+            } else {
+                Self->GetProgressTxController().FinishProposeOnComplete(txId, ctx);
+            }
         }
 
         Self->TryRegisterMediatorTimeCast();

ydb/core/tx/columnshard/transactions/operators/backup.cpp

@@ -31,24 +31,32 @@ bool TBackupTransactionOperator::DoParse(TColumnShard& owner, const TString& dat
     NArrow::NSerialization::TSerializerContainer serializer(std::make_shared<NArrow::NSerialization::TNativeSerializer>());
     ExportTask = std::make_shared<NOlap::NExport::TExportTask>(id.DetachResult(), selector.DetachResult(), storeInitializer.DetachResult(), serializer, GetTxId());
     NOlap::NBackground::TTask task(::ToString(ExportTask->GetIdentifier().GetPathId()), std::make_shared<NOlap::NBackground::TFakeStatusChannel>(), ExportTask);
-    TxAddTask = owner.GetBackgroundSessionsManager()->TxAddTask(task);
-    if (!TxAddTask) {
-        AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "cannot_add_task");
-        return false;
+    if (!owner.GetBackgroundSessionsManager()->HasTask(task)) {
+        TxAddTask = owner.GetBackgroundSessionsManager()->TxAddTask(task);
+        if (!TxAddTask) {
+            AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "cannot_add_task");
+            return false;
+        }
+    } else {
+        TaskExists = true;
     }
     return true;
 }
 
 TBackupTransactionOperator::TProposeResult TBackupTransactionOperator::DoStartProposeOnExecute(TColumnShard& /*owner*/, NTabletFlatExecutor::TTransactionContext& txc) {
-    AFL_VERIFY(!!TxAddTask);
-    AFL_VERIFY(TxAddTask->Execute(txc, NActors::TActivationContext::AsActorContext()));
+    if (!TaskExists) {
+        AFL_VERIFY(!!TxAddTask);
+        AFL_VERIFY(TxAddTask->Execute(txc, NActors::TActivationContext::AsActorContext()));
+    }
     return TProposeResult();
 }
 
 void TBackupTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& ctx) {
-    AFL_VERIFY(!!TxAddTask);
-    TxAddTask->Complete(ctx);
-    TxAddTask.reset();
+    if (!TaskExists) {
+        AFL_VERIFY(!!TxAddTask);
+        TxAddTask->Complete(ctx);
+        TxAddTask.reset();
+    }
 }
 
 bool TBackupTransactionOperator::ExecuteOnProgress(TColumnShard& /*owner*/, const NOlap::TSnapshot& /*version*/, NTabletFlatExecutor::TTransactionContext& /*txc*/) {

ydb/core/tx/columnshard/transactions/operators/backup.h

@@ -11,6 +11,7 @@ class TBackupTransactionOperator: public IProposeTxOperator {
     using TBase = IProposeTxOperator;
 
     std::shared_ptr<NOlap::NExport::TExportTask> ExportTask;
+    bool TaskExists = false;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxAddTask;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxConfirm;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxAbort;

ydb/core/tx/columnshard/transactions/operators/sharing.cpp

@@ -23,6 +23,7 @@ bool TSharingTransactionOperator::DoParse(TColumnShard& owner, const TString& da
 
     auto currentSession = SharingSessionsManager->GetDestinationSession(SharingTask->GetSessionId());
     if (currentSession) {
+        SessionExistsFlag = true;
         SharingTask = currentSession;
     } else {
         SharingTask->Confirm();
@@ -34,14 +35,18 @@ bool TSharingTransactionOperator::DoParse(TColumnShard& owner, const TString& da
 }
 
 TSharingTransactionOperator::TProposeResult TSharingTransactionOperator::DoStartProposeOnExecute(TColumnShard& /*owner*/, NTabletFlatExecutor::TTransactionContext& txc) {
-    AFL_VERIFY(!!TxPropose);
-    AFL_VERIFY(TxPropose->Execute(txc, NActors::TActivationContext::AsActorContext()));
+    if (!SessionExistsFlag) {
+        AFL_VERIFY(!!TxPropose);
+        AFL_VERIFY(TxPropose->Execute(txc, NActors::TActivationContext::AsActorContext()));
+    }
     return TProposeResult();
 }
 
 void TSharingTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& ctx) {
-    AFL_VERIFY(!!TxPropose);
-    TxPropose->Complete(ctx);
+    if (!SessionExistsFlag) {
+        AFL_VERIFY(!!TxPropose);
+        TxPropose->Complete(ctx);
+    }
     TxPropose.release();
 }
 

ydb/core/tx/columnshard/transactions/operators/sharing.h

@@ -12,6 +12,7 @@ class TSharingTransactionOperator: public IProposeTxOperator {
 
     std::shared_ptr<NOlap::NDataSharing::TSessionsManager> SharingSessionsManager;
     std::shared_ptr<NOlap::NDataSharing::TDestinationSession> SharingTask;
+    bool SessionExistsFlag = false;
     using TProposeResult = TTxController::TProposeResult;
     mutable std::unique_ptr<NTabletFlatExecutor::ITransaction> TxPropose;
     mutable std::unique_ptr<NTabletFlatExecutor::ITransaction> TxConfirm;

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

@@ -332,6 +332,14 @@ class TTxController {
     ui64 GetMemoryUsage() const;
     bool HaveOutdatedTxs() const;
 
+    bool IsActualOperator(const std::shared_ptr<TTxController::ITransactionOperator>& op) const {
+        auto opActual = GetTxOperator(op->GetTxId());
+        if (!opActual || (ui64)opActual.get() != (ui64)op.get()) {
+            return false;
+        }
+        return true;
+    }
+
     bool Load(NTabletFlatExecutor::TTransactionContext& txc);
 
     [[nodiscard]] std::shared_ptr<TTxController::ITransactionOperator> UpdateTxSourceInfo(const TFullTxInfo& tx, NTabletFlatExecutor::TTransactionContext& txc);