fix TxProgress enqueue (#43594)

Co-authored-by: Kirill Vasilenko <[email protected]>

18 files changed, 307 insertions, 212 deletions

diff --git a/ydb/core/kqp/ut/olap/locks_ut.cpp b/ydb/core/kqp/ut/olap/locks_ut.cpp
index 37536e1123d..38a52802361 100644
--- a/ydb/core/kqp/ut/olap/locks_ut.cpp
+++ b/ydb/core/kqp/ut/olap/locks_ut.cpp
@@ -241,7 +241,7 @@ Y_UNIT_TEST_SUITE(KqpOlapLocks) {
         const auto resultSets = resultSelect.GetResultSets();
         UNIT_ASSERT_VALUES_EQUAL(resultSets.size(), 1);
         const auto resultSet = resultSets[0];
-        if (shardCount > 1 && reboot) {
+        if (reboot) {
             const auto deleteUnavailiable = resultDelete.GetStatus() == NYdb::Dev::EStatus::UNAVAILABLE;
             const auto deleteUndetermined = resultDelete.GetStatus() == NYdb::Dev::EStatus::UNDETERMINED;
             UNIT_ASSERT_C(
@@ -254,13 +254,14 @@ Y_UNIT_TEST_SUITE(KqpOlapLocks) {
                 resultDelete.GetStatus()
             );
         } else {
-            UNIT_ASSERT_VALUES_EQUAL(resultSet.RowsCount(), 0); // not need locks
+            // without a reboot the commit always goes through, so the row is deleted
+            UNIT_ASSERT_VALUES_EQUAL(resultSet.RowsCount(), 0);
         }
 
         //DELETE 0 rows from every shard
         const auto resultDelete2 =
             client.ExecuteQuery("DELETE from `/Root/ttt` WHERE id < 100", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).GetValueSync();
-        if (shardCount > 1 && reboot) {
+        if (reboot) {
             UNIT_ASSERT_C(
                 (!resultDelete2.IsSuccess() && resultSet.RowsCount() == 1) ||
 
diff --git a/ydb/core/tx/columnshard/blobs_action/transaction/tx_blobs_written.cpp b/ydb/core/tx/columnshard/blobs_action/transaction/tx_blobs_written.cpp
index de9d5dcb7f3..0f0355237e5 100644
--- a/ydb/core/tx/columnshard/blobs_action/transaction/tx_blobs_written.cpp
+++ b/ydb/core/tx/columnshard/blobs_action/transaction/tx_blobs_written.cpp
@@ -83,7 +83,7 @@ bool TTxBlobsWritingFinished::DoExecute(TTransactionContext& txc, const TActorCo
                 Self->GetOperationsManager().AddEventForLock(*Self, op->GetLockId(), evWrite);
                 // No tx writes (bulk upsert) must break decent/proper txs.
                 // Decent/proper txs asked for isolation, so we have to give them isolation.
-                Self->OperationsManager->BreakConflictingTxs(op->GetLockId(), txc);
+                Self->OperationsManager->BreakConflictingTxs(op->GetLockId());
             }
         } else {
             auto& info = Self->OperationsManager->GetLockVerified(op->GetLockId());
diff --git a/ydb/core/tx/columnshard/blobs_action/transaction/tx_write_index.cpp b/ydb/core/tx/columnshard/blobs_action/transaction/tx_write_index.cpp
index baa633ecc62..6ca37e212e2 100644
--- a/ydb/core/tx/columnshard/blobs_action/transaction/tx_write_index.cpp
+++ b/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, std::nullopt);
+    Self->EnqueueProgressTx(ctx);
     return true;
 }
 
diff --git a/ydb/core/tx/columnshard/columnshard.cpp b/ydb/core/tx/columnshard/columnshard.cpp
index a35579c4c2a..f9a8ca36748 100644
--- a/ydb/core/tx/columnshard/columnshard.cpp
+++ b/ydb/core/tx/columnshard/columnshard.cpp
@@ -82,7 +82,7 @@ void TColumnShard::TrySwitchToWork(const TActorContext& ctx) {
         SignalTabletActive(ctx);
         AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "initialize_shard")("step", "SignalTabletActive");
         TryRegisterMediatorTimeCast();
-        EnqueueProgressTx(ctx, std::nullopt);
+        EnqueueProgressTx(ctx);
         OnTieringModified();
     }
     Counters.GetCSCounters().OnIndexMetadataLimit(NOlap::IColumnEngine::GetMetadataLimit());
diff --git a/ydb/core/tx/columnshard/columnshard__notify_tx_completion.cpp b/ydb/core/tx/columnshard/columnshard__notify_tx_completion.cpp
index 71da59b6740..e5d166ab4b7 100644
--- a/ydb/core/tx/columnshard/columnshard__notify_tx_completion.cpp
+++ b/ydb/core/tx/columnshard/columnshard__notify_tx_completion.cpp
@@ -15,7 +15,7 @@ public:
         LOG_S_DEBUG("TTxNotifyTxCompletion.Execute at tablet " << Self->TabletID());
 
         const ui64 txId = Ev->Get()->Record.GetTxId();
-        auto txOperator = Self->ProgressTxController->GetTxOperatorOptional(txId);
+        auto txOperator = Self->ProgressTxController->GetTxOperator(txId, ETxOperatorStatus::Any, /*optional*/ true);
         if (txOperator) {
             txOperator->RegisterSubscriber(Ev->Sender);
             return true;
diff --git a/ydb/core/tx/columnshard/columnshard__plan_step.cpp b/ydb/core/tx/columnshard/columnshard__plan_step.cpp
index aecaad71b0c..44540f3b7d5 100644
--- a/ydb/core/tx/columnshard/columnshard__plan_step.cpp
+++ b/ydb/core/tx/columnshard/columnshard__plan_step.cpp
@@ -102,7 +102,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, std::nullopt);
+        Self->EnqueueProgressTx(ctx);
     }
     return true;
 }
diff --git a/ydb/core/tx/columnshard/columnshard__progress_tx.cpp b/ydb/core/tx/columnshard/columnshard__progress_tx.cpp
index 37efe44f750..ff25e3469d3 100644
--- a/ydb/core/tx/columnshard/columnshard__progress_tx.cpp
+++ b/ydb/core/tx/columnshard/columnshard__progress_tx.cpp
@@ -30,11 +30,8 @@ public:
 
     bool Execute(TTransactionContext& txc, const TActorContext& ctx) override {
         NActors::TLogContextGuard logGuard = NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD_TX)(
-            "tablet_id", Self->TabletID())("tx_state", "TTxProgressTx::Execute")("tx_current", Self->ProgressTxInFlight);
-        if (!Self->ProgressTxInFlight) {
-            AbortedThroughRemoveExpired = true;
-            return true;
-        }
+            "tablet_id", Self->TabletID())("tx_state", "TTxProgressTx::Execute")("tx_current", Self->InProgressTxId);
+        AFL_VERIFY(Self->ProgressTxScheduled);
         Self->Counters.GetTabletCounters()->SetCounter(COUNTER_TX_COMPLETE_LAG, Self->GetTxCompleteLag().MilliSeconds());
 
         const size_t removedCount = Self->ProgressTxController->CleanExpiredTxs();
@@ -45,6 +42,7 @@ public:
             return true;
         }
 
+        Self->ProgressTxScheduled = false;
         // Process a single transaction at the front of the queue
         const auto plannedItem = Self->ProgressTxController->GetFirstPlannedTx();
         if (!!plannedItem) {
@@ -52,19 +50,19 @@ public:
             const ui64 step = plannedItem->PlanStep;
             const ui64 txId = plannedItem->TxId;
             NActors::TLogContextGuard logGuardTx = NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD_TX)("tx_id", txId);
