cancel on prepare has been fixed (#45578)

21 files changed, 550 insertions, 79 deletions

diff --git a/ydb/core/tx/columnshard/backup/import/control.cpp b/ydb/core/tx/columnshard/backup/import/control.cpp
index a95c9805a0a..c7a64eb8f95 100644
--- a/ydb/core/tx/columnshard/backup/import/control.cpp
+++ b/ydb/core/tx/columnshard/backup/import/control.cpp
@@ -13,6 +13,11 @@ NKikimr::TConclusionStatus TConfirmSessionControl::DoApply(const std::shared_ptr
 NKikimr::TConclusionStatus TAbortSessionControl::DoApply(const std::shared_ptr<NBackground::ISessionLogic>& session) const {
     auto exportSession = dynamic_pointer_cast<TSession>(session);
     AFL_VERIFY(exportSession);
+    if (exportSession->IsFinished() || exportSession->IsReadyForRemoveOnFinished()) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "abort_control_skipped_terminal_session")("is_finished", exportSession->IsFinished())(
+            "is_aborted", exportSession->IsReadyForRemoveOnFinished());
+        return TConclusionStatus::Success();
+    }
     exportSession->Abort("Aborted by user");
     return TConclusionStatus::Success();
 }
diff --git a/ydb/core/tx/columnshard/backup/import/import_actor.cpp b/ydb/core/tx/columnshard/backup/import/import_actor.cpp
index 27dcd189a96..9276bbcd35b 100644
--- a/ydb/core/tx/columnshard/backup/import/import_actor.cpp
+++ b/ydb/core/tx/columnshard/backup/import/import_actor.cpp
@@ -58,7 +58,7 @@ void TImportActor::ScheduleTimeoutCheck() {
 }
 
 void TImportActor::HandleWakeup() {
-    if (Stage == EStage::Finished) {
+    if (Stage == EStage::Finished || Stage == EStage::WaitSaveCursor) {
         return;
     }
     const auto elapsed = TInstant::Now() - StageStartTime;
@@ -84,7 +84,15 @@ void TImportActor::AbortImport(const TString& errorMessage) {
         AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "abort_after_import_finished")("message", errorMessage);
         return;
     }
+    if (Stage == EStage::WaitSaveCursor) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "abort_during_save_cursor")("message", errorMessage);
+        return;
+    }
     ImportSession->Abort(errorMessage);
+    if (ImportActorId) {
+        Send(ImportActorId, new NActors::TEvents::TEvPoisonPill());
+        ImportActorId = {};
+    }
     if (Stage == EStage::WaitData) {
         Counters.OnReadFinished(TInstant::Now() - ReadStartTime);
     } else if (Stage == EStage::WaitWriting) {
@@ -118,13 +126,20 @@ void TImportActor::OnSessionStateSaved() {
 }
 
 void TImportActor::Handle(NColumnShard::TEvPrivate::TEvBackupImportRecordBatch::TPtr& ev) {
+    if (ImportSession->IsFinished() || ImportSession->IsReadyForRemoveOnFinished()) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "import_batch_after_terminal")("stage", StageToString(Stage))(
+            "is_finished", ImportSession->IsFinished())("is_aborted", ImportSession->IsReadyForRemoveOnFinished());
+        return;
+    }
     Counters.OnReadFinished(TInstant::Now() - ReadStartTime);
 
     if (ev->Get()->Error) {
         Counters.OnError();
         ImportSession->Abort(ev->Get()->Error);
     } else if (ev->Get()->IsLast) {
-        ImportSession->Finish();
+        if (ImportSession->IsStarted()) {
+            ImportSession->Finish();
+        }
     }
 
     if (!ev->Get()->Data || ev->Get()->IsLast || ev->Get()->Error) {
@@ -159,6 +174,11 @@ void TImportActor::Handle(NColumnShard::TEvPrivate::TEvBackupImportRecordBatch::
 }
 
 void TImportActor::Handle(NEvents::TDataEvents::TEvWriteResult::TPtr& ev) {
+    if (ImportSession->IsFinished() || ImportSession->IsReadyForRemoveOnFinished()) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "import_write_result_after_terminal")("stage", StageToString(Stage))(
+            "is_finished", ImportSession->IsFinished())("is_aborted", ImportSession->IsReadyForRemoveOnFinished());
+        return;
+    }
     Counters.OnWriteFinished(TInstant::Now() - WriteStartTime);
     AFL_VERIFY(ev->Get()->IsComplete())("error", ev->Get()->GetError());
     SwitchStage(EStage::WaitWriting, EStage::WaitData);
diff --git a/ydb/core/tx/columnshard/backup/import/session.cpp b/ydb/core/tx/columnshard/backup/import/session.cpp
index 81bf6e7950c..6e12f34f863 100644
--- a/ydb/core/tx/columnshard/backup/import/session.cpp
+++ b/ydb/core/tx/columnshard/backup/import/session.cpp
@@ -30,7 +30,11 @@ bool TSession::IsStarted() const {
 }
 
 void TSession::Abort(const TString &errorMessage) {
-    AFL_VERIFY(Status != EStatus::Finished && Status != EStatus::Aborted);
+    if (Status == EStatus::Finished || Status == EStatus::Aborted) {
+        AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "import_session_abort_ignored")("status", (ui64)Status)(
+            "message", "import session is already in terminal state");
+        return;
+    }
     Status = EStatus::Aborted;
     ErrorMessage = errorMessage;
 }
diff --git a/ydb/core/tx/columnshard/backup/import/ut/ut_import.cpp b/ydb/core/tx/columnshard/backup/import/ut/ut_import.cpp
index b2df970d7b4..5ac0e9a9d34 100644
--- a/ydb/core/tx/columnshard/backup/import/ut/ut_import.cpp
+++ b/ydb/core/tx/columnshard/backup/import/ut/ut_import.cpp
@@ -1,3 +1,5 @@
+#include <ydb/core/tx/columnshard/backup/import/session.h>
+#include <ydb/core/tx/columnshard/backup/import/task.h>
 #include <ydb/core/tx/columnshard/columnshard.h>
 #include <ydb/core/tx/columnshard/hooks/abstract/abstract.h>
 #include <ydb/core/tx/columnshard/hooks/testing/controller.h>
