simplify tx (#45970)

17 files changed, 204 insertions, 89 deletions

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 24f477a1404..369726d2ddf 100644
--- a/ydb/core/tx/columnshard/backup/import/ut/ut_import.cpp
+++ b/ydb/core/tx/columnshard/backup/import/ut/ut_import.cpp
@@ -202,10 +202,20 @@ Y_UNIT_TEST_SUITE(Restore) {
             AFL_VERIFY(csControllerGuard->GetFinishedExportsCount() == 0);
             planStep = ProposeTx(runtime, sender, NKikimrTxColumnShard::TX_KIND_BACKUP, txBody.SerializeAsString(), ++txId);
             AFL_VERIFY(csControllerGuard->GetFinishedExportsCount() == 1);
+            {
+                auto evSubscribe = std::make_unique<TEvColumnShard::TEvNotifyTxCompletion>(txId);
+                ForwardToTablet(runtime, TTestTxConfig::TxTablet0, sender, evSubscribe.release());
+            }
             PlanTx(runtime, sender, NKikimrTxColumnShard::TX_KIND_BACKUP, NOlap::TSnapshot(planStep, txId), false);
+            {
+                auto ev = runtime.GrabEdgeEvent<TEvColumnShard::TEvNotifyTxCompletionResult>(sender);
+                UNIT_ASSERT(ev);
+                UNIT_ASSERT(ev->Get()->Record.GetOpResult().GetSuccess());
+            }
             TestWaitCondition(runtime, "export", [&]() {
                 return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
             });
+            VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
         }
 
         // restore
@@ -257,10 +267,20 @@ Y_UNIT_TEST_SUITE(Restore) {
             AFL_VERIFY(csControllerGuard->GetFinishedImportsCount() == 0);
             planStep = ProposeTx(runtime, sender, NKikimrTxColumnShard::TX_KIND_RESTORE, txBody.SerializeAsString(), ++txId);
             AFL_VERIFY(csControllerGuard->GetFinishedImportsCount() == 1);
+            {
+                auto evSubscribe = std::make_unique<TEvColumnShard::TEvNotifyTxCompletion>(txId);
+                ForwardToTablet(runtime, TTestTxConfig::TxTablet0, sender, evSubscribe.release());
+            }
             PlanTx(runtime, sender, NKikimrTxColumnShard::TX_KIND_RESTORE, NOlap::TSnapshot(planStep, txId), false);
+            {
+                auto ev = runtime.GrabEdgeEvent<TEvColumnShard::TEvNotifyTxCompletionResult>(sender);
+                UNIT_ASSERT(ev);
+                UNIT_ASSERT(ev->Get()->Record.GetOpResult().GetSuccess());
+            }
             TestWaitCondition(runtime, "import", [&]() {
                 return true;
             });
+            VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
 
             {
                 YDB_LOG_CREATE_CONTEXT_COMP(NKikimrServices::TX_COLUMNSHARD,
diff --git a/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp b/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
index b64ec229121..d75f16fe537 100644
--- a/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
+++ b/ydb/core/tx/columnshard/columnshard__propose_transaction.cpp
@@ -119,8 +119,7 @@ public:
 
         YDB_LOG_DEBUG("",
             {"event", "actual tx operator"});
-        const bool waitForAsyncCompletion = internalOp->IsAsync() && (!internalOp->NeedResendReply() || !internalOp->IsProposeReplyReady(*Self));
-        if (waitForAsyncCompletion) {
+        if (internalOp->IsAsync()) {
             Self->GetProgressTxController().StartProposeOnComplete(*internalOp, ctx);
         } else {
             Self->GetProgressTxController().FinishProposeOnComplete(*internalOp, ctx);
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 a80d7f9a282..1468e57eb5f 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
@@ -238,6 +238,7 @@ Y_UNIT_TEST_SUITE(TExportActorLifecycle) {
         runtime.SimulateSleep(TDuration::Seconds(1));
 
         UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST(AliveCounterReturnsToZeroAfterExportCancel) {
@@ -281,6 +282,7 @@ Y_UNIT_TEST_SUITE(TExportActorLifecycle) {
         runtime.SimulateSleep(TDuration::Seconds(1));
 
         UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST(SaveSessionProgressSurvivesTabletReboot) {
@@ -318,6 +320,7 @@ Y_UNIT_TEST_SUITE(TExportActorLifecycle) {
         runtime.SimulateSleep(TDuration::Seconds(1));
 
         UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST(RepeatedCopyExportDropCycle) {
@@ -366,6 +369,7 @@ Y_UNIT_TEST_SUITE(TExportActorLifecycle) {
 
         UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
         UNIT_ASSERT_C(csControllerGuard->GetFinishedExportsCount() >= kCycleCount, "expected at least " << kCycleCount << " finished exports");
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST(CancelDuringSlowS3Operations) {
@@ -422,6 +426,7 @@ Y_UNIT_TEST_SUITE(TExportActorLifecycle) {
         runtime.SimulateSleep(TDuration::Seconds(1));
 
         UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
 
         blockExportBatch.Stop().Unblock();
     }
@@ -525,6 +530,7 @@ Y_UNIT_TEST_SUITE(TExportActorLifecycle) {
         runtime.SimulateSleep(TDuration::Seconds(1));
 
         UNIT_ASSERT_VALUES_EQUAL(GetExportActorsAliveCounter(runtime), 0);
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST(CancelAfterDuplicateProposeDoesNotCrash) {
diff --git a/ydb/core/tx/columnshard/hooks/testing/controller.cpp b/ydb/core/tx/columnshard/hooks/testing/controller.cpp
index 5239fee17b9..c7c3e13d00f 100644
--- a/ydb/core/tx/columnshard/hooks/testing/controller.cpp
+++ b/ydb/core/tx/columnshard/hooks/testing/controller.cpp
@@ -1,5 +1,6 @@
 #include "controller.h"
 
+#include <ydb/core/tx/columnshard/bg_tasks/manager/manager.h>
 #include <ydb/core/tx/columnshard/blobs_action/abstract/gc.h>
 #include <ydb/core/tx/columnshard/columnshard_impl.h>
 #include <ydb/core/tx/columnshard/engines/changes/compaction.h>
@@ -140,4 +141,24 @@ void TController::OnAfterLocalTxCommitted(
     }
 }
 
+ui32 TController::GetBackgroundSessionsCount() const {
+    TGuard<TMutex> g(Mutex);
+    ui32 count = 0;
+    for (auto&& i : ShardActuals) {
+        if (auto mgr = i.second->GetBackgroundSessionsManager()) {
+            count += mgr->GetSessionsInfoForReport().size();
+        }
+    }
+    return count;
+}
+
+ui32 TController::GetTxOperatorsCount() const {
+    TGuard<TMutex> g(Mutex);
+    ui32 count = 0;
+    for (auto&& i : ShardActuals) {
+        count += i.second->GetProgressTxController().GetTxs().size();
+    }
+    return count;
+}
+
 }   // namespace NKikimr::NYDBTest::NColumnShard
diff --git a/ydb/core/tx/columnshard/hooks/testing/controller.h b/ydb/core/tx/columnshard/hooks/testing/controller.h
index 7d9862c0146..411e1150b78 100644
--- a/ydb/core/tx/columnshard/hooks/testing/controller.h
+++ b/ydb/core/tx/columnshard/hooks/testing/controller.h
@@ -367,6 +367,10 @@ public:
         return result;
     }
 
+    ui32 GetBackgroundSessionsCount() const;
+
+    ui32 GetTxOperatorsCount() const;
+
     void SetExpectedShardsCount(const ui32 value) {
         ExpectedShardsCount = value;
     }
diff --git a/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.cpp b/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.cpp
index b7e0c235316..9db1dfa9863 100644
--- a/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.cpp
+++ b/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.cpp
@@ -12,6 +12,8 @@
 #include <ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/sys_view/portions/schema.h>
 #include <ydb/core/tx/columnshard/engines/storage/indexes/max/meta.h>
 #include <ydb/core/tx/columnshard/hooks/testing/controller.h>
+#include <ydb/core/tx/columnshard/transactions/operators/backup.h>
+#include <ydb/core/tx/columnshard/transactions/operators/restore.h>
 #include <ydb/core/tx/data_events/common/modification_type.h>
 #include <ydb/core/tx/data_events/payload_helper.h>
 #include <ydb/core/tx/long_tx_service/public/snapshot_registry.h>
@@ -19,6 +21,8 @@
 #include <ydb/core/tx/tiering/tier/object.h>
 #include <ydb/core/tx/tx_processing.h>
 
+#include <ydb/public/lib/value/value.h>
+
 #include <library/cpp/testing/unittest/registar.h>
 
 namespace NKikimr::NTxUT {
@@ -516,6 +520,80 @@ void TTestSchema::InitSchema(const std::vector<NArrow::NTest::TTestColumn>& colu
     }
 }
 
+namespace {
+
+NKikimrMiniKQL::TResult LocalMiniKQL(TTestBasicRuntime& runtime, ui64 tabletId, const TString& query) {
+    TActorId sender = runtime.AllocateEdgeActor();
+
+    auto evTx = new TEvTablet::TEvLocalMKQL;
+    evTx->Record.MutableProgram()->MutableProgram()->SetText(query);
+    ForwardToTablet(runtime, tabletId, sender, evTx);
+
+    auto event = runtime.GrabEdgeEvent<TEvTablet::TEvLocalMKQLResponse>(sender);
+    UNIT_ASSERT(event);
+    UNIT_ASSERT_VALUES_EQUAL_C(event->Get()->Record.GetStatus(), NKikimrProto::OK, event->Get()->Record.GetMiniKQLErrors());
+    return event->Get()->Record.GetExecutionEngineEvaluatedResponse();
+}
+
+}   // namespace
+
+ui64 CountLocalDbTableRows(
+    TTestBasicRuntime& runtime, ui64 tabletId, const TString& tableName, const TString& rangeSpec, const TString& fieldsSpec) {
+    const TString query = Sprintf(R"(
+        (
+            (let range %s)
+            (let fields %s)
+            (return (AsList
+                (SetResult 'Result (SelectRange '%s range fields '()))
+            ))
+        )
+    )", rangeSpec.c_str(), fieldsSpec.c_str(), tableName.c_str());
+    const auto result = LocalMiniKQL(runtime, tabletId, query);
+    return NClient::TValue::Create(result)[0]["List"].Size();
+}
+
+ui64 CountTxInfoRows(TTestBasicRuntime& runtime, ui64 tabletId) {
+    const auto result = LocalMiniKQL(runtime, tabletId, R"(
+        (
+            (let range '(
+                '('TxId (Null) (Void))
+            ))
+            (let fields '('TxId))
+            (return (AsList
+                (SetResult 'Result (SelectRange 'TxInfo range fields '()))
+            ))
+        )
+    )");
+    return NClient::TValue::Create(result)[0]["List"].Size();
+}
+
+ui64 CountBackgroundSessionsRows(TTestBasicRuntime& runtime, ui64 tabletId) {
+    const auto result = LocalMiniKQL(runtime, tabletId, R"(
+        (
+            (let range '(
+                '('ClassName (Null) (Void))
+                '('Identifier (Null) (Void))
+            ))
+            (let fields '('ClassName))
+            (return (AsList
+                (SetResult 'Result (SelectRange 'BackgroundSessions range fields '()))
+            ))
+        )
+    )");
+    return NClient::TValue::Create(result)[0]["List"].Size();
+}
+
+void VerifyNoBackupOrRestoreArtifacts(TTestBasicRuntime& runtime, const NYDBTest::NColumnShard::TController* csController, ui64 tabletId) {
+    UNIT_ASSERT(csController);
+    UNIT_ASSERT_VALUES_EQUAL(NColumnShard::TBackupTransactionOperator::GetCounter().Val(), 0);
+    UNIT_ASSERT_VALUES_EQUAL(NColumnShard::TRestoreTransactionOperator::GetCounter().Val(), 0);
+    UNIT_ASSERT_VALUES_EQUAL(csController->GetBackgroundSessionsCount(), 0);
+    UNIT_ASSERT_VALUES_EQUAL(csController->GetTxOperatorsCount(), 0);
+
+    UNIT_ASSERT_VALUES_EQUAL(CountTxInfoRows(runtime, tabletId), 0u);
+    UNIT_ASSERT_VALUES_EQUAL(CountBackgroundSessionsRows(runtime, tabletId), 0u);
+}
+
 }   // namespace NKikimr::NTxUT
 
 namespace NKikimr::NColumnShard {
diff --git a/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.h b/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.h
index 5f986410bc5..c14b763390f 100644
--- a/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.h
+++ b/ydb/core/tx/columnshard/test_helper/columnshard_ut_common.h
@@ -9,6 +9,7 @@
 #include <ydb/core/tx/columnshard/blob_cache.h>
 #include <ydb/core/tx/columnshard/common/path_id.h>
 #include <ydb/core/tx/columnshard/common/snapshot.h>
+#include <ydb/core/tx/columnshard/hooks/testing/controller.h>
 #include <ydb/core/tx/columnshard/test_helper/helper.h>
 #include <ydb/core/tx/data_events/common/modification_type.h>
 #include <ydb/core/tx/long_tx_service/public/types.h>
@@ -497,6 +498,12 @@ inline void PlanCommit(TTestBasicRuntime& runtime, TActorId& sender, TPlanStep p
 
 void Wakeup(TTestBasicRuntime& runtime, const TActorId& sender, const ui64 shardId);
 
+ui64 CountLocalDbTableRows(
+    TTestBasicRuntime& runtime, ui64 tabletId, const TString& tableName, const TString& rangeSpec, const TString& fieldsSpec);
+
+void VerifyNoBackupOrRestoreArtifacts(
+    TTestBasicRuntime& runtime, const NYDBTest::NColumnShard::TController* csController, ui64 tabletId = TTestTxConfig::TxTablet0);
+
 struct TTestBlobOptions {
     THashSet<TString> NullColumns;
     THashSet<TString> SameValueColumns;
diff --git a/ydb/core/tx/columnshard/test_helper/ya.make b/ydb/core/tx/columnshard/test_helper/ya.make
index 50594b55bf8..86a4e1a3b80 100644
--- a/ydb/core/tx/columnshard/test_helper/ya.make
+++ b/ydb/core/tx/columnshard/test_helper/ya.make
@@ -14,6 +14,7 @@ PEERDIR(
     ydb/core/tx/columnshard
     ydb/core/tx/long_tx_service/public
     ydb/core/wrappers
+    ydb/public/lib/value
 )
 
 SRCS(
diff --git a/ydb/core/tx/columnshard/transactions/operators/backup.cpp b/ydb/core/tx/columnshard/transactions/operators/backup.cpp
index 96b3bda8c45..dd6e8bcd45b 100644
--- a/ydb/core/tx/columnshard/transactions/operators/backup.cpp
+++ b/ydb/core/tx/columnshard/transactions/operators/backup.cpp
@@ -4,6 +4,7 @@
 #include <ydb/core/tx/columnshard/bg_tasks/manager/manager.h>
 #include <ydb/core/tx/columnshard/common/snapshot.h>
 #include <ydb/core/tx/columnshard/common/tablet_id.h>
+#include <ydb/core/tx/columnshard/transactions/transactions/tx_finish_async.h>
 
 #define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::TX_COLUMNSHARD
 
@@ -19,14 +20,12 @@ bool TBackupTransactionOperator::DoParse(TColumnShard& owner, const TString& dat
     }
 
     if (const auto* completedTx = owner.LastCompletedBackupTransactionsByTxId.FindPtr(GetTxId())) {
-        if (completedTx->GetOpResult().GetSuccess()) {
-            YDB_LOG_INFO("",
-                {"event", "backup_already_completed"},
-                {"txId", GetTxId()},
-                {"success", completedTx->GetOpResult().GetSuccess()},
-                {"explain", completedTx->GetOpResult().GetExplain()});
-            AlreadyCompleted = true;
-        }
+        YDB_LOG_INFO("",
+            {"event", "backup_already_completed"},
+            {"txId", GetTxId()},
+            {"success", completedTx->GetOpResult().GetSuccess()},
+            {"explain", completedTx->GetOpResult().GetExplain()});
+        AlreadyCompleted = true;
         return true;
     }
 
@@ -73,11 +72,25 @@ TBackupTransactionOperator::TProposeResult TBackupTransactionOperator::DoStartPr
     return TProposeResult();
 }
 
-void TBackupTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& ctx) {
+void TBackupTransactionOperator::DoStartProposeOnComplete(TColumnShard& owner, const TActorContext& ctx) {
     if (!TaskExists && TxAddTask) {
         TxAddTask->Complete(ctx);
         TxAddTask.reset();
     }
+
+    if (AlreadyCompleted) {
+        owner.Execute(new TTxFinishAsyncTransaction(owner, GetTxId()), ctx);
+        return;
+    }
+    if (!ExportTask) {
+        return;
+    }
+    const auto pathId = ExportTask->GetIdentifier().GetSchemeShardLocalPathId();
+    if (!owner.GetBackgroundSessionsManager()->IsSessionComplete(ExportTask->GetClassName(), ::ToString(pathId.GetRawValue()))) {
+        return;
+    }
+    AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "backup_session_complete_finish_async_propose")("tx_id", GetTxId());
+    owner.Execute(new TTxFinishAsyncTransaction(owner, GetTxId()), ctx);
 }
 
 bool TBackupTransactionOperator::ProgressOnExecute(
@@ -87,11 +100,7 @@ bool TBackupTransactionOperator::ProgressOnExecute(
     }
     AFL_VERIFY(!TxRemove);
     const auto schemeShardLocalPathId = ExportTask->GetIdentifier().GetSchemeShardLocalPathId();
-    const TString sessionId = ::ToString(schemeShardLocalPathId.GetRawValue());
-    if (!owner.GetBackgroundSessionsManager()->IsSessionComplete(ExportTask->GetClassName(), sessionId)) {
-        return true;
-    }
-    auto status = owner.GetBackgroundSessionsManager()->GetStatus(ExportTask->GetClassName(), sessionId);
+    auto status = owner.GetBackgroundSessionsManager()->GetStatus(ExportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()));
 
     NKikimrTxColumnShard::TCompletedBackupTransaction backupTx;
     backupTx.SetTxId(GetTxId());
@@ -99,7 +108,7 @@ bool TBackupTransactionOperator::ProgressOnExecute(
     opResult.SetSuccess(status.Success);
     opResult.SetExplain(status.ErrorMessage);
 
-    TxRemove = owner.GetBackgroundSessionsManager()->TxRemove(ExportTask->GetClassName(), sessionId);
+    TxRemove = owner.GetBackgroundSessionsManager()->TxRemove(ExportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()));
     NIceDb::TNiceDb db(txc.DB);
 
     const auto tableId = owner.TablesManager.ResolveInternalPathIdVerified(schemeShardLocalPathId, false);
@@ -188,15 +197,7 @@ TString TBackupTransactionOperator::DoDebugString() const {
 }
 
 bool TBackupTransactionOperator::DoIsAsync() const {
-    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()));
+    return true;
 }
 
 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 6dbe1e51472..560c4af1a23 100644
--- a/ydb/core/tx/columnshard/transactions/operators/backup.h
+++ b/ydb/core/tx/columnshard/transactions/operators/backup.h
@@ -27,7 +27,6 @@ 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 032dcf8ce40..8ccf5d598d8 100644
--- a/ydb/core/tx/columnshard/transactions/operators/restore.cpp
+++ b/ydb/core/tx/columnshard/transactions/operators/restore.cpp
@@ -4,6 +4,7 @@
 #include <ydb/core/protos/tx_columnshard.pb.h>
 #include <ydb/core/tx/columnshard/bg_tasks/manager/manager.h>
 #include <ydb/core/tx/columnshard/common/tablet_id.h>
+#include <ydb/core/tx/columnshard/transactions/transactions/tx_finish_async.h>
 
 #define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::TX_COLUMNSHARD
 
@@ -57,11 +58,25 @@ TRestoreTransactionOperator::TProposeResult TRestoreTransactionOperator::DoStart
     return TProposeResult();
 }
 
-void TRestoreTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& ctx) {
+void TRestoreTransactionOperator::DoStartProposeOnComplete(TColumnShard& owner, const TActorContext& ctx) {
     if (!TaskExists && TxAddTask) {
         TxAddTask->Complete(ctx);
         TxAddTask.reset();
     }
+
+    if (AlreadyCompleted) {
+        owner.Execute(new TTxFinishAsyncTransaction(owner, GetTxId()), ctx);
+        return;
+    }
+    if (!ImportTask) {
+        return;
+    }
+    const auto schemeShardLocalPathId = ImportTask->GetSchemeShardLocalPathId();
+    if (!owner.GetBackgroundSessionsManager()->IsSessionComplete(ImportTask->GetClassName(), ::ToString(schemeShardLocalPathId.GetRawValue()))) {
+        return;
+    }
+    AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "restore_session_complete_finish_async_propose")("tx_id", GetTxId());
+    owner.Execute(new TTxFinishAsyncTransaction(owner, GetTxId()), ctx);
 }
 
 bool TRestoreTransactionOperator::ProgressOnExecute(
@@ -168,15 +183,7 @@ TString TRestoreTransactionOperator::DoDebugString() const {
 }
 
 bool TRestoreTransactionOperator::DoIsAsync() const {
-    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()));
+    return true;
 }
 
 TString TRestoreTransactionOperator::DoGetOpType() const {
diff --git a/ydb/core/tx/columnshard/transactions/operators/restore.h b/ydb/core/tx/columnshard/transactions/operators/restore.h
index b4b29735ae4..dfd4648c3f5 100644
--- a/ydb/core/tx/columnshard/transactions/operators/restore.h
+++ b/ydb/core/tx/columnshard/transactions/operators/restore.h
@@ -27,7 +27,6 @@ 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/schema.cpp b/ydb/core/tx/columnshard/transactions/operators/schema.cpp
index b72be58c596..63823336002 100644
--- a/ydb/core/tx/columnshard/transactions/operators/schema.cpp
+++ b/ydb/core/tx/columnshard/transactions/operators/schema.cpp
@@ -347,12 +347,7 @@ void TSchemaTransactionOperator::DoOnTabletInit(TColumnShard& owner) {
             {"event", "wait_on_propose"},
             {"txId", GetTxId()});
         owner.Subscribers->RegisterSubscriber(WaitOnPropose);
-    } else {
-        YDB_LOG_WARN("",
-            {"event", "remove_pathes_cleaned"},
-            {"txId", GetTxId()});
-        owner.Execute(new TTxFinishAsyncTransaction(owner, GetTxId()));
-    }
+    }   // else we need to wait for SS resend
 }
 
 void TSchemaTransactionOperator::DoStartProposeOnComplete(TColumnShard& /*owner*/, const TActorContext& /*ctx*/) {
diff --git a/ydb/core/tx/columnshard/transactions/tx_controller.cpp b/ydb/core/tx/columnshard/transactions/tx_controller.cpp
index 898e2ac2fa3..f523b6b8480 100644
--- a/ydb/core/tx/columnshard/transactions/tx_controller.cpp
+++ b/ydb/core/tx/columnshard/transactions/tx_controller.cpp
@@ -112,9 +112,7 @@ bool TTxController::Load(NTabletFlatExecutor::TTransactionContext& txc) {
 std::shared_ptr<TTxController::ITransactionOperator> TTxController::UpdateTxSourceInfo(
     const TFullTxInfo& tx, NTabletFlatExecutor::TTransactionContext& txc) {
     auto op = GetTxOperator(tx.GetTxId(), ETxOperatorStatus::InProgress);
-    const bool sourceChanged = op->GetTxInfo().Source != tx.Source;
-    op->ResetStatusOnUpdate(sourceChanged);
-
+    op->ResetStatusOnUpdate();
     auto& txInfo = op->MutableTxInfo();
     txInfo.Source = tx.Source;
     txInfo.MinStep = tx.MinStep;
@@ -489,11 +487,8 @@ void TTxController::FinishProposeOnComplete(ITransactionOperator& txOperator, co
         {"event", "start"},
         {"txInfo", txOperator.GetTxInfo().DebugString()});
     AFL_VERIFY(!txOperator.IsFail());
-    const bool shouldSendReply = txOperator.ShouldSendReplyOnComplete();
     txOperator.FinishProposeOnComplete(Owner, ctx);
-    if (shouldSendReply) {
-        txOperator.SendReply(Owner, ctx);
-    }
+    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 ae19dd31237..51835c485c8 100644
--- a/ydb/core/tx/columnshard/transactions/tx_controller.h
+++ b/ydb/core/tx/columnshard/transactions/tx_controller.h
@@ -217,7 +217,6 @@ public:
 
     private:
         mutable TAtomicCounter PreparationsStarted = 0;
-        bool NeedResendReplyFlag = false;
         std::optional<bool> StartedAsync;
 
         friend class TTxController;
@@ -243,10 +242,6 @@ public:
             return false;
         }
 
-        virtual bool DoIsProposeReplyReady(TColumnShard& /*owner*/) const {
-            return true;
-        }
-
         virtual std::unique_ptr<NTabletFlatExecutor::ITransaction> DoBuildTxPrepareForProgress(TColumnShard* /*owner*/) const {
             return nullptr;
         }
@@ -260,24 +255,7 @@ public:
         virtual void DoOnTabletInit(TColumnShard& /*owner*/) {
         }
 
-        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;
-            }
+        void ResetStatusOnUpdate() {
             Status = {};
         }
 
@@ -292,18 +270,6 @@ public:
             return DoIsInProgress();
         }
 
-        bool NeedResendReply() const {
-            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);
         }
@@ -407,8 +373,6 @@ 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);
@@ -417,7 +381,6 @@ public:
             } else {
                 SwitchStateVerified(EStatus::ProposeFinishedOnComplete, EStatus::ReplySent);
             }