-            TxOperator = Self->ProgressTxController->GetTxOperatorVerified(txId);
+            TxOperator = Self->ProgressTxController->GetTxOperator(txId, ETxOperatorStatus::InProgress);
             AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "PlannedItemStart")("op_type", TxOperator->GetOpType());
             if (auto txPrepare = TxOperator->BuildTxPrepareForProgress(Self)) {
                 AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "PlannedItemStart")("details", "BuildTxPrepareForProgress")(
                     "op_type", TxOperator->GetOpType());
                 AbortedThroughRemoveExpired = true;
-                Self->ProgressTxInFlight = txId;
+                Self->InProgressTxId = txId;
                 Self->Execute(txPrepare.release(), ctx);
                 return true;
             } else if (TxOperator->IsInProgress()) {
                 AbortedThroughRemoveExpired = true;
                 AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "PlannedItemContinue")("op_type", TxOperator->GetOpType());
-                AFL_VERIFY(Self->ProgressTxInFlight == txId);
+                AFL_VERIFY(Self->InProgressTxId == txId);
                 return true;
             } else {
                 AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "PlannedItemStart")("details", "PopFirstPlannedTx")(
@@ -84,9 +82,9 @@ public:
             Self->ProgressTxController->ProgressOnExecute(txId, txc);
             Self->Counters.GetTabletCounters()->IncCounter(COUNTER_PLANNED_TX_COMPLETED);
         }
-        Self->ProgressTxInFlight = std::nullopt;
+        Self->InProgressTxId = 0;
         if (!!Self->ProgressTxController->GetPlannedTx()) {
-            Self->EnqueueProgressTx(ctx, std::nullopt);
+            Self->EnqueueProgressTx(ctx);
         }
         return true;
     }
@@ -117,15 +115,17 @@ public:
     }
 };
 
-void TColumnShard::EnqueueProgressTx(const TActorContext& ctx, const std::optional<ui64> continueTxId) {
+void TColumnShard::EnqueueProgressTx(const TActorContext& ctx, const ui64 continueTxId) {
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "EnqueueProgressTx")("tablet_id", TabletID())("tx_id", continueTxId);
-    if (continueTxId) {
-        AFL_VERIFY(!ProgressTxInFlight || ProgressTxInFlight == continueTxId)("current", ProgressTxInFlight)("expected", continueTxId);
+    // While a particular tx is in the distributed commit process, only its own nudge can advance it.
+    // So, skip everything else.
+    if (InProgressTxId != 0 && continueTxId != InProgressTxId) {
+        return;
     }
-    if (!ProgressTxInFlight || ProgressTxInFlight == continueTxId) {
+    if (!ProgressTxScheduled) {
         AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "EnqueueProgressTxStart")("tablet_id", TabletID())("tx_id", continueTxId)(
-            "tx_current", ProgressTxInFlight);
-        ProgressTxInFlight = continueTxId.value_or(0);
+            "tx_current", InProgressTxId);
+        ProgressTxScheduled = true;
         Execute(new TTxProgressTx(this), ctx);
     }
 }
diff --git a/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp b/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
index 1be790a65f7..678129ffd49 100644
--- a/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
+++ b/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
@@ -91,7 +91,7 @@ public:
             TxOperator->SendReply(*Self, ctx);
             return;
         }
-        auto internalOp = Self->GetProgressTxController().GetTxOperatorOptional(txId);
+        auto internalOp = Self->GetProgressTxController().GetTxOperator(txId, ETxOperatorStatus::InProgress, /*optional*/ true);
         if (!internalOp) {
             AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "removed tx operator");
             return;
diff --git a/ydb/core/tx/columnshard/columnshard__tx_abort.cpp b/ydb/core/tx/columnshard/columnshard__tx_abort.cpp
index 19cbb6242fe..da49af09dd5 100644
--- a/ydb/core/tx/columnshard/columnshard__tx_abort.cpp
+++ b/ydb/core/tx/columnshard/columnshard__tx_abort.cpp
@@ -14,15 +14,20 @@ public:
     bool Execute(TTransactionContext& txc, const TActorContext&) override {
         LOG_S_DEBUG("TTxTxAbort.Execute at tablet " << Self->TabletID());
 
-        auto txOperator = Self->ProgressTxController->GetTxOperatorOptional(TxId);
+        auto txOperator = Self->ProgressTxController->GetTxOperator(TxId, ETxOperatorStatus::InProgress, /*optional*/ true);
         if (txOperator) {
+            DoComplete = true;
             return txOperator->ExecuteOnAbort(*Self, txc);
         }
         return true;
     }
 
     void Complete(const TActorContext& ctx) override {
-        auto txOperator = Self->ProgressTxController->GetTxOperatorOptional(TxId);
+        if (!DoComplete) {
+            return;
+        }
+
+        auto txOperator = Self->ProgressTxController->GetTxOperator(TxId, ETxOperatorStatus::Any, /*optional*/ true);
         if (txOperator) {
             txOperator->CompleteOnAbort(*Self, ctx);
         }
@@ -47,6 +52,7 @@ public:
 
 private:
     ui64 TxId = 0;
+    bool DoComplete = false;
     TActorId Subscriber;
 };
 
diff --git a/ydb/core/tx/columnshard/columnshard_impl.cpp b/ydb/core/tx/columnshard/columnshard_impl.cpp
index e36d4dfda78..f28ca5b5cf6 100644
--- a/ydb/core/tx/columnshard/columnshard_impl.cpp
+++ b/ydb/core/tx/columnshard/columnshard_impl.cpp
@@ -1225,14 +1225,15 @@ void TColumnShard::Handle(NActors::TEvents::TEvUndelivered::TPtr& ev, const TAct
 }
 
 void TColumnShard::Handle(TEvTxProcessing::TEvReadSet::TPtr& ev, const TActorContext& ctx) {
+    auto& txController = GetProgressTxController();
     const ui64 txId = ev->Get()->Record.GetTxId();
     const ui64 tabletDest = ev->Get()->Record.GetTabletProducer();
-    if (!GetProgressTxController().GetTxOperatorOptional(txId)) {
+    auto op = txController.GetTxOperatorAs<TEvWriteCommitSyncTransactionOperator>(txId, ETxOperatorStatus::Any, /*optional*/ true);
+    if (!op) {
         AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "read_set_ignored")("proto", ev->Get()->Record.DebugString());
         TEvWriteCommitSyncTransactionOperator::SendBrokenFlagAck(*this, ev->Get()->Record.GetStep(), txId, tabletDest);
         return;
     }
-    auto op = GetProgressTxController().GetTxOperatorVerifiedAs<TEvWriteCommitSyncTransactionOperator>(txId);
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "read_set")("proto", ev->Get()->Record.DebugString())("lock_id", op->GetLockId());
     NKikimrTx::TReadSetData data;
     AFL_VERIFY(data.ParseFromArray(ev->Get()->Record.GetReadSet().data(), ev->Get()->Record.GetReadSet().size()));