@@ -18,6 +20,78 @@ namespace NKikimr {
 using namespace NColumnShard;
 using namespace NTxUT;
 
+Y_UNIT_TEST_SUITE(ImportSessionStateMachine) {
+    static std::shared_ptr<NOlap::NImport::TSession> MakeSession() {
+        NKikimrSchemeOp::TRestoreTask restoreTask;
+        restoreTask.MutableS3Settings()->SetBucket("test");
+        restoreTask.MutableS3Settings()->SetEndpoint("localhost:9000");
+        restoreTask.SetTableId(1);
+        using TNameTypeInfo = std::pair<TString, NScheme::TTypeInfo>;
+        THashMap<ui32, TNameTypeInfo> emptyColumns;
+        auto task = std::make_shared<NOlap::NImport::TImportTask>(NColumnShard::TSchemeShardLocalPathId::FromRawValue(1),
+            /*columns=*/emptyColumns, restoreTask, /*schemaVersion=*/std::optional<ui64>{ 1 }, /*txId=*/std::optional<ui64>{ 42 });
+        return std::make_shared<NOlap::NImport::TSession>(task);
+    }
+
+    Y_UNIT_TEST(SessionIsNotStartedAfterAbort) {
+        auto session = MakeSession();
+        session->Confirm();
+        UNIT_ASSERT(session->IsConfirmed());
+        UNIT_ASSERT(!session->IsStarted());
+        session->Abort("simulated error: downloader timed out");
+        UNIT_ASSERT(!session->IsStarted());
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+    }
+
+    Y_UNIT_TEST(AbortOnConfirmedIsLegal) {
+        auto session = MakeSession();
+        session->Confirm();
+        UNIT_ASSERT(session->IsConfirmed());
+        session->Abort("abort from confirmed state");
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+        UNIT_ASSERT(!session->IsStarted());
+        UNIT_ASSERT(!session->IsFinished());
+    }
+
+    Y_UNIT_TEST(IsStartedReturnsFalseForNonStartedStates) {
+        auto session = MakeSession();
+        UNIT_ASSERT(!session->IsStarted());
+        session->Confirm();
+        UNIT_ASSERT(!session->IsStarted());
+        session->Abort("abort");
+        UNIT_ASSERT(!session->IsStarted());
+        UNIT_ASSERT(!session->IsFinished());
+    }
+
+    Y_UNIT_TEST(DoubleAbortDoesNotCrash) {
+        auto session = MakeSession();
+        session->Confirm();
+        session->Abort("first abort: timeout");
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+        session->Abort("second abort: racing error batch");
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+    }
+
+    Y_UNIT_TEST(AbortOnFinishedSessionDoesNotCrash) {
+        auto session = MakeSession();
+        session->Confirm();
+        session->Abort("abort on confirmed");
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+        session->Abort("abort again");
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+    }
+
+    Y_UNIT_TEST(AbortSessionControlOnAlreadyAbortedIsNoOp) {
+        auto session = MakeSession();
+        session->Confirm();
+        session->Abort("first abort: operation timed out");
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+        const bool shouldCallAbort = !session->IsFinished() && !session->IsReadyForRemoveOnFinished();
+        UNIT_ASSERT_C(!shouldCallAbort, "TAbortSessionControl::DoApply would call Abort() on an already-Aborted "
+                                        "session, reproducing the crash at session.cpp:33");
+    }
+}
+
 Y_UNIT_TEST_SUITE(Restore) {
     [[nodiscard]] TPlanStep ProposeTx(
         TTestBasicRuntime & runtime, TActorId & sender, NKikimrTxColumnShard::ETransactionKind txKind, const TString& txBody, const ui64 txId) {
diff --git a/ydb/core/tx/columnshard/bg_tasks/manager/actor.cpp b/ydb/core/tx/columnshard/bg_tasks/manager/actor.cpp
index 40e801f37ef..162bb93b1bb 100644
--- a/ydb/core/tx/columnshard/bg_tasks/manager/actor.cpp
+++ b/ydb/core/tx/columnshard/bg_tasks/manager/actor.cpp
@@ -27,7 +27,11 @@ void TSessionActor::SaveSessionProgress() {
 }
 
 void TSessionActor::SaveSessionState() {
-    AFL_VERIFY(!SaveSessionStateTx);
+    if (SaveSessionStateTx) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "save_session_state_skipped")("self_id", SelfId())("tablet_id", TabletId)(
+            "in_flight_tx", *SaveSessionStateTx);
+        return;
+    }
     const ui64 txId = GetNextTxId();
     SaveSessionStateTx.emplace(txId);
     auto tx = std::make_unique<TTxSaveSessionState>(Session, SelfId(), Adapter, txId);
@@ -38,6 +42,9 @@ void TSessionActor::SaveSessionState() {
 }
 
 void TSessionActor::Handle(TEvLocalTransactionCompleted::TPtr& ev) {
+    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "session_actor_local_tx_completed")("self_id", SelfId())("tablet_id", TabletId)(
+        "internal_tx_id", ev->Get()->GetInternalTxId())("save_progress_tx", SaveSessionProgressTx ? *SaveSessionProgressTx : 0)(
+        "save_state_tx", SaveSessionStateTx ? *SaveSessionStateTx : 0);
     if (SaveSessionProgressTx && *SaveSessionProgressTx == ev->Get()->GetInternalTxId()) {
         SaveSessionProgressTx.reset();
         OnSessionProgressSaved();
@@ -45,11 +52,15 @@ void TSessionActor::Handle(TEvLocalTransactionCompleted::TPtr& ev) {
         SaveSessionStateTx.reset();
         OnSessionStateSaved();
     } else {
+        AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "session_actor_on_tx_completed")("self_id", SelfId())("tablet_id", TabletId)(
+            "internal_tx_id", ev->Get()->GetInternalTxId());
         OnTxCompleted(ev->Get()->GetInternalTxId());
     }
 }
 
 void TSessionActor::Handle(TEvSessionControl::TPtr& ev) {
+    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "session_actor_handle_control")("self_id", SelfId())("tablet_id", TabletId)(
+        "save_state_tx", SaveSessionStateTx ? *SaveSessionStateTx : 0)("save_progress_tx", SaveSessionProgressTx ? *SaveSessionProgressTx : 0);
     TSessionControlContainer control;
     {
         auto conclusion = control.DeserializeFromProto(ev->Get()->Record);
@@ -65,6 +76,7 @@ void TSessionActor::Handle(TEvSessionControl::TPtr& ev) {
             return;
         }
     }
+    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "session_actor_control_saving_state")("self_id", SelfId())("tablet_id", TabletId);
     SaveSessionState();
 }
 
diff --git a/ydb/core/tx/columnshard/bg_tasks/manager/manager.cpp b/ydb/core/tx/columnshard/bg_tasks/manager/manager.cpp
index 52e0b647696..d8b76a2d540 100644
--- a/ydb/core/tx/columnshard/bg_tasks/manager/manager.cpp
+++ b/ydb/core/tx/columnshard/bg_tasks/manager/manager.cpp
@@ -109,4 +109,13 @@ ISessionLogic::TStatus TSessionsManager::GetStatus(const TString& className, con
     return session->GetStatus();
 }
 
+bool TSessionsManager::IsSessionComplete(const TString& className, const TString& identifier) const {
+    auto session = Storage->GetSession(className, identifier);
+    if (!session) {
+        return true;
+    }
+    const auto& logic = session->GetLogicContainer();
+    return logic->IsFinished() || logic->IsReadyForRemoveOnFinished();
+}
+
 }   // namespace NKikimr::NOlap::NBackground
diff --git a/ydb/core/tx/columnshard/bg_tasks/manager/manager.h b/ydb/core/tx/columnshard/bg_tasks/manager/manager.h
index 7c430d0954b..bbd149c385a 100644
--- a/ydb/core/tx/columnshard/bg_tasks/manager/manager.h
+++ b/ydb/core/tx/columnshard/bg_tasks/manager/manager.h
@@ -42,6 +42,7 @@ public:
         const NKikimrTxBackgroundProto::TSessionControlContainer& controlProto);
     [[nodiscard]] std::unique_ptr<NTabletFlatExecutor::ITransaction> TxApplyControl(const TSessionControlContainer& control);
     [[nodiscard]] ISessionLogic::TStatus GetStatus(const TString& className, const TString& identifier) const;
+    [[nodiscard]] bool IsSessionComplete(const TString& className, const TString& identifier) const;
 
     bool HasTask(const TTask& task) const;
 
diff --git a/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp b/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
index ff831fbd53b..1be790a65f7 100644
--- a/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
+++ b/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
@@ -104,7 +104,8 @@ public:
         }
 
         AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "actual tx operator");