-            NeedResendReplyFlag = false;
             return DoSendReply(owner, ctx);
         }
 
diff --git a/ydb/core/tx/columnshard/ut_rw/ut_backup.cpp b/ydb/core/tx/columnshard/ut_rw/ut_backup.cpp
index 27c9289bc92..e9e56387c68 100644
--- a/ydb/core/tx/columnshard/ut_rw/ut_backup.cpp
+++ b/ydb/core/tx/columnshard/ut_rw/ut_backup.cpp
@@ -196,10 +196,14 @@ Y_UNIT_TEST_SUITE(Backup) {
         AFL_VERIFY(csControllerGuard->GetFinishedExportsCount() == 0);
         planStep = ProposeTx(runtime, sender, NKikimrTxColumnShard::TX_KIND_BACKUP, txBody.SerializeAsString(), ++txId);
         AFL_VERIFY(csControllerGuard->GetFinishedExportsCount() == 1);
+        SubscribeTxCompletion(runtime, sender, txId);
         PlanTx(runtime, sender, NOlap::TSnapshot(planStep, txId));
+        WaitNotifyTxCompletions(runtime, sender, { txId });
         TestWaitCondition(runtime, "export", [&]() {
             return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
         });
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST(ParallelBackupWithPerTableTracking) {
@@ -376,6 +380,8 @@ Y_UNIT_TEST_SUITE(Backup) {
             // Verify that the TxIds are different (per-table tracking)
             UNIT_ASSERT_UNEQUAL(backupTxId1, backupTxId2);
         }
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 }
 
diff --git a/ydb/core/tx/columnshard/ut_schema/ut_columnshard_backup.cpp b/ydb/core/tx/columnshard/ut_schema/ut_columnshard_backup.cpp
index 33949b8e8ea..914fe2f822e 100644
--- a/ydb/core/tx/columnshard/ut_schema/ut_columnshard_backup.cpp
+++ b/ydb/core/tx/columnshard/ut_schema/ut_columnshard_backup.cpp
@@ -183,6 +183,8 @@ Y_UNIT_TEST_SUITE(BackupWithRestart) {
         TestWaitCondition(runtime, "export", [&]() {
             return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
         });
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST_DUO(BackupWithUncommittedData, Reboot) {
@@ -274,6 +276,8 @@ Y_UNIT_TEST_SUITE(BackupWithRestart) {
         TestWaitCondition(runtime, "export", [&]() {
             return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
         });
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST_DUO(BackupWithMultipleRestarts, Reboot) {
@@ -359,6 +363,8 @@ Y_UNIT_TEST_SUITE(BackupWithRestart) {
         TestWaitCondition(runtime, "export", [&]() {
             return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
         });
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST_DUO(BackupDuringPropose, Reboot) {
@@ -424,6 +430,8 @@ Y_UNIT_TEST_SUITE(BackupWithRestart) {
         TestWaitCondition(runtime, "export", [&]() {
             return NTestUtils::GetObjectKeys("test", s3Client).size() == 3;
         });
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST_DUO(BackupWithSchemaVerification, Reboot) {
@@ -498,6 +506,8 @@ Y_UNIT_TEST_SUITE(BackupWithRestart) {
         // Verify that the export completed successfully even after reboot
         auto keys = NTestUtils::GetObjectKeys("test", s3Client);
         UNIT_ASSERT_EQUAL(keys.size(), 3);
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST_DUO(BackupWithComplexSchema, Reboot) {
@@ -587,6 +597,8 @@ Y_UNIT_TEST_SUITE(BackupWithRestart) {
         // Verify export completed
         auto keys = NTestUtils::GetObjectKeys("test", s3Client);
         UNIT_ASSERT_EQUAL(keys.size(), 3);
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 
     Y_UNIT_TEST_DUO(BackupDuringExportExecution, Reboot) {
@@ -669,6 +681,8 @@ Y_UNIT_TEST_SUITE(BackupWithRestart) {
         // Verify export completed
         auto keys = NTestUtils::GetObjectKeys("test", s3Client);
         UNIT_ASSERT_EQUAL(keys.size(), 3);
+
+        VerifyNoBackupOrRestoreArtifacts(runtime, csControllerGuard.operator->());
     }
 }