@@ -1241,13 +1242,14 @@ void TColumnShard::Handle(TEvTxProcessing::TEvReadSet::TPtr& ev, const TActorCon
 }
 
 void TColumnShard::Handle(TEvTxProcessing::TEvReadSetAck::TPtr& ev, const TActorContext& ctx) {
-    auto opPtr = GetProgressTxController().GetTxOperatorOptional(ev->Get()->Record.GetTxId());
-    if (!opPtr) {
+    auto& txController = GetProgressTxController();
+    const ui64 txId = ev->Get()->Record.GetTxId();
+    auto op = txController.GetTxOperatorAs<TEvWriteCommitSyncTransactionOperator>(txId, ETxOperatorStatus::Any, /*optional*/ true);
+    if (!op) {
         AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "missed_read_set_ack")("proto", ev->Get()->Record.DebugString())(
             "tx_id", ev->Get()->Record.GetTxId());
         return;
     }
-    auto op = TValidator::CheckNotNull(dynamic_pointer_cast<TEvWriteCommitSyncTransactionOperator>(opPtr));
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "read_set_ack")("proto", ev->Get()->Record.DebugString())("lock_id", op->GetLockId());
     auto tx = op->CreateReceiveResultAckTx(*this, ev->Get()->Record.GetTabletConsumer());
     Execute(tx.release(), ctx);
diff --git a/ydb/core/tx/columnshard/columnshard_impl.h b/ydb/core/tx/columnshard/columnshard_impl.h
index 708f3044618..06b9983f309 100644
--- a/ydb/core/tx/columnshard/columnshard_impl.h
+++ b/ydb/core/tx/columnshard/columnshard_impl.h
@@ -572,7 +572,8 @@ private:
     NOlap::NResourceBroker::NSubscribe::TTaskContext CompactTaskSubscription;
     NOlap::NResourceBroker::NSubscribe::TTaskContext TTLTaskSubscription;
 
-    std::optional<ui64> ProgressTxInFlight;
+    ui64 InProgressTxId = 0;
+    bool ProgressTxScheduled = false;
     THashMap<ui64, TInstant> ScanTxInFlight;
     TMultiMap<NOlap::TSnapshot, TEvDataShard::TEvKqpScan::TPtr> WaitingScans;
     TBackgroundController BackgroundController;
@@ -685,7 +686,7 @@ public:
         return TablesManager;
     }
 