-        if (internalOp->IsAsync() && !internalOp->NeedResendReply()) {
+        const bool waitForAsyncCompletion = internalOp->IsAsync() && (!internalOp->NeedResendReply() || !internalOp->IsProposeReplyReady(*Self));
+        if (waitForAsyncCompletion) {
             Self->GetProgressTxController().StartProposeOnComplete(*internalOp, ctx);
         } else {
             Self->GetProgressTxController().FinishProposeOnComplete(*internalOp, ctx);
diff --git a/ydb/core/tx/columnshard/columnshard__tx_abort.cpp b/ydb/core/tx/columnshard/columnshard__tx_abort.cpp
index d9fc0039ced..19cbb6242fe 100644
--- a/ydb/core/tx/columnshard/columnshard__tx_abort.cpp
+++ b/ydb/core/tx/columnshard/columnshard__tx_abort.cpp
@@ -4,9 +4,10 @@ namespace NKikimr::NColumnShard {
 
 class TTxTxAbort: public TTransactionBase<TColumnShard> {
 public:
-    TTxTxAbort(TColumnShard* self, ui64 txId)
+    TTxTxAbort(TColumnShard* self, ui64 txId, const TActorId& subscriber)
         : TBase(self)
         , TxId(txId)
+        , Subscriber(subscriber)
     {
     }
 
@@ -25,6 +26,19 @@ public:
         if (txOperator) {
             txOperator->CompleteOnAbort(*Self, ctx);
         }
+
+        if (const auto* backupTx = Self->LastCompletedBackupTransactionsByTxId.FindPtr(TxId)) {
+            auto event = MakeHolder<TEvColumnShard::TEvNotifyTxCompletionResult>(Self->TabletID(), TxId);
+            auto& opResult = *event->Record.MutableOpResult();
+            opResult = backupTx->GetOpResult();
+            ctx.Send(Subscriber, event.Release(), 0, 0);
+        } else {   // for backward compatibility (remove it after stable-26.3.1)
+            auto event = MakeHolder<TEvColumnShard::TEvNotifyTxCompletionResult>(Self->TabletID(), TxId);
+            auto& opResult = *event->Record.MutableOpResult();
+            opResult.SetSuccess(false);
+            opResult.SetExplain("Cancelled manually");
+            ctx.Send(Subscriber, event.Release(), 0, 0);
+        }
     }
 
     TTxType GetTxType() const override {
@@ -33,16 +47,17 @@ public:
 
 private:
     ui64 TxId = 0;
+    TActorId Subscriber;
 };
 
 void TColumnShard::Handle(TEvDataShard::TEvCancelBackup::TPtr& ev, const TActorContext& ctx) {
     const ui64 txId = ev->Get()->Record.GetBackupTxId();
-    Execute(new TTxTxAbort(this, txId), ctx);
+    Execute(new TTxTxAbort(this, txId, ev->Sender), ctx);
 }
 
 void TColumnShard::Handle(TEvDataShard::TEvCancelRestore::TPtr& ev, const TActorContext& ctx) {
     const ui64 txId = ev->Get()->Record.GetRestoreTxId();
-    Execute(new TTxTxAbort(this, txId), ctx);
+    Execute(new TTxTxAbort(this, txId, ev->Sender), ctx);
 }
 
 }   // namespace NKikimr::NColumnShard
diff --git a/ydb/core/tx/columnshard/columnshard_impl.cpp b/ydb/core/tx/columnshard/columnshard_impl.cpp
index 30ba92e8fce..e36d4dfda78 100644
--- a/ydb/core/tx/columnshard/columnshard_impl.cpp
+++ b/ydb/core/tx/columnshard/columnshard_impl.cpp
@@ -1755,6 +1755,7 @@ void TColumnShard::Enqueue(STFUNC_SIG) {
         HFunc(TEvPrivate::TEvAskTabletDataAccessors, Handle);
         HFunc(TEvTxProxySchemeCache::TEvWatchNotifyUpdated, Handle);
         HFunc(TEvTxProxySchemeCache::TEvWatchNotifyUnavailable, Handle);
+        HFunc(TEvColumnShard::TEvNotifyTxCompletion, Handle);
         default:
             AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "unexpected event in enqueue");
             return NTabletFlatExecutor::TTabletExecutedFlat::Enqueue(ev);
diff --git a/ydb/core/tx/columnshard/columnshard_impl.h b/ydb/core/tx/columnshard/columnshard_impl.h
index 33fd09a8613..708f3044618 100644
--- a/ydb/core/tx/columnshard/columnshard_impl.h
+++ b/ydb/core/tx/columnshard/columnshard_impl.h
@@ -546,7 +546,6 @@ private:
     ui64 LastExportNo = 0;
     THashMap<TSchemeShardLocalPathId, NKikimrTxColumnShard::TCompletedBackupTransaction> LastCompletedBackupTransactions;
     THashMap<ui64, NKikimrTxColumnShard::TCompletedBackupTransaction> LastCompletedBackupTransactionsByTxId;   // TxId -> BackupTransaction
-
     ui64 StatsReportRound = 0;
     TString OwnerPath;
 
diff --git a/ydb/core/tx/columnshard/export/actor/export_actor.cpp b/ydb/core/tx/columnshard/export/actor/export_actor.cpp
index 782fec8e8d4..191162d44a2 100644
--- a/ydb/core/tx/columnshard/export/actor/export_actor.cpp
+++ b/ydb/core/tx/columnshard/export/actor/export_actor.cpp
@@ -32,7 +32,7 @@ void TActor::ScheduleTimeoutCheck() {
 }
 
 void TActor::HandleWakeup() {
-    if (Stage == EStage::Finished) {
+    if (Stage == EStage::Finished || Stage == EStage::WaitSaveCursor) {
         return;
     }
     const auto elapsed = TInstant::Now() - StageStartTime;
@@ -56,11 +56,15 @@ void TActor::HandleWakeup() {
 }
 
 void TActor::AbortExport(const TString& errorMessage) {
-    ErrorMessage = errorMessage;
-    if (ExportSession->GetCursor().IsFinished()) {
-        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "abort_after_cursor_finished")("message", errorMessage);
+    if (ExportSession->IsFinished() || ExportSession->IsReadyForRemoveOnFinished()) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "abort_after_export_finished")("message", errorMessage);
+        return;
+    }
+    if (Stage == EStage::WaitSaveCursor) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "abort_during_save_cursor")("message", errorMessage);
         return;
     }
+    ExportSession->Abort(errorMessage);
     ExportSession->MutableCursor().InitNext({}, true);
     if (Stage == EStage::WaitData) {
         Counters.OnAckResponse();
@@ -85,6 +89,7 @@ void TActor::KillExporter() {
 }
 
 void TActor::PassAway() {
+    AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "export_actor_pass_away")("self_id", SelfId())("tablet_id", TabletId);
     KillExporter();
     Counters.OnActorDead();
     TBase::PassAway();
@@ -104,32 +109,37 @@ void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanInitActor::TPtr& ev) {
 void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanError::TPtr& ev) {
     AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "scan_error")("message", ev->Get()->Record.ShortDebugString());
     Counters.OnError();
-    if (ExportSession->GetCursor().IsFinished()) {
+    if (ExportSession->IsFinished() || ExportSession->IsReadyForRemoveOnFinished()) {
         if (Stage == EStage::WaitData) {
             Counters.OnAckResponse();
         }
         AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "scan_error_after_finish")(
-            "message", "ignoring scan error because cursor is already finished");
+            "message", "ignoring scan error because export is already finished");
         return;
     }
     AbortExport("Scan error: " + ev->Get()->Record.ShortDebugString());
 }
 
 void TActor::OnTxCompleted(const ui64 /*txId*/) {
+    AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "export_actor_on_tx_completed")("self_id", SelfId())("tablet_id", TabletId);
     Session->FinishActor();
 }
 
 void TActor::OnSessionProgressSaved() {
     Counters.OnSaveCursorFinished(TInstant::Now() - SaveCursorStartTime);
-    if (ExportSession->GetCursor().IsFinished()) {
-        if (ErrorMessage) {
-            ExportSession->Abort(ErrorMessage);
-        } else {
+    if (ExportSession->IsFinished() || ExportSession->IsReadyForRemoveOnFinished()) {
+        AFL_VERIFY(Stage == EStage::WaitData || Stage == EStage::WaitSaveCursor)("real", (ui32)Stage);
+        Stage = EStage::Finished;
+        StageStartTime = TInstant::Now();
+        SaveSessionState();
+    } else if (ExportSession->GetCursor().IsFinished()) {
+        if (ExportSession->IsStarted()) {
             ExportSession->Finish();
         }
         SaveSessionState();
         SwitchStage(EStage::WaitSaveCursor, EStage::Finished);
     } else {
+        AFL_VERIFY(Stage == EStage::WaitSaveCursor)("real", (ui32)Stage);
         SwitchStage(EStage::WaitSaveCursor, EStage::WaitData);
         AFL_VERIFY(ScanActorId);
         Counters.OnReadStarted();
@@ -150,15 +160,16 @@ void TActor::HandleExecute(NColumnShard::TEvPrivate::TEvBackupExportRecordBatchR
 
 void TActor::HandleExecute(NColumnShard::TEvPrivate::TEvBackupExportError::TPtr& ev) {
     Counters.OnError();
-    if (ExportSession->GetCursor().IsFinished()) {
+    if (ExportSession->IsFinished() || ExportSession->IsReadyForRemoveOnFinished()) {
         AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "export_error_after_finish")(
-            "message", "ignoring export error because cursor is already finished");
+            "message", "ignoring export error because export is already finished");
         return;
     }
     AbortExport(ev->Get()->ErrorMessage);
 }
 
 void TActor::OnBootstrap(const TActorContext& /*ctx*/) {
+    AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "export_actor_bootstrap")("self_id", SelfId())("tablet_id", TabletId);
     Counters.OnActorAlive();
     StageStartTime = TInstant::Now();
     ScheduleTimeoutCheck();
@@ -167,7 +178,9 @@ void TActor::OnBootstrap(const TActorContext& /*ctx*/) {
     if (ExportSession->GetCursor().IsFinished()) {
         AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "export_actor_bootstrap_cursor_finished")("tablet_id", TabletId)(
             "session_id", ExportSession->GetIdentifier().ToString());
-        ExportSession->Finish();
+        if (ExportSession->IsStarted()) {
+            ExportSession->Finish();
+        }
         SaveSessionState();
         SwitchStage(EStage::Initialization, EStage::Finished);
         Become(&TActor::StateFunc);
@@ -269,12 +282,15 @@ public:
 };
 
 void TActor::OnSessionStateSaved() {
-    AFL_VERIFY(ExportSession->IsFinished() || ExportSession->IsAborted());
+    AFL_VERIFY(ExportSession->IsFinished() || ExportSession->IsReadyForRemoveOnFinished());
     NYDBTest::TControllers::GetColumnShardController()->OnExportFinished();
     if (ExportSession->GetTxId()) {
+        AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "export_actor_session_state_saved_with_tx")("self_id", SelfId())(
+            "tablet_id", TabletId)("tx_id", *ExportSession->GetTxId());
         ExecuteTransaction(std::make_unique<TTxProposeFinish>(
             GetShardVerified<NColumnShard::TColumnShard>(), *ExportSession->GetTxId(), SelfId(), GetNextTxId()));
     } else {
+        AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "export_actor_session_state_saved_no_tx")("self_id", SelfId())("tablet_id", TabletId);
         Session->FinishActor();
     }
 }
diff --git a/ydb/core/tx/columnshard/export/actor/export_actor.h b/ydb/core/tx/columnshard/export/actor/export_actor.h
index 063e6e9293c..ee12ab69adc 100644
--- a/ydb/core/tx/columnshard/export/actor/export_actor.h
+++ b/ydb/core/tx/columnshard/export/actor/export_actor.h
@@ -32,7 +32,6 @@ private:
     EStage Stage = EStage::Initialization;
     std::shared_ptr<NExport::TSession> ExportSession;
     TActorId Exporter;
-    TString ErrorMessage;
     static inline const ui64 FreeSpace = ((ui64)128) << 20;
     static constexpr TDuration OperationTimeout = TDuration::Minutes(240);
     static constexpr TDuration WarningInterval = TDuration::Seconds(60);
diff --git a/ydb/core/tx/columnshard/export/actor/ut/ut_export_actor_lifecycle.cpp b/ydb/core/tx/columnshard/export/actor/ut/ut_export_actor_lifecycle.cpp
index eacfa70485f..a80d7f9a282 100644
--- a/ydb/core/tx/columnshard/export/actor/ut/ut_export_actor_lifecycle.cpp
+++ b/ydb/core/tx/columnshard/export/actor/ut/ut_export_actor_lifecycle.cpp
@@ -1,8 +1,11 @@
 #include <ydb/core/base/counters.h>
 #include <ydb/core/testlib/actors/block_events.h>
 #include <ydb/core/testlib/tablet_helpers.h>
+#include <ydb/core/tx/columnshard/bg_tasks/events/local.h>
 #include <ydb/core/tx/columnshard/columnshard.h>
 #include <ydb/core/tx/columnshard/columnshard_private_events.h>
+#include <ydb/core/tx/columnshard/export/session/session.h>
+#include <ydb/core/tx/columnshard/export/session/task.h>
 #include <ydb/core/tx/columnshard/hooks/abstract/abstract.h>
 #include <ydb/core/tx/columnshard/hooks/testing/controller.h>
 #include <ydb/core/tx/columnshard/operations/write_data.h>
@@ -22,6 +25,44 @@ namespace NKikimr {
 using namespace NColumnShard;
 using namespace NTxUT;
 
+Y_UNIT_TEST_SUITE(ExportSessionStateMachine) {
+    static std::shared_ptr<NOlap::NExport::TSession> MakeExportSession() {
+        NKikimrSchemeOp::TBackupTask backupTask;
+        backupTask.SetTableName("test_table");
+        backupTask.SetTableId(1);
+        backupTask.MutableS3Settings()->SetBucket("test");
+        backupTask.MutableS3Settings()->SetEndpoint("localhost:9000");
+
+        auto task =
+            std::make_shared<NOlap::NExport::TExportTask>(NOlap::NExport::TIdentifier(NColumnShard::TSchemeShardLocalPathId::FromRawValue(1)),
+                /*columns=*/std::vector<std::pair<TString, NScheme::TTypeInfo>>{}, backupTask, /*txId=*/std::optional<ui64>{ 42 });
+
+        return std::make_shared<NOlap::NExport::TSession>(task);
+    }
+
+    Y_UNIT_TEST(FinishGuardedAfterAbort) {
+        auto session = MakeExportSession();
+        session->Confirm();
+        UNIT_ASSERT(session->IsConfirmed());
+
+        session->Abort("Aborted by user");
+        UNIT_ASSERT(!session->IsStarted());
+        UNIT_ASSERT(session->IsReadyForRemoveOnFinished());
+
+        UNIT_ASSERT(!session->IsStarted());
+    }
+
+    Y_UNIT_TEST(DoubleAbortIsNoOp) {
+        auto session = MakeExportSession();
+        session->Confirm();
+        session->Abort("first abort");
+        UNIT_ASSERT(session->IsAborted());
+
+        session->Abort("second abort");
+        UNIT_ASSERT(session->IsAborted());
+    }
+}
+
 namespace {
 
 i64 GetExportActorsAliveCounter(TTestBasicRuntime& runtime) {
@@ -384,6 +425,153 @@ Y_UNIT_TEST_SUITE(TExportActorLifecycle) {
 
         blockExportBatch.Stop().Unblock();
     }
+
+    Y_UNIT_TEST(CancelDuringProgressSaveDoesNotCrash) {
+        Aws::S3::S3Client s3Client = NTestUtils::MakeS3Client();
+        NTestUtils::CreateBucket("test", s3Client);
+
+        TTestBasicRuntime runtime;
+        TTester::Setup(runtime);
+
+        const ui64 tableId = 1;
+        const std::vector<NArrow::NTest::TTestColumn> schema = { NArrow::NTest::TTestColumn("key1", TTypeInfo(NTypeIds::Uint64)),
+            NArrow::NTest::TTestColumn("key2", TTypeInfo(NTypeIds::Uint64)), NArrow::NTest::TTestColumn("field", TTypeInfo(NTypeIds::Utf8)) };
+        auto csControllerGuard = NKikimr::NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
+        auto planStep = PrepareTablet(runtime, tableId, schema, 2);
+        ui64 txId = 111;
+
+        TActorId sender = runtime.AllocateEdgeActor();
+
+        std::vector<ui64> writeIds;
+        ui64 writeId = 1;
+        UNIT_ASSERT(WriteData(runtime, sender, writeId++, tableId, MakeTestBlob({ 0, 5 }, schema), schema, true, &writeIds));
+        planStep = ProposeCommit(runtime, sender, ++txId, writeIds);
+        PlanCommit(runtime, sender, planStep, txId);
+
+        const NOlap::TSnapshot backupSnapshot(planStep.Val(), txId);
+        const auto txBody = MakeBackupTxBody(tableId, backupSnapshot);
+        const ui64 backupTxId = ++txId;
+
+        std::atomic<int> localTxCompletedCount{ 0 };
+        NActors::TBlockEvents<NOlap::NBackground::TEvLocalTransactionCompleted> blockLocalTxCompleted(runtime, [&](const auto& /*ev*/) {
+            return localTxCompletedCount.fetch_add(1) >= 1;
+        });
+
+        ForwardToTablet(runtime, TTestTxConfig::TxTablet0, sender,
+            new TEvColumnShard::TEvProposeTransaction(NKikimrTxColumnShard::TX_KIND_BACKUP, sender, backupTxId, txBody.SerializeAsString()));
+
+        TestWaitCondition(runtime, "blocked_local_tx_cursor_finished", [&]() {
+            return !blockLocalTxCompleted.empty();
+        }, TDuration::Seconds(30));
+
+        ForwardToTablet(runtime, TTestTxConfig::TxTablet0, sender, new TEvDataShard::TEvCancelBackup(backupTxId, tableId));
+
+        runtime.SimulateSleep(TDuration::MilliSeconds(100));
+
+        UNIT_ASSERT_GE(blockLocalTxCompleted.size(), 2u);
+
+        blockLocalTxCompleted.Unblock(1);
+        runtime.SimulateSleep(TDuration::MilliSeconds(100));
+
+        blockLocalTxCompleted.Stop().Unblock();
+
+        runtime.SimulateSleep(TDuration::Seconds(2));
+
+        UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+    }
+
+    Y_UNIT_TEST(PlanAfterDuplicateProposeDoesNotCrash) {
+        Aws::S3::S3Client s3Client = NTestUtils::MakeS3Client();
+        NTestUtils::CreateBucket("test", s3Client);
+
+        TTestBasicRuntime runtime;
+        TTester::Setup(runtime);
+
+        const ui64 tableId = 1;
+        const std::vector<NArrow::NTest::TTestColumn> schema = { NArrow::NTest::TTestColumn("key1", TTypeInfo(NTypeIds::Uint64)),
+            NArrow::NTest::TTestColumn("key2", TTypeInfo(NTypeIds::Uint64)), NArrow::NTest::TTestColumn("field", TTypeInfo(NTypeIds::Utf8)) };
+        auto csControllerGuard = NKikimr::NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
+        auto planStep = PrepareTablet(runtime, tableId, schema, 2);
+        ui64 txId = 111;
+
+        TActorId sender = runtime.AllocateEdgeActor();
+
+        RunBackupExport(runtime, sender, csControllerGuard, txId, planStep, tableId, schema);
+        TestWaitCondition(runtime, "export", [&]() {
+            return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
+        });
+        runtime.SimulateSleep(TDuration::Seconds(1));
+        UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+
+        const ui64 backupTxId = txId;
+
+        std::vector<ui64> writeIds;
+        ui64 writeId = 100;
+        UNIT_ASSERT(WriteData(runtime, sender, writeId++, tableId, MakeTestBlob({ 0, 5 }, schema), schema, true, &writeIds));
+        auto commitPlanStep = ProposeCommit(runtime, sender, ++txId, writeIds);
+        PlanCommit(runtime, sender, commitPlanStep, txId);
+
+        const NOlap::TSnapshot backupSnapshot(commitPlanStep.Val(), txId);
+        const auto txBody = MakeBackupTxBody(tableId, backupSnapshot);
+
+        ForwardToTablet(runtime, TTestTxConfig::TxTablet0, sender,
+            new TEvColumnShard::TEvProposeTransaction(NKikimrTxColumnShard::TX_KIND_BACKUP, sender, backupTxId, txBody.SerializeAsString()));
+        auto ev = runtime.GrabEdgeEvent<TEvColumnShard::TEvProposeTransactionResult>(sender);
+        UNIT_ASSERT_EQUAL(ev->Get()->Record.GetTxId(), backupTxId);
+        const TPlanStep dupPlanStep{ ev->Get()->Record.GetMinStep() };
+
+        PlanTx(runtime, sender, NKikimrTxColumnShard::TX_KIND_BACKUP, NOlap::TSnapshot(dupPlanStep, backupTxId), false);
+
+        runtime.SimulateSleep(TDuration::Seconds(1));
+
+        UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+    }
+
+    Y_UNIT_TEST(CancelAfterDuplicateProposeDoesNotCrash) {
+        Aws::S3::S3Client s3Client = NTestUtils::MakeS3Client();
+        NTestUtils::CreateBucket("test", s3Client);
+
+        TTestBasicRuntime runtime;
+        TTester::Setup(runtime);
+
+        const ui64 tableId = 1;
+        const std::vector<NArrow::NTest::TTestColumn> schema = { NArrow::NTest::TTestColumn("key1", TTypeInfo(NTypeIds::Uint64)),
+            NArrow::NTest::TTestColumn("key2", TTypeInfo(NTypeIds::Uint64)), NArrow::NTest::TTestColumn("field", TTypeInfo(NTypeIds::Utf8)) };
+        auto csControllerGuard = NKikimr::NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
+        auto planStep = PrepareTablet(runtime, tableId, schema, 2);
+        ui64 txId = 111;
+
+        TActorId sender = runtime.AllocateEdgeActor();
+
+        RunBackupExport(runtime, sender, csControllerGuard, txId, planStep, tableId, schema);
+        TestWaitCondition(runtime, "export", [&]() {
+            return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
+        });
+        runtime.SimulateSleep(TDuration::Seconds(1));
+        UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+
+        const ui64 backupTxId = txId;
+
+        std::vector<ui64> writeIds;
+        ui64 writeId = 100;
+        UNIT_ASSERT(WriteData(runtime, sender, writeId++, tableId, MakeTestBlob({ 0, 5 }, schema), schema, true, &writeIds));
+        auto commitPlanStep = ProposeCommit(runtime, sender, ++txId, writeIds);
+        PlanCommit(runtime, sender, commitPlanStep, txId);
+
+        const NOlap::TSnapshot backupSnapshot(commitPlanStep.Val(), txId);
+        const auto txBody = MakeBackupTxBody(tableId, backupSnapshot);
+
+        ForwardToTablet(runtime, TTestTxConfig::TxTablet0, sender,
+            new TEvColumnShard::TEvProposeTransaction(NKikimrTxColumnShard::TX_KIND_BACKUP, sender, backupTxId, txBody.SerializeAsString()));
+        auto ev = runtime.GrabEdgeEvent<TEvColumnShard::TEvProposeTransactionResult>(sender);
+        UNIT_ASSERT_EQUAL(ev->Get()->Record.GetTxId(), backupTxId);
+
+        ForwardToTablet(runtime, TTestTxConfig::TxTablet0, sender, new TEvDataShard::TEvCancelBackup(backupTxId, tableId));
+
+        runtime.SimulateSleep(TDuration::Seconds(1));
+
+        UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+    }
 }
 
 }   // namespace NKikimr
diff --git a/ydb/core/tx/columnshard/export/session/control.cpp b/ydb/core/tx/columnshard/export/session/control.cpp
index 4b4da7128af..a2f2e4ab918 100644
--- a/ydb/core/tx/columnshard/export/session/control.cpp
+++ b/ydb/core/tx/columnshard/export/session/control.cpp
@@ -13,6 +13,11 @@ NKikimr::TConclusionStatus TConfirmSessionControl::DoApply(const std::shared_ptr
 NKikimr::TConclusionStatus TAbortSessionControl::DoApply(const std::shared_ptr<NBackground::ISessionLogic>& session) const {
     auto exportSession = dynamic_pointer_cast<TSession>(session);
     AFL_VERIFY(exportSession);
+    if (exportSession->IsFinished() || exportSession->IsReadyForRemoveOnFinished()) {
+        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "abort_control_skipped_terminal_session")("is_finished", exportSession->IsFinished())(
+            "is_aborted", exportSession->IsReadyForRemoveOnFinished());
+        return TConclusionStatus::Success();
+    }
     exportSession->Abort("Aborted by user");
     return TConclusionStatus::Success();
 }
diff --git a/ydb/core/tx/columnshard/transactions/operators/backup.cpp b/ydb/core/tx/columnshard/transactions/operators/backup.cpp
index de6accfd9d6..99a0fab55e9 100644
--- a/ydb/core/tx/columnshard/transactions/operators/backup.cpp
+++ b/ydb/core/tx/columnshard/transactions/operators/backup.cpp
@@ -15,6 +15,16 @@ bool TBackupTransactionOperator::DoParse(TColumnShard& owner, const TString& dat
     if (!txBody.HasBackupTask()) {
         return false;
     }
+
+    if (const auto* completedTx = owner.LastCompletedBackupTransactionsByTxId.FindPtr(GetTxId())) {
+        if (completedTx->GetOpResult().GetSuccess()) {
+            AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "backup_already_completed")("tx_id", GetTxId())(
+                "success", completedTx->GetOpResult().GetSuccess())("explain", completedTx->GetOpResult().GetExplain());
+            AlreadyCompleted = true;
+        }
+        return true;
+    }
+
     TConclusion<NOlap::NExport::TIdentifier> id = NOlap::NExport::TIdentifier::BuildFromProto(txBody);
     if (!id) {
         AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "cannot_parse_id")("problem", id.GetErrorMessage());
@@ -32,25 +42,26 @@ bool TBackupTransactionOperator::DoParse(TColumnShard& owner, const TString& dat
         columns.emplace_back(it->second);
     }
     ExportTask = std::make_shared<NOlap::NExport::TExportTask>(id.DetachResult(), columns, txBody.GetBackupTask(), GetTxId());
+    return true;
+}
+
+TBackupTransactionOperator::TProposeResult TBackupTransactionOperator::DoStartProposeOnExecute(
+    TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc) {
+    if (AlreadyCompleted) {
+        return TProposeResult();
+    }
     NOlap::NBackground::TTask task(::ToString(ExportTask->GetIdentifier().GetSchemeShardLocalPathId().GetRawValue()),
         std::make_shared<NOlap::NBackground::TFakeStatusChannel>(), ExportTask);
     if (!owner.GetBackgroundSessionsManager()->HasTask(task)) {
         TxAddTask = owner.GetBackgroundSessionsManager()->TxAddTask(task);
         if (!TxAddTask) {
             AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "cannot_add_task");
-            return false;
+            return TProposeResult(NKikimrTxColumnShard::EResultStatus::ERROR, "Cannot add backup task");
         }
+        AFL_VERIFY(TxAddTask->Execute(txc, NActors::TActivationContext::AsActorContext()));
     } else {
         TaskExists = true;
     }
-    return true;
-}
-
-TBackupTransactionOperator::TProposeResult TBackupTransactionOperator::DoStartProposeOnExecute(
-    TColumnShard& /*owner*/, NTabletFlatExecutor::TTransactionContext& txc) {
-    if (!TaskExists && TxAddTask) {
-        AFL_VERIFY(TxAddTask->Execute(txc, NActors::TActivationContext::AsActorContext()));
-    }
     return TProposeResult();
 }
 
@@ -63,9 +74,16 @@ void TBackupTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner*
 
 bool TBackupTransactionOperator::ProgressOnExecute(
     TColumnShard& owner, const NOlap::TSnapshot& /*version*/, NTabletFlatExecutor::TTransactionContext& txc) {
+    if (AlreadyCompleted) {
+        return true;
+    }
     AFL_VERIFY(!TxRemove);
     const auto schemeShardLocalPathId = ExportTask->GetIdentifier().GetSchemeShardLocalPathId();
-    auto status = owner.GetBackgroundSessionsManager()->GetStatus(ExportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()));
+    const TString sessionId = ::ToString(schemeShardLocalPathId.GetRawValue());
+    if (!owner.GetBackgroundSessionsManager()->IsSessionComplete(ExportTask->GetClassName(), sessionId)) {
+        return true;
+    }
+    auto status = owner.GetBackgroundSessionsManager()->GetStatus(ExportTask->GetClassName(), sessionId);
 
     NKikimrTxColumnShard::TCompletedBackupTransaction backupTx;
     backupTx.SetTxId(GetTxId());
@@ -73,7 +91,7 @@ bool TBackupTransactionOperator::ProgressOnExecute(
     opResult.SetSuccess(status.Success);
     opResult.SetExplain(status.ErrorMessage);
 
-    TxRemove = owner.GetBackgroundSessionsManager()->TxRemove(ExportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()));
+    TxRemove = owner.GetBackgroundSessionsManager()->TxRemove(ExportTask->GetClassName(), sessionId);
     NIceDb::TNiceDb db(txc.DB);
 
     const auto tableId = owner.TablesManager.ResolveInternalPathIdVerified(schemeShardLocalPathId, false);
@@ -93,6 +111,9 @@ bool TBackupTransactionOperator::ProgressOnExecute(
 }
 
 bool TBackupTransactionOperator::ProgressOnComplete(TColumnShard& owner, const TActorContext& ctx) {
+    if (AlreadyCompleted) {
+        return true;
+    }
     auto status = owner.GetBackgroundSessionsManager()->GetStatus(
         ExportTask->GetClassName(), ::ToString(ExportTask->GetIdentifier().GetSchemeShardLocalPathId().GetRawValue()));
     for (TActorId subscriber : NotifySubscribers) {
@@ -108,27 +129,45 @@ bool TBackupTransactionOperator::ProgressOnComplete(TColumnShard& owner, const T
 }
 
 bool TBackupTransactionOperator::ExecuteOnAbort(TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc) {
+    if (!ExportTask) {
+        return true;
+    }
     if (!TxAbort) {
         auto control = ExportTask->BuildAbortControl();
         TxAbort = owner.GetBackgroundSessionsManager()->TxApplyControl(control);
     }
+
+    const auto schemeShardLocalPathId = ExportTask->GetIdentifier().GetSchemeShardLocalPathId();
+    NKikimrTxColumnShard::TCompletedBackupTransaction backupTx;
+    backupTx.SetTxId(GetTxId());
+    auto& opResult = *backupTx.MutableOpResult();
+    opResult.SetSuccess(false);
+    opResult.SetExplain("Cancelled");
+
+    NIceDb::TNiceDb db(txc.DB);
+    const auto tableId = owner.TablesManager.ResolveInternalPathIdVerified(schemeShardLocalPathId, false);
+    if (const auto* previousBackupTx = owner.LastCompletedBackupTransactions.FindPtr(schemeShardLocalPathId)) {
+        owner.LastCompletedBackupTransactionsByTxId.erase(previousBackupTx->GetTxId());
+    }
+    const TString serializedBackupTx = backupTx.SerializeAsString();
+    owner.LastCompletedBackupTransactions[schemeShardLocalPathId] = backupTx;
+    owner.LastCompletedBackupTransactionsByTxId[backupTx.GetTxId()] = backupTx;
+    owner.TablesManager.SetLastCompletedBackupTransaction(schemeShardLocalPathId, serializedBackupTx);
+
+    db.Table<Schema::TableInfoV1>()
+        .Key(tableId.GetRawValue(), schemeShardLocalPathId.GetRawValue())
+        .Update(NIceDb::TUpdate<Schema::TableInfoV1::LastCompletedBackupTransaction>(serializedBackupTx));
+
     if (!TxAbort) {
         return true;
     }
     return TxAbort->Execute(txc, NActors::TActivationContext::AsActorContext());
 }
 
-bool TBackupTransactionOperator::CompleteOnAbort(TColumnShard& owner, const TActorContext& ctx) {
+bool TBackupTransactionOperator::CompleteOnAbort(TColumnShard& /*owner*/, const TActorContext& ctx) {
     if (TxAbort) {
         TxAbort->Complete(ctx);
     }
-    for (TActorId subscriber : NotifySubscribers) {
-        auto event = MakeHolder<TEvColumnShard::TEvNotifyTxCompletionResult>(owner.TabletID(), GetTxId());
-        auto& opResult = *event->Record.MutableOpResult();
-        opResult.SetSuccess(false);
-        opResult.SetExplain("Cancelled");
-        ctx.Send(subscriber, event.Release(), 0, 0);
-    }
     return true;
 }
 
@@ -141,7 +180,15 @@ TString TBackupTransactionOperator::DoDebugString() const {
 }
 
 bool TBackupTransactionOperator::DoIsAsync() const {
-    return true;
+    return !AlreadyCompleted;
+}
+
+bool TBackupTransactionOperator::DoIsProposeReplyReady(TColumnShard& owner) const {
+    if (AlreadyCompleted || !ExportTask) {
+        return true;
+    }
+    const auto schemeShardLocalPathId = ExportTask->GetIdentifier().GetSchemeShardLocalPathId();
+    return owner.GetBackgroundSessionsManager()->IsSessionComplete(ExportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()));
 }
 
 TString TBackupTransactionOperator::DoGetOpType() const {
diff --git a/ydb/core/tx/columnshard/transactions/operators/backup.h b/ydb/core/tx/columnshard/transactions/operators/backup.h
index d507fc7750c..6dbe1e51472 100644
--- a/ydb/core/tx/columnshard/transactions/operators/backup.h
+++ b/ydb/core/tx/columnshard/transactions/operators/backup.h
@@ -13,6 +13,7 @@ private:
 
     std::shared_ptr<NOlap::NExport::TExportTask> ExportTask;
     bool TaskExists = false;
+    bool AlreadyCompleted = false;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxAddTask;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxRemove;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxAbort;
@@ -26,6 +27,7 @@ private:
     virtual void DoFinishProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& /*ctx*/) override;
     virtual TString DoGetOpType() const override;
     virtual bool DoIsAsync() const override;
+    virtual bool DoIsProposeReplyReady(TColumnShard& owner) const override;
     virtual bool DoParse(TColumnShard& owner, const TString& data) override;
     virtual TString DoDebugString() const override;
 
diff --git a/ydb/core/tx/columnshard/transactions/operators/restore.cpp b/ydb/core/tx/columnshard/transactions/operators/restore.cpp
index e3d6ce52f6f..48829f3d102 100644
--- a/ydb/core/tx/columnshard/transactions/operators/restore.cpp
+++ b/ydb/core/tx/columnshard/transactions/operators/restore.cpp
@@ -15,31 +15,39 @@ bool TRestoreTransactionOperator::DoParse(TColumnShard& owner, const TString& da
     if (!txBody.HasRestoreTask()) {
         return false;
     }
+
+    if (const auto* completedTx = owner.LastCompletedBackupTransactionsByTxId.FindPtr(GetTxId())) {
+        AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "restore_already_completed")("tx_id", GetTxId())(
+            "success", completedTx->GetOpResult().GetSuccess())("explain", completedTx->GetOpResult().GetExplain());
+        AlreadyCompleted = true;
+        return true;
+    }
+
     auto schema = owner.TablesManager.GetPrimaryIndex()->GetVersionedIndex().GetLastSchema();
     const auto& columns = schema->GetIndexInfo().GetColumns();
     const auto schemeShardLocalPathId = TSchemeShardLocalPathId::FromRawValue(txBody.GetRestoreTask().GetTableId());
     ImportTask =
         std::make_shared<NOlap::NImport::TImportTask>(schemeShardLocalPathId, columns, txBody.GetRestoreTask(), schema->GetVersion(), GetTxId());
-    NOlap::NBackground::TTask task(
-        ::ToString(schemeShardLocalPathId.GetRawValue()), std::make_shared<NOlap::NBackground::TFakeStatusChannel>(), ImportTask);
+    return true;
+}
+
+TRestoreTransactionOperator::TProposeResult TRestoreTransactionOperator::DoStartProposeOnExecute(
+    TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc) {
+    if (AlreadyCompleted) {
+        return TProposeResult();
+    }
+    NOlap::NBackground::TTask task(::ToString(ImportTask->GetSchemeShardLocalPathId().GetRawValue()),
+        std::make_shared<NOlap::NBackground::TFakeStatusChannel>(), ImportTask);
     if (!owner.GetBackgroundSessionsManager()->HasTask(task)) {
         TxAddTask = owner.GetBackgroundSessionsManager()->TxAddTask(task);
         if (!TxAddTask) {
             AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "cannot_add_task");
-            return false;
+            return TProposeResult(NKikimrTxColumnShard::EResultStatus::ERROR, "Cannot add restore task");
         }
+        AFL_VERIFY(TxAddTask->Execute(txc, NActors::TActivationContext::AsActorContext()));
     } else {
         TaskExists = true;
     }
-    return true;
-}
-
-TRestoreTransactionOperator::TProposeResult TRestoreTransactionOperator::DoStartProposeOnExecute(
-    TColumnShard& /*owner*/, NTabletFlatExecutor::TTransactionContext& txc) {
-    if (!TaskExists) {
-        AFL_VERIFY(!!TxAddTask);
-        AFL_VERIFY(TxAddTask->Execute(txc, NActors::TActivationContext::AsActorContext()));
-    }
     return TProposeResult();
 }
 
@@ -52,6 +60,9 @@ void TRestoreTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner
 
 bool TRestoreTransactionOperator::ProgressOnExecute(
     TColumnShard& owner, const NOlap::TSnapshot& /*version*/, NTabletFlatExecutor::TTransactionContext& txc) {
+    if (AlreadyCompleted) {
+        return true;
+    }
     AFL_VERIFY(!TxRemove);
     const auto schemeShardLocalPathId = ImportTask->GetSchemeShardLocalPathId();
     auto status = owner.GetBackgroundSessionsManager()->GetStatus(ImportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()));
@@ -66,7 +77,6 @@ bool TRestoreTransactionOperator::ProgressOnExecute(
     NIceDb::TNiceDb db(txc.DB);
 
     const auto tableId = owner.TablesManager.ResolveInternalPathIdVerified(schemeShardLocalPathId, false);
-
     if (const auto* previousBackupTx = owner.LastCompletedBackupTransactions.FindPtr(schemeShardLocalPathId)) {
         owner.LastCompletedBackupTransactionsByTxId.erase(previousBackupTx->GetTxId());
     }
@@ -83,6 +93,9 @@ bool TRestoreTransactionOperator::ProgressOnExecute(
 }
 
 bool TRestoreTransactionOperator::ProgressOnComplete(TColumnShard& owner, const TActorContext& ctx) {
+    if (AlreadyCompleted) {
+        return true;
+    }
     auto status = owner.GetBackgroundSessionsManager()->GetStatus(
         ImportTask->GetClassName(), ::ToString(ImportTask->GetSchemeShardLocalPathId().GetRawValue()));
     for (TActorId subscriber : NotifySubscribers) {
@@ -98,36 +111,66 @@ bool TRestoreTransactionOperator::ProgressOnComplete(TColumnShard& owner, const
 }
 
 bool TRestoreTransactionOperator::ExecuteOnAbort(TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc) {
+    if (!ImportTask) {
+        return true;
+    }
     if (!TxAbort) {
         auto control = ImportTask->BuildAbortControl();
         TxAbort = owner.GetBackgroundSessionsManager()->TxApplyControl(control);
     }
+
+    const auto schemeShardLocalPathId = ImportTask->GetSchemeShardLocalPathId();
+    NKikimrTxColumnShard::TCompletedBackupTransaction backupTx;
+    backupTx.SetTxId(GetTxId());
+    auto& opResult = *backupTx.MutableOpResult();
+    opResult.SetSuccess(false);
+    opResult.SetExplain("Cancelled");
+
+    NIceDb::TNiceDb db(txc.DB);
+    const auto tableId = owner.TablesManager.ResolveInternalPathIdVerified(schemeShardLocalPathId, false);
+    if (const auto* previousBackupTx = owner.LastCompletedBackupTransactions.FindPtr(schemeShardLocalPathId)) {
+        owner.LastCompletedBackupTransactionsByTxId.erase(previousBackupTx->GetTxId());
+    }
+    const TString serializedBackupTx = backupTx.SerializeAsString();
+    owner.LastCompletedBackupTransactions[schemeShardLocalPathId] = backupTx;
+    owner.LastCompletedBackupTransactionsByTxId[backupTx.GetTxId()] = backupTx;
+    owner.TablesManager.SetLastCompletedBackupTransaction(schemeShardLocalPathId, serializedBackupTx);
+
+    db.Table<Schema::TableInfoV1>()
+        .Key(tableId.GetRawValue(), schemeShardLocalPathId.GetRawValue())
+        .Update(NIceDb::TUpdate<Schema::TableInfoV1::LastCompletedBackupTransaction>(serializedBackupTx));
+
     if (!TxAbort) {
         return true;
     }
     return TxAbort->Execute(txc, NActors::TActivationContext::AsActorContext());
 }
 
-TString TRestoreTransactionOperator::DoDebugString() const {
-    return "RESTORE";
-}
-
-bool TRestoreTransactionOperator::CompleteOnAbort(TColumnShard& owner, const TActorContext& ctx) {
+bool TRestoreTransactionOperator::CompleteOnAbort(TColumnShard& /*owner*/, const TActorContext& ctx) {
     if (TxAbort) {
         TxAbort->Complete(ctx);
     }
-    for (TActorId subscriber : NotifySubscribers) {
-        auto event = MakeHolder<TEvColumnShard::TEvNotifyTxCompletionResult>(owner.TabletID(), GetTxId());
-        auto& opResult = *event->Record.MutableOpResult();
-        opResult.SetSuccess(false);
-        opResult.SetExplain("Cancelled");
-        ctx.Send(subscriber, event.Release(), 0, 0);
-    }
     return true;
 }
 
+void TRestoreTransactionOperator::RegisterSubscriber(const TActorId& actorId) {
+    NotifySubscribers.insert(actorId);
+}
+
+TString TRestoreTransactionOperator::DoDebugString() const {
+    return "RESTORE";
+}
+
 bool TRestoreTransactionOperator::DoIsAsync() const {
-    return true;
+    return !AlreadyCompleted;
+}
+
+bool TRestoreTransactionOperator::DoIsProposeReplyReady(TColumnShard& owner) const {
+    if (AlreadyCompleted || !ImportTask) {
+        return true;
+    }
+    const auto schemeShardLocalPathId = ImportTask->GetSchemeShardLocalPathId();
+    return owner.GetBackgroundSessionsManager()->IsSessionComplete(ImportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()));
 }
 
 TString TRestoreTransactionOperator::DoGetOpType() const {
@@ -140,8 +183,4 @@ void TRestoreTransactionOperator::DoFinishProposeOnComplete(TColumnShard& /*owne
 void TRestoreTransactionOperator::DoFinishProposeOnExecute(TColumnShard& /*owner*/, NTabletFlatExecutor::TTransactionContext& /*txc*/) {
 }
 
-void TRestoreTransactionOperator::RegisterSubscriber(const TActorId& actorId) {
-    NotifySubscribers.insert(actorId);
-}
-
 }   // namespace NKikimr::NColumnShard
diff --git a/ydb/core/tx/columnshard/transactions/operators/restore.h b/ydb/core/tx/columnshard/transactions/operators/restore.h
index 67d4756976d..b4b29735ae4 100644
--- a/ydb/core/tx/columnshard/transactions/operators/restore.h
+++ b/ydb/core/tx/columnshard/transactions/operators/restore.h
@@ -13,6 +13,7 @@ private:
 
     std::shared_ptr<NOlap::NImport::TImportTask> ImportTask;
     bool TaskExists = false;
+    bool AlreadyCompleted = false;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxAddTask;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxRemove;
     std::unique_ptr<NTabletFlatExecutor::ITransaction> TxAbort;
@@ -26,6 +27,7 @@ private:
     virtual void DoFinishProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& /*ctx*/) override;
     virtual TString DoGetOpType() const override;
     virtual bool DoIsAsync() const override;
+    virtual bool DoIsProposeReplyReady(TColumnShard& owner) const override;
     virtual bool DoParse(TColumnShard& owner, const TString& data) override;
     virtual TString DoDebugString() const override;
 
diff --git a/ydb/core/tx/columnshard/transactions/tx_controller.cpp b/ydb/core/tx/columnshard/transactions/tx_controller.cpp
index 72275bb2519..b2024a1135e 100644
--- a/ydb/core/tx/columnshard/transactions/tx_controller.cpp
+++ b/ydb/core/tx/columnshard/transactions/tx_controller.cpp
@@ -100,7 +100,8 @@ bool TTxController::Load(NTabletFlatExecutor::TTransactionContext& txc) {
 std::shared_ptr<TTxController::ITransactionOperator> TTxController::UpdateTxSourceInfo(
     const TFullTxInfo& tx, NTabletFlatExecutor::TTransactionContext& txc) {
     auto op = GetTxOperatorVerified(tx.GetTxId());
-    op->ResetStatusOnUpdate();
+    const bool sourceChanged = op->GetTxInfo().Source != tx.Source;
+    op->ResetStatusOnUpdate(sourceChanged);
     auto& txInfo = op->MutableTxInfo();
     txInfo.Source = tx.Source;
     txInfo.MinStep = tx.MinStep;
@@ -384,10 +385,12 @@ void TTxController::FinishProposeOnComplete(ITransactionOperator& txOperator, co
     NActors::TLogContextGuard lGuard =
         NActors::TLogContextBuilder::Build()("method", "TTxController::FinishProposeOnComplete")("tx_id", txOperator.GetTxId());
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_TX)("event", "start")("tx_info", txOperator.GetTxInfo().DebugString());
-    TTxController::TProposeResult proposeResult = txOperator.GetProposeStartInfoVerified();
     AFL_VERIFY(!txOperator.IsFail());
+    const bool shouldSendReply = txOperator.ShouldSendReplyOnComplete();
     txOperator.FinishProposeOnComplete(Owner, ctx);
-    txOperator.SendReply(Owner, ctx);
+    if (shouldSendReply) {
+        txOperator.SendReply(Owner, ctx);
+    }
     Counters.OnFinishProposeOnComplete(txOperator.GetOpType());
 }
 
diff --git a/ydb/core/tx/columnshard/transactions/tx_controller.h b/ydb/core/tx/columnshard/transactions/tx_controller.h
index ccd4a976cac..181bc0dd1c4 100644
--- a/ydb/core/tx/columnshard/transactions/tx_controller.h
+++ b/ydb/core/tx/columnshard/transactions/tx_controller.h
@@ -235,6 +235,10 @@ public:
             return false;
         }
 
+        virtual bool DoIsProposeReplyReady(TColumnShard& /*owner*/) const {
+            return true;
+        }
+
         virtual std::unique_ptr<NTabletFlatExecutor::ITransaction> DoBuildTxPrepareForProgress(TColumnShard* /*owner*/) const {
             return nullptr;
         }
@@ -248,8 +252,22 @@ public:
         virtual void DoOnTabletInit(TColumnShard& /*owner*/) {
         }
 
-        void ResetStatusOnUpdate() {
-            if (Status && *Status == EStatus::ReplySent) {
+        void ResetStatusOnUpdate(const bool sourceChanged) {
+            if (Status) {
+                switch (*Status) {
+                    case EStatus::ReplySent:
+                        NeedResendReplyFlag = sourceChanged;
+                        return;
+                    case EStatus::ProposeStartedOnExecute:
+                    case EStatus::ProposeFinishedOnExecute:
+                    case EStatus::ProposeFinishedOnComplete:
+                        NeedResendReplyFlag = true;
+                        break;
+                    default:
+                        break;
+                }
+            } else if (ProposeStartInfo) {
+                // Transaction was loaded from DB after tablet restart; propose had already been accepted.
                 NeedResendReplyFlag = true;
             }
             Status = {};
@@ -270,6 +288,14 @@ public:
             return NeedResendReplyFlag;
         }
 
+        bool ShouldSendReplyOnComplete() const {
+            return Status != EStatus::ReplySent || NeedResendReplyFlag;
+        }
+
+        bool IsProposeReplyReady(TColumnShard& owner) const {
+            return DoIsProposeReplyReady(owner);
+        }
+
         bool PingTimeout(TColumnShard& owner, const TMonotonic now) {
             return DoPingTimeout(owner, now);
         }
@@ -370,6 +396,8 @@ public:
         void SendReply(TColumnShard& owner, const TActorContext& ctx) {
             // It means that we had already processed this event
             if (Status == EStatus::ReplySent) {
+                AFL_VERIFY(NeedResendReplyFlag);
+                NeedResendReplyFlag = false;
                 return DoSendReply(owner, ctx);
             }
             AFL_VERIFY(!!ProposeStartInfo);
@@ -378,6 +406,7 @@ public:
             } else {
                 SwitchStateVerified(EStatus::ProposeFinishedOnComplete, EStatus::ReplySent);
             }
+            NeedResendReplyFlag = false;
             return DoSendReply(owner, ctx);
         }