-    void EnqueueProgressTx(const TActorContext& ctx, const std::optional<ui64> continueTxId);
+    void EnqueueProgressTx(const TActorContext& ctx, const ui64 continueTxId = 0);
 
     NOlap::TSnapshot GetLastTxSnapshot() const {
         return NOlap::TSnapshot(LastPlannedStep, LastPlannedTxId);
diff --git a/ydb/core/tx/columnshard/columnshard_schema.h b/ydb/core/tx/columnshard/columnshard_schema.h
index 48e5ce6997f..a6ea272b057 100644
--- a/ydb/core/tx/columnshard/columnshard_schema.h
+++ b/ydb/core/tx/columnshard/columnshard_schema.h
@@ -491,10 +491,10 @@ struct Schema: NIceDb::Schema {
 
         struct LockId: Column<2, NScheme::NTypeIds::Uint64> {};
 
-        struct Broken: Column<3, NScheme::NTypeIds::Bool> {};
+        struct Broken_Deprecated: Column<3, NScheme::NTypeIds::Bool> {};
 
         using TKey = TableKey<TxId, LockId>;
-        using TColumns = TableColumns<TxId, LockId, Broken>;
+        using TColumns = TableColumns<TxId, LockId, Broken_Deprecated>;
     };
 
     struct TierBlobsDraft: NIceDb::Schema::Table<(ui32)ETierTables::TierBlobsDraft> {
diff --git a/ydb/core/tx/columnshard/operations/manager.cpp b/ydb/core/tx/columnshard/operations/manager.cpp
index 9d047a1ce3f..c3a7e60142f 100644
--- a/ydb/core/tx/columnshard/operations/manager.cpp
+++ b/ydb/core/tx/columnshard/operations/manager.cpp
@@ -69,15 +69,14 @@ bool TOperationsManager::Load(NTabletFlatExecutor::TTransactionContext& txc) {
         while (!rowset.EndOfSet()) {
             const ui64 lockId = rowset.GetValue<Schema::OperationTxIds::LockId>();
             const ui64 txId = rowset.GetValue<Schema::OperationTxIds::TxId>();
-            const bool broken = rowset.GetValueOrDefault<Schema::OperationTxIds::Broken>(true);
 
             auto it = LockFeatures.try_emplace(lockId, lockId, 0).first;
             auto& lock = it->second;
 
             lock.SetTxId(txId);
-            if (broken) {
-                lock.SetBroken();
-            }
+            // we cannot persist the lock state reliably and cheaply enough,
+            // so if the shard restarted/crashed/whatever, we assume the lock is broken
+            lock.SetBroken();
             AFL_VERIFY(Tx2Lock.emplace(txId, lockId).second);
             if (!rowset.Next()) {
                 return false;
@@ -88,15 +87,12 @@ bool TOperationsManager::Load(NTabletFlatExecutor::TTransactionContext& txc) {
     return true;
 }
 
-void TOperationsManager::BreakConflictingTxs(const TLockFeatures& lock, NTabletFlatExecutor::TTransactionContext& txc) {
+void TOperationsManager::BreakConflictingTxs(const TLockFeatures& lock) {
     for (auto&& lockIdToBreak : lock.GetBreakOnCommit()) {
         if (auto lockToBreak = GetLockOptional(lockIdToBreak)) {
             AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("broken_lock_id", lockIdToBreak);
             if (!lockToBreak->IsBroken()) {
                 lockToBreak->SetBroken();
-                if (lockToBreak->IsTxIdAssigned()) {
-                    PersistLock(*lockToBreak, txc);
-                }
             }
         }
     }
@@ -107,9 +103,9 @@ void TOperationsManager::BreakConflictingTxs(const TLockFeatures& lock, NTabletF
     }
 }
 
-void TOperationsManager::BreakConflictingTxs(const ui64 lockId, NTabletFlatExecutor::TTransactionContext& txc) {
+void TOperationsManager::BreakConflictingTxs(const ui64 lockId) {
     auto& lock = GetLockVerified(lockId);
-    BreakConflictingTxs(lock, txc);
+    BreakConflictingTxs(lock);
 }
 
 void TOperationsManager::CommitTransactionOnExecute(
@@ -123,7 +119,7 @@ void TOperationsManager::CommitTransactionOnExecute(
         commited.emplace_back(opPtr);
     }
 
-    BreakConflictingTxs(lock, txc);
+    BreakConflictingTxs(lock);
 
     OnTransactionFinishOnExecute(commited, lock, txId, txc);
 }
@@ -226,15 +222,9 @@ std::optional<ui64> TOperationsManager::GetLockForTx(const ui64 txId) const {
 }
 
 void TOperationsManager::LinkTransactionOnExecute(TLockFeatures& lock, NTabletFlatExecutor::TTransactionContext& txc) {
-    PersistLock(lock, txc);
-    Tx2Lock[lock.GetTxId()] = lock.GetLockId();
-}
-
-void TOperationsManager::PersistLock(TLockFeatures& lock, NTabletFlatExecutor::TTransactionContext& txc) {
     NIceDb::TNiceDb db(txc.DB);
-    db.Table<Schema::OperationTxIds>()
-        .Key(lock.GetTxId(), lock.GetLockId())
-        .Update(NIceDb::TUpdate<Schema::OperationTxIds::Broken>(lock.IsBroken()));
+    db.Table<Schema::OperationTxIds>().Key(lock.GetTxId(), lock.GetLockId()).Update();
+    Tx2Lock[lock.GetTxId()] = lock.GetLockId();
 }
 
 void TOperationsManager::LinkTransactionOnComplete(const ui64 /*lockId*/, const ui64 /*txId*/) {
diff --git a/ydb/core/tx/columnshard/operations/manager.h b/ydb/core/tx/columnshard/operations/manager.h
index 57dc61eea77..d28217a0196 100644
--- a/ydb/core/tx/columnshard/operations/manager.h
+++ b/ydb/core/tx/columnshard/operations/manager.h
@@ -249,13 +249,12 @@ public:
         TColumnShard& owner, const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc, const NOlap::TSnapshot& snapshot);
     void CommitTransactionOnComplete(TColumnShard& owner, const ui64 txId, const ui64 lockId, const NOlap::TSnapshot& snapshot);
     void LinkTransactionOnExecute(TLockFeatures& lock, NTabletFlatExecutor::TTransactionContext& txc);
-    void PersistLock(TLockFeatures& lock, NTabletFlatExecutor::TTransactionContext& txc);
     void LinkTransactionOnComplete(const ui64 lockId, const ui64 txId);
     void AbortTransactionOnExecute(TColumnShard& owner, const ui64 txId, const ui64 lockId, NTabletFlatExecutor::TTransactionContext& txc);
     void AbortTransactionOnComplete(TColumnShard& owner, const ui64 txId, const ui64 lockId);
 
-    void BreakConflictingTxs(const TLockFeatures& lock, NTabletFlatExecutor::TTransactionContext& txc);
-    void BreakConflictingTxs(const ui64 lockId, NTabletFlatExecutor::TTransactionContext& txc);
+    void BreakConflictingTxs(const TLockFeatures& lock);
+    void BreakConflictingTxs(const ui64 lockId);
 
     std::optional<ui64> GetLockForTx(const ui64 txId) const;
 
diff --git a/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h b/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h
index 4c2ed6a39af..b903f2e6221 100644
--- a/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h
+++ b/ydb/core/tx/columnshard/transactions/operators/ev_write/primary.h
@@ -23,6 +23,7 @@ private:
     std::set<ui64> WaitShardsBrokenFlags;
     std::set<ui64> WaitShardsResultAck;
     std::optional<bool> TxBroken;
+    bool SelfBrokenFlagPersisted = false;
 
     virtual void DoSerializeToProto(NKikimrTxColumnShard::TCommitWriteTxBody& result) const override {
         auto& data = *result.MutablePrimaryTabletData();
@@ -76,6 +77,51 @@ private:
         return "EV_WRITE_PRIMARY";
     }
 
+    class TTxStartPreparation: public TExtendedTransactionBase {
+    private:
+        using TBase = TExtendedTransactionBase;
+        const ui64 TxId;
+
+        virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& /*ctx*/) override {
+            auto& txController = Self->GetProgressTxController();
+            auto& lock = Self->GetOperationsManager().GetLockFeaturesForTxVerified(TxId);
+            auto op = txController.GetTxOperatorAs<TEvWriteCommitPrimaryTransactionOperator>(TxId, ETxOperatorStatus::InProgress);
+            if (op->WaitShardsBrokenFlags.contains(Self->TabletID())) {
+                // TxStartPreparation may be executed AFTER some ReadSets from the secondary.
+                // So, TxBroken may already be set, we must not ignore that.
+                op->TxBroken = op->TxBroken.value_or(false) || lock.IsBroken();
+                AFL_VERIFY(op->WaitShardsBrokenFlags.erase(Self->TabletID()));
+                AFL_VERIFY(op->WaitShardsResultAck.erase(Self->TabletID()));
+                AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "remove_tablet_id")(
+                    "wait_broken_flags", JoinSeq(",", op->WaitShardsBrokenFlags))("wait_result_ack", JoinSeq(",", op->WaitShardsResultAck))(
+                    "receive", Self->TabletID());
+                txController.WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
+            } else {
+                op->SelfBrokenFlagPersisted = true;
+                op->InitializeRequests(*Self);
+            }
+            return true;
+        }
+
+        virtual void DoComplete(const NActors::TActorContext& /*ctx*/) override {
+            auto op = Self->GetProgressTxController().GetTxOperatorAs<TEvWriteCommitPrimaryTransactionOperator>(
+                TxId, ETxOperatorStatus::InProgress, /*optional*/ true);
+            if (op) {
+                if (!op->SelfBrokenFlagPersisted) {
+                    op->SelfBrokenFlagPersisted = true;
+                    op->InitializeRequests(*Self);
+                }
+            }
+        }
+
+    public:
+        TTxStartPreparation(TColumnShard* owner, const ui64 txId)
+            : TBase(owner, "start_preparation")
+            , TxId(txId)
+        {
+        }
+    };
+
     class TTxWriteReceivedBrokenFlag: public TExtendedTransactionBase {
     private:
         using TBase = TExtendedTransactionBase;
@@ -86,7 +132,8 @@ private:
         std::unique_ptr<TEvTxProcessing::TEvReadSetAck> BrokenFlagAck;
 
         virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& /*ctx*/) override {
-            auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs<TEvWriteCommitPrimaryTransactionOperator>(TxId, true);
+            auto& txController = Self->GetProgressTxController();
+            auto op = txController.GetTxOperatorAs<TEvWriteCommitPrimaryTransactionOperator>(TxId, ETxOperatorStatus::Any, /*optional*/ true);
             if (!op) {
                 AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("event", "repeated shard broken_flag info, operator not found")(
                     "shard_id", TabletId);
@@ -96,15 +143,20 @@ private:
             }
 
             BrokenFlagAck = TEvWriteCommitSyncTransactionOperator::MakeBrokenFlagAck(op->GetStep(), op->GetTxId(), Self->TabletID(), TabletId);
-            if (!op->WaitShardsBrokenFlags.erase(TabletId)) {
-                AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("event", "repeated shard broken_flag info")("shard_id", TabletId);
-                // we cannot send the ack here, because the previous transaction (that successfully erased TabletId) may be not completed yet
-            } else {
+            if (txController.IsTxCompleting(TxId)) {
+                return true;
+            }
+
+            // op is still in progress
+            if (op->WaitShardsBrokenFlags.erase(TabletId)) {
                 op->TxBroken = op->TxBroken.value_or(false) || BrokenFlag;
-                Self->GetProgressTxController().WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
+                txController.WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
                 // we cannot send the ack right away, we must make sure that we have stored the TxBroken value
                 // but we can proceed right away if we have collected all the broken flags from the secondary
                 op->InitializeRequests(*Self);
+            } else {
+                AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("event", "repeated shard broken_flag info")("shard_id", TabletId);
+                // we cannot send the ack here, because the previous transaction (that successfully erased TabletId) may be not completed yet
             }
             return true;
         }
@@ -126,11 +178,6 @@ private:
         }
     };
 
-    virtual std::unique_ptr<NTabletFlatExecutor::ITransaction> CreateReceiveBrokenFlagTx(
-        TColumnShard& owner, const ui64 sendTabletId, const bool broken) const override {
-        return std::make_unique<TTxWriteReceivedBrokenFlag>(owner, GetTxId(), GetStep(), sendTabletId, broken);
-    }
-
     class TTxWriteReceivedResultAck: public TExtendedTransactionBase {
     private:
         using TBase = TExtendedTransactionBase;
@@ -138,18 +185,20 @@ private:
         const ui64 TabletId;
 
         virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& /*ctx*/) override {
-            auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs<TEvWriteCommitPrimaryTransactionOperator>(TxId, true);
+            auto& txController = Self->GetProgressTxController();
+            auto op =
+                txController.GetTxOperatorAs<TEvWriteCommitPrimaryTransactionOperator>(TxId, ETxOperatorStatus::InProgress, /*optional*/ true);
             if (!op) {
                 AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "ack_tablet_duplication")("receive", TabletId)(
                     "reason", "operation absent");
-            } else if (!op->WaitShardsResultAck.erase(TabletId)) {
-                AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("event", "ack_tablet_duplication")("wait", JoinSeq(",", op->WaitShardsResultAck))(
-                    "receive", TabletId);
-            } else {
+            } else if (op->WaitShardsResultAck.erase(TabletId)) {
                 AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "ack_tablet")("wait", JoinSeq(",", op->WaitShardsResultAck))(
                     "receive", TabletId);
-                Self->GetProgressTxController().WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
+                txController.WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
                 op->CheckFinished(*Self);
+            } else {
+                AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("event", "ack_tablet_duplication")("wait", JoinSeq(",", op->WaitShardsResultAck))(
+                    "receive", TabletId);
             }
             return true;
         }
@@ -166,13 +215,18 @@ private:
         }
     };
 
+    virtual std::unique_ptr<NTabletFlatExecutor::ITransaction> CreateReceiveBrokenFlagTx(
+        TColumnShard& owner, const ui64 sendTabletId, const bool broken) const override {
+        return std::make_unique<TTxWriteReceivedBrokenFlag>(owner, GetTxId(), GetStep(), sendTabletId, broken);
+    }
+
     virtual bool IsTxBroken() const override {
         AFL_VERIFY(TxBroken);
         return *TxBroken;
     }
 
     void InitializeRequests(TColumnShard& owner) {
-        if (WaitShardsBrokenFlags.empty()) {
+        if (SelfBrokenFlagPersisted && WaitShardsBrokenFlags.empty()) {
             if (WaitShardsResultAck.size()) {
                 SendResult(owner);
             } else {
@@ -209,40 +263,6 @@ private:
     virtual void DoOnTabletInit(TColumnShard& /*owner*/) override {
     }
 
-    class TTxStartPreparation: public TExtendedTransactionBase {
-    private:
-        using TBase = TExtendedTransactionBase;
-        const ui64 TxId;
-
-        virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& /*ctx*/) override {
-            auto& lock = Self->GetOperationsManager().GetLockVerified(Self->GetOperationsManager().GetLockForTxVerified(TxId));
-            auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs<TEvWriteCommitPrimaryTransactionOperator>(TxId);
-            if (op->WaitShardsBrokenFlags.contains(Self->TabletID())) {
-                // TxStartPreparation may be executed AFTER all the ReadSets from secondary.
-                // So, TxBroken may already be set, we must not ignore that.
-                op->TxBroken = op->TxBroken.value_or(false) || lock.IsBroken();
-                AFL_VERIFY(op->WaitShardsBrokenFlags.erase(Self->TabletID()));
-                AFL_VERIFY(op->WaitShardsResultAck.erase(Self->TabletID()));
-                AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "remove_tablet_id")(
-                    "wait_broken_flags", JoinSeq(",", op->WaitShardsBrokenFlags))("wait_result_ack", JoinSeq(",", op->WaitShardsResultAck))(
-                    "receive", Self->TabletID());
-                Self->GetProgressTxController().WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
-                op->InitializeRequests(*Self);
-            }
-            return true;
-        }
-
-        virtual void DoComplete(const NActors::TActorContext& /*ctx*/) override {
-        }
-
-    public:
-        TTxStartPreparation(TColumnShard* owner, const ui64 txId)
-            : TBase(owner, "start_preparation")
-            , TxId(txId)
-        {
-        }
-    };
-
     virtual void OnTimeout(TColumnShard& owner) override {
         InitializeRequests(owner);
     }
diff --git a/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h b/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h
index a0f4a777ee2..bd17a03bc00 100644
--- a/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h
+++ b/ydb/core/tx/columnshard/transactions/operators/ev_write/secondary.h
@@ -18,9 +18,11 @@ private:
     virtual void DoSerializeToProto(NKikimrTxColumnShard::TCommitWriteTxBody& result) const override {
         auto& data = *result.MutableSecondaryTabletData();
         if (TxBroken) {
-            data.SetTxBroken(*TxBroken);
+            data.SetTxBroken(TxBroken.value());
+        }
+        if (SelfBroken.has_value()) {
+            data.SetSelfBroken(SelfBroken.value());
         }
-        data.SetSelfBroken(SelfBroken);
         data.SetNeedReceiveBroken(NeedReceiveBroken);
         data.SetReceiveAck(ReceiveAck);
         data.SetArbiterTabletId(ArbiterTabletId);
@@ -30,8 +32,9 @@ private:
     ui64 ArbiterTabletId;
     bool NeedReceiveBroken = false;
     bool ReceiveAck = false;
-    bool SelfBroken = false;
+    std::optional<bool> SelfBroken;
     std::optional<bool> TxBroken;
+    bool SelfBrokenFlagPersisted = false;
 
     virtual bool DoParseImpl(TColumnShard& /*owner*/, const NKikimrTxColumnShard::TCommitWriteTxBody& commitTxBody) override {
         if (!commitTxBody.HasSecondaryTabletData()) {
@@ -39,7 +42,9 @@ private:
             return false;
         }
         auto& protoData = commitTxBody.GetSecondaryTabletData();
-        SelfBroken = protoData.GetSelfBroken();
+        if (protoData.HasSelfBroken()) {
+            SelfBroken = protoData.GetSelfBroken();
+        }
         ArbiterTabletId = protoData.GetArbiterTabletId();
         NeedReceiveBroken = protoData.GetNeedReceiveBroken();
         ReceiveAck = protoData.GetReceiveAck();
@@ -58,13 +63,57 @@ private:
         return "EV_WRITE_SECONDARY";
     }
 
+    class TTxStartPreparation: public TExtendedTransactionBase {
+    private:
+        using TBase = TExtendedTransactionBase;
+        const ui64 TxId;
+
+        virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& /*ctx*/) override {
+            auto& txController = Self->GetProgressTxController();
+            auto op = txController.GetTxOperatorAs<TEvWriteCommitSecondaryTransactionOperator>(TxId, ETxOperatorStatus::InProgress);
+            if (op->SelfBroken.has_value()) {
+                op->SelfBrokenFlagPersisted = true;
+                if (!op->ReceiveAck) {
+                    // We can send the result here, because we are sure that SelfBroken is persisted
+                    op->SendResult(*Self);
+                }
+            } else {
+                auto& lock = Self->GetOperationsManager().GetLockFeaturesForTxVerified(TxId);
+                op->SelfBroken = lock.IsBroken();
+                txController.WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
+            }
+            return true;
+        }
+
+        virtual void DoComplete(const NActors::TActorContext& /*ctx*/) override {
+            auto op = Self->GetProgressTxController().GetTxOperatorAs<TEvWriteCommitSecondaryTransactionOperator>(
+                TxId, ETxOperatorStatus::InProgress, /*optional*/ true);
+            if (op) {
+                if (!op->SelfBrokenFlagPersisted) {
+                    op->SelfBrokenFlagPersisted = true;
+                    if (!op->ReceiveAck) {
+                        op->SendResult(*Self);
+                    }
+                }
+            }
+        }
+
+    public:
+        TTxStartPreparation(TColumnShard* owner, const ui64 txId)
+            : TBase(owner, "start_preparation")
+            , TxId(txId)
+        {
+        }
+    };
+
     class TTxWriteReceivedAck: public TExtendedTransactionBase {
     private:
         using TBase = TExtendedTransactionBase;
         const ui64 TxId;
 
         virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& ctx) override {
-            auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs<TEvWriteCommitSecondaryTransactionOperator>(TxId, true);
+            auto op = Self->GetProgressTxController().GetTxOperatorAs<TEvWriteCommitSecondaryTransactionOperator>(
+                TxId, ETxOperatorStatus::InProgress, /*optional*/ true);
             if (!op || op->ReceiveAck) {
                 AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("event", "duplication_tablet_ack_flag")("txId", TxId);
             } else {
@@ -91,12 +140,6 @@ private:
         }
     };
 
-    virtual std::unique_ptr<NTabletFlatExecutor::ITransaction> CreateReceiveResultAckTx(
-        TColumnShard& owner, const ui64 recvTabletId) const override {
-        AFL_VERIFY(recvTabletId == ArbiterTabletId)("recv", recvTabletId)("arbiter", ArbiterTabletId);
-        return std::make_unique<TTxWriteReceivedAck>(owner, GetTxId());
-    }
-
     class TTxWriteReceivedBrokenFlag: public TExtendedTransactionBase {
     private:
         using TBase = TExtendedTransactionBase;
@@ -107,7 +150,8 @@ private:
         std::unique_ptr<TEvTxProcessing::TEvReadSetAck> BrokenFlagAck;
 
         virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& ctx) override {
-            auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs<TEvWriteCommitSecondaryTransactionOperator>(TxId, true);
+            auto& txController = Self->GetProgressTxController();
+            auto op = txController.GetTxOperatorAs<TEvWriteCommitSecondaryTransactionOperator>(TxId, ETxOperatorStatus::Any, /*optional*/ true);
             if (!op) {
                 AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("event", "duplication_tablet_broken_flag")("txId", TxId);
                 // send the ack anyway, so that the primary waits less time to progress
@@ -117,12 +161,17 @@ private:
 
             BrokenFlagAck =
                 TEvWriteCommitSyncTransactionOperator::MakeBrokenFlagAck(op->GetStep(), op->GetTxId(), Self->TabletID(), ArbiterTabletId);
+            if (txController.IsTxCompleting(TxId)) {
+                return true;
+            }
+
+            // op is still in progress
             if (op->TxBroken.has_value()) {
                 AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("event", "duplication_tablet_broken_flag")("txId", TxId);
                 // we cannot send the ack here, because the previous transaction (that successfully set TxBroken) may be not completed yet
             } else {
                 op->TxBroken = BrokenFlag;
-                Self->GetProgressTxController().WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
+                txController.WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
 
                 // me must check IsInProgress, because the ReadSet and the Ack from the primary may come in any order,
                 // and we do not want to enqueue the progress tx twice
@@ -156,6 +205,12 @@ private:
         }
     };
 
+    virtual std::unique_ptr<NTabletFlatExecutor::ITransaction> CreateReceiveResultAckTx(
+        TColumnShard& owner, const ui64 recvTabletId) const override {
+        AFL_VERIFY(recvTabletId == ArbiterTabletId)("recv", recvTabletId)("arbiter", ArbiterTabletId);
+        return std::make_unique<TTxWriteReceivedAck>(owner, GetTxId());
+    }
+
     virtual std::unique_ptr<NTabletFlatExecutor::ITransaction> CreateReceiveBrokenFlagTx(
         TColumnShard& owner, const ui64 sendTabletId, const bool broken) const override {
         AFL_VERIFY(ArbiterTabletId == sendTabletId);
@@ -167,8 +222,10 @@ private:
             AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "skip_continue");
             return;
         }
+        AFL_VERIFY(SelfBroken.has_value());
         NKikimrTx::TReadSetData readSetData;
-        readSetData.SetDecision(SelfBroken ? NKikimrTx::TReadSetData::DECISION_ABORT : NKikimrTx::TReadSetData::DECISION_COMMIT);
+        bool broken = SelfBroken.value();
+        readSetData.SetDecision(broken ? NKikimrTx::TReadSetData::DECISION_ABORT : NKikimrTx::TReadSetData::DECISION_COMMIT);
         auto event = std::make_unique<TEvTxProcessing::TEvReadSet>(
             0, GetTxId(), owner.TabletID(), ArbiterTabletId, owner.TabletID(), readSetData.SerializeAsString());
         TEvWriteCommitSyncTransactionOperator::SendPersistent(owner, std::move(event), ArbiterTabletId, GetTxId());
@@ -177,36 +234,6 @@ private:
     virtual void DoOnTabletInit(TColumnShard& /*owner*/) override {
     }
 
-    class TTxStartPreparation: public TExtendedTransactionBase {
-    private:
-        using TBase = TExtendedTransactionBase;
-        const ui64 TxId;
-
-        virtual bool DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const NActors::TActorContext& /*ctx*/) override {
-            auto& lock = Self->GetOperationsManager().GetLockVerified(Self->GetOperationsManager().GetLockForTxVerified(TxId));
-            auto op = Self->GetProgressTxController().GetTxOperatorVerifiedAs<TEvWriteCommitSecondaryTransactionOperator>(TxId);
-            op->SelfBroken = lock.IsBroken();
-            Self->GetProgressTxController().WriteTxOperatorInfo(txc, TxId, op->SerializeToProto().SerializeAsString());
-            if (!op->ReceiveAck) {
-                // We send the result here before SelfBroken is truly persisted, yes.
-                // But we persist lock.IsBroken(), so if the secondary crushes and restarts,
-                // the secondary will send the same result again, and the primary will ignore it (if already processed)
-                op->SendResult(*Self);
-            }
-            return true;
-        }
-
-        virtual void DoComplete(const NActors::TActorContext& /*ctx*/) override {
-        }
-
-    public:
-        TTxStartPreparation(TColumnShard* owner, const ui64 txId)
-            : TBase(owner, "start_preparation")
-            , TxId(txId)
-        {
-        }
-    };
-
     virtual bool DoIsInProgress() const override {
         return !ReceiveAck || !TxBroken.has_value();
     }
@@ -217,7 +244,7 @@ private:
     }
 
     virtual void OnTimeout(TColumnShard& owner) override {
-        if (!ReceiveAck) {
+        if (SelfBrokenFlagPersisted && !ReceiveAck) {
             SendResult(owner);
         }
     }
diff --git a/ydb/core/tx/columnshard/transactions/tx_controller.cpp b/ydb/core/tx/columnshard/transactions/tx_controller.cpp
index b2024a1135e..c857a4c30d5 100644
--- a/ydb/core/tx/columnshard/transactions/tx_controller.cpp
+++ b/ydb/core/tx/columnshard/transactions/tx_controller.cpp
@@ -28,8 +28,8 @@ ui64 TTxController::GetAllowedStep() const {
 }
 
 ui64 TTxController::GetMemoryUsage() const {
-    return Operators.size() * (sizeof(TTxController::ITransactionOperator) + 24) + DeadlineQueue.size() * sizeof(TPlanQueueItem) +
-           (PlanQueue.size() + RunningQueue.size()) * sizeof(TPlanQueueItem);
+    return (Operators.size() + CompletingOperators.size()) * (sizeof(TTxController::ITransactionOperator) + 24) +
+           DeadlineQueue.size() * sizeof(TPlanQueueItem) + (PlanQueue.size() + RunningQueue.size()) * sizeof(TPlanQueueItem);
 }
 
 TTxController::TPlanQueueItem TTxController::GetFrontTx() const {
@@ -99,9 +99,10 @@ bool TTxController::Load(NTabletFlatExecutor::TTransactionContext& txc) {
 
 std::shared_ptr<TTxController::ITransactionOperator> TTxController::UpdateTxSourceInfo(
     const TFullTxInfo& tx, NTabletFlatExecutor::TTransactionContext& txc) {
-    auto op = GetTxOperatorVerified(tx.GetTxId());
+    auto op = GetTxOperator(tx.GetTxId(), ETxOperatorStatus::InProgress);
     const bool sourceChanged = op->GetTxInfo().Source != tx.Source;
     op->ResetStatusOnUpdate(sourceChanged);
+
     auto& txInfo = op->MutableTxInfo();
     txInfo.Source = tx.Source;
     txInfo.MinStep = tx.MinStep;
@@ -153,9 +154,20 @@ bool TTxController::AbortTx(const TPlanQueueItem planQueueItem) {
     return true;
 }
 
+bool TTxController::ExecuteOnCancel(const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc) {
+    ITransactionOperator::TPtr op = MoveOperatorToCompleting(txId);
+    AFL_VERIFY(op->GetTxInfo().PlanStep == 0)("tx_id", txId)("plan_step", op->GetTxInfo().PlanStep);
+
+    op->ExecuteOnAbort(Owner, txc);
+
+    NIceDb::TNiceDb db(txc.DB);
+    Schema::EraseTxInfo(db, txId);
+    return true;
+}
+
 bool TTxController::CompleteOnCancel(const ui64 txId, const TActorContext& ctx) {
-    auto opIt = Operators.find(txId);
-    AFL_VERIFY(opIt != Operators.end())("tx_id", txId);
+    auto opIt = CompletingOperators.find(txId);
+    AFL_VERIFY(opIt != CompletingOperators.end())("tx_id", txId);
     AFL_VERIFY(opIt->second->GetTxInfo().PlanStep == 0)("tx_id", txId)("plan_step", opIt->second->GetTxInfo().PlanStep);
 
     opIt->second->CompleteOnAbort(Owner, ctx);
@@ -168,21 +180,9 @@ bool TTxController::CompleteOnCancel(const ui64 txId, const TActorContext& ctx)
     return true;
 }
 
-bool TTxController::ExecuteOnCancel(const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc) {
-    auto opIt = Operators.find(txId);
-    AFL_VERIFY(opIt != Operators.end());
-    AFL_VERIFY(opIt->second->GetTxInfo().PlanStep == 0)("tx_id", txId)("plan_step", opIt->second->GetTxInfo().PlanStep);
-
-    opIt->second->ExecuteOnAbort(Owner, txc);
-
-    NIceDb::TNiceDb db(txc.DB);
-    Schema::EraseTxInfo(db, txId);
-    return true;
-}
-
 std::optional<TTxController::TTxInfo> TTxController::GetFirstPlannedTx() const {
     if (!PlanQueue.empty()) {
-        return GetTxInfoVerified(PlanQueue.begin()->TxId);
+        return GetTxInfoVerified(PlanQueue.begin()->TxId, ETxOperatorStatus::InProgress);
     }
     return std::nullopt;
 }
@@ -194,16 +194,15 @@ std::optional<TTxController::TTxInfo> TTxController::PopFirstPlannedTx() {
         auto txId = item.TxId;
         TPlanQueueItem tx(item.Step, item.TxId);
         RunningQueue.emplace(std::move(item));
-        return GetTxInfoVerified(txId);
+        return GetTxInfoVerified(txId, ETxOperatorStatus::InProgress);
     }
     return std::nullopt;
 }
 
 void TTxController::ProgressOnExecute(const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc) {
+    ITransactionOperator::TPtr op = MoveOperatorToCompleting(txId);
     NIceDb::TNiceDb db(txc.DB);
-    auto opIt = Operators.find(txId);
-    AFL_VERIFY(opIt != Operators.end())("tx_id", txId);
-    Counters.OnFinishPlannedTx(opIt->second->GetOpType());
+    Counters.OnFinishPlannedTx(op->GetOpType());
     AFL_NOTICE(NKikimrServices::TX_COLUMNSHARD_TX)("event", "finished_tx")("tx_id", txId);
     Schema::EraseTxInfo(db, txId);
 }
@@ -213,11 +212,28 @@ void TTxController::ProgressOnComplete(const TPlanQueueItem& txItem) {
     AFL_VERIFY(RunningQueue.erase(txItem))("info", txItem.DebugString());
 }
 
+TTxController::ITransactionOperator::TPtr TTxController::MoveOperatorToCompleting(const ui64 txId) {
+    auto opIt = Operators.find(txId);
+    AFL_VERIFY(opIt != Operators.end())("tx_id", txId);
+
+    ITransactionOperator::TPtr op = std::move(opIt->second);
+    Operators.erase(opIt);
+
+    AFL_VERIFY(CompletingOperators.emplace(txId, op).second)("tx_id", txId);
+    return op;
+}
+
 void TTxController::OnTxCompleted(const ui64 txId) {
-    AFL_VERIFY(Operators.erase(txId));
+    AFL_VERIFY(CompletingOperators.erase(txId))("tx_id", txId);
     Owner.Subscribers->OnEvent(std::make_shared<NColumnShard::NSubscriber::TEventTxCompleted>(txId));
 }
 
+void TTxController::WriteTxOperatorInfo(NTabletFlatExecutor::TTransactionContext& txc, const ui64 txId, const TString& data) {
+    AFL_VERIFY(GetTxOperator(txId, ETxOperatorStatus::InProgress))("tx_id", txId);
+    NIceDb::TNiceDb db(txc.DB);
+    NColumnShard::Schema::UpdateTxInfoBody(db, txId, data);
+}
+
 THashSet<ui64> TTxController::GetTxs() const {
     THashSet<ui64> result;
     for (const auto& [txId, _] : Operators) {
@@ -233,18 +249,53 @@ std::optional<TTxController::TPlanQueueItem> TTxController::GetPlannedTx() const
     return *PlanQueue.begin();
 }
 
-std::optional<TTxController::TTxInfo> TTxController::GetTxInfo(const ui64 txId) const {
-    auto it = Operators.find(txId);
-    if (it != Operators.end()) {
-        return it->second->GetTxInfo();
+TTxController::ITransactionOperator::TPtr TTxController::GetTxOperator(const ui64 txId, ETxOperatorStatus status, const bool optional) const {
+    ITransactionOperator::TPtr result;
+    switch (status) {
+        case ETxOperatorStatus::InProgress: {
+            auto it = Operators.find(txId);
+            if (it != Operators.end()) {
+                result = it->second;
+            }
+            break;
+        }
+        case ETxOperatorStatus::Completing: {
+            auto it = CompletingOperators.find(txId);
+            if (it != CompletingOperators.end()) {
+                result = it->second;
+            }
+            break;
+        }
+        case ETxOperatorStatus::Any: {
+            auto it = Operators.find(txId);
+            if (it != Operators.end()) {
+                result = it->second;
+            }
+            if (result == nullptr) {
+                auto it = CompletingOperators.find(txId);
+                if (it != CompletingOperators.end()) {
+                    result = it->second;
+                }
+            }
+            break;
+        }
+        default:
+            AFL_VERIFY(false)("status", status);
     }
-    return std::nullopt;
+    AFL_VERIFY(optional || result)("tx_id", txId)("status", status);
+    return result;
 }
 
-TTxController::TTxInfo TTxController::GetTxInfoVerified(const ui64 txId) const {
-    auto it = Operators.find(txId);
-    AFL_VERIFY(it != Operators.end());
-    return it->second->GetTxInfo();
+std::optional<TTxController::TTxInfo> TTxController::GetTxInfo(const ui64 txId, ETxOperatorStatus status) const {
+    auto op = GetTxOperator(txId, status, /*optional*/ true);
+    if (!op) {
+        return std::nullopt;
+    }
+    return op->GetTxInfo();
+}
+
+TTxController::TTxInfo TTxController::GetTxInfoVerified(const ui64 txId, ETxOperatorStatus status) const {
+    return GetTxOperator(txId, status)->GetTxInfo();
 }
 
 NEvents::TDataEvents::TCoordinatorInfo TTxController::BuildCoordinatorInfo(const TTxInfo& txInfo) const {
@@ -334,9 +385,9 @@ std::shared_ptr<TTxController::ITransactionOperator> TTxController::StartPropose
     }
     Counters.OnStartProposeOnExecute(txOperator->GetOpType());
 
-    auto txInfoPtr = GetTxInfo(txInfo.TxId);
-    if (!!txInfoPtr) {
-        if (!txOperator->CheckAllowUpdate(*txInfoPtr)) {
+    auto txInfoPtr = GetTxInfo(txInfo.TxId, ETxOperatorStatus::Any);
+    if (txInfoPtr.has_value()) {
+        if (IsTxCompleting(txInfo.TxId) || !txOperator->CheckAllowUpdate(*txInfoPtr)) {
             AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("error", "incorrect duplication")("actual_tx", txInfoPtr->DebugString());
             TTxController::TProposeResult proposeResult(NKikimrTxColumnShard::EResultStatus::ERROR,
                 TStringBuilder() << "Another commit TxId# " << txInfo.TxId << " has already been proposed");
@@ -372,7 +423,7 @@ void TTxController::StartProposeOnComplete(ITransactionOperator& txOperator, con
 
 void TTxController::FinishProposeOnExecute(const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc) {
     NActors::TLogContextGuard lGuard = NActors::TLogContextBuilder::Build()("method", "TTxController::FinishProposeOnExecute")("tx_id", txId);
-    if (auto txOperator = GetTxOperatorOptional(txId)) {
+    if (auto txOperator = GetTxOperator(txId, ETxOperatorStatus::InProgress, /*optional*/ true)) {
         AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "start");
         txOperator->FinishProposeOnExecute(Owner, txc);
         Counters.OnFinishProposeOnExecute(txOperator->GetOpType());
@@ -395,7 +446,7 @@ void TTxController::FinishProposeOnComplete(ITransactionOperator& txOperator, co
 }
 
 void TTxController::FinishProposeOnComplete(const ui64 txId, const TActorContext& ctx) {
-    auto txOperator = GetTxOperatorOptional(txId);
+    auto txOperator = GetTxOperator(txId, ETxOperatorStatus::InProgress, /*optional*/ true);
     if (!txOperator) {
         AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("error", "cannot found txOperator in propose transaction finish")("tx_id", txId);
         return;
diff --git a/ydb/core/tx/columnshard/transactions/tx_controller.h b/ydb/core/tx/columnshard/transactions/tx_controller.h
index 181bc0dd1c4..76c38c9a5d2 100644
--- a/ydb/core/tx/columnshard/transactions/tx_controller.h
+++ b/ydb/core/tx/columnshard/transactions/tx_controller.h
@@ -164,6 +164,12 @@ public:
     }
 };
 
+enum class ETxOperatorStatus {
+    InProgress,
+    Completing,
+    Any
+};
+
 class TTxController {
 public:
     struct TPlanQueueItem {
@@ -488,9 +494,11 @@ private:
     TTxProgressCounters Counters;
 
     THashMap<ui64, ITransactionOperator::TPtr> Operators;
+    THashMap<ui64, ITransactionOperator::TPtr> CompletingOperators;
 
 private:
     bool AbortTx(const TPlanQueueItem planQueueItem);
+    ITransactionOperator::TPtr MoveOperatorToCompleting(const ui64 txId);
 
     TTxInfo RegisterTx(const std::shared_ptr<TTxController::ITransactionOperator>& txOperator, const TString& txBody,
         NTabletFlatExecutor::TTransactionContext& txc);
@@ -504,25 +512,18 @@ public:
 
     ui64 GetAllowedStep() const;
 
-    ITransactionOperator::TPtr GetTxOperatorOptional(const ui64 txId) const {
-        auto it = Operators.find(txId);
-        if (it == Operators.end()) {
-            return nullptr;
-        }
-        return it->second;
+    bool IsTxCompleting(const ui64 txId) const {
+        return CompletingOperators.contains(txId);
     }
 
-    ITransactionOperator::TPtr GetTxOperatorVerified(const ui64 txId) const {
-        return TValidator::CheckNotNull(GetTxOperatorOptional(txId));
-    }
+    ITransactionOperator::TPtr GetTxOperator(const ui64 txId, ETxOperatorStatus status, const bool optional = false) const;
 
     template <class TExpectedTransactionOperator>
-    std::shared_ptr<TExpectedTransactionOperator> GetTxOperatorVerifiedAs(const ui64 txId, const bool optionalExists = false) const {
-        auto result = GetTxOperatorOptional(txId);
-        if (optionalExists && !result) {
+    std::shared_ptr<TExpectedTransactionOperator> GetTxOperatorAs(const ui64 txId, ETxOperatorStatus status, const bool optional = false) const {
+        auto result = GetTxOperator(txId, status, optional);
+        if (!result) {
             return nullptr;
         }
-        AFL_VERIFY(result)("tx_id", txId);
         auto resultClass = dynamic_pointer_cast<TExpectedTransactionOperator>(result);
         AFL_VERIFY(resultClass)("tx_id", txId);
         return resultClass;
@@ -533,7 +534,7 @@ public:
         if (!txInfo) {
             return;
         }
-        GetTxOperatorVerified(txInfo->GetTxId())->PingTimeout(Owner, now);
+        GetTxOperator(txInfo->GetTxId(), ETxOperatorStatus::InProgress)->PingTimeout(Owner, now);
     }
 
     ui64 GetMemoryUsage() const;
@@ -551,10 +552,7 @@ public:
     void FinishProposeOnComplete(ITransactionOperator& txOperator, const TActorContext& ctx);
     void FinishProposeOnComplete(const ui64 txId, const TActorContext& ctx);
 
-    void WriteTxOperatorInfo(NTabletFlatExecutor::TTransactionContext& txc, const ui64 txId, const TString& data) {
-        NIceDb::TNiceDb db(txc.DB);
-        NColumnShard::Schema::UpdateTxInfoBody(db, txId, data);
-    }
+    void WriteTxOperatorInfo(NTabletFlatExecutor::TTransactionContext& txc, const ui64 txId, const TString& data);
 
     bool ExecuteOnCancel(const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc);
     bool CompleteOnCancel(const ui64 txId, const TActorContext& ctx);
@@ -567,8 +565,8 @@ public:
 
     std::optional<TPlanQueueItem> GetPlannedTx() const;
     TPlanQueueItem GetFrontTx() const;
-    std::optional<TTxInfo> GetTxInfo(const ui64 txId) const;
-    TTxInfo GetTxInfoVerified(const ui64 txId) const;
+    std::optional<TTxInfo> GetTxInfo(const ui64 txId, ETxOperatorStatus status) const;
+    TTxInfo GetTxInfoVerified(const ui64 txId, ETxOperatorStatus status) const;
     NEvents::TDataEvents::TCoordinatorInfo BuildCoordinatorInfo(const TTxInfo& txInfo) const;
 
     size_t CleanExpiredTxs();