[BACKPORT 2024.1.3][#24394] docdb: Fix heap-use-after-free in Transac…

…tionParticipant::Impl::SubmitUpdateTransaction

Summary:
TransactionParticipant::Impl::SubmitUpdateTransaction moves the passed in
std::unique_ptr<tablet::UpdateTxnOperation> to TabletPeer::SubmitUpdateTransaction, but then
calls ToString on it in event of failure, but at this point the operation has been destroyed.

This diff changes this codepath to not destroy the operation object on failure.

This is the cause of YbAdminSnapshotScheduleTest.ConsistentTxnRestore failures.
Jira: DB-13303

Original commit: b57d257 / D38953

Test Plan: Jenkins

Reviewers: sergei

Reviewed By: sergei

Subscribers: ybase

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D39832

src/yb/tablet/tablet_peer.cc

@@ -720,7 +720,7 @@ void TabletPeer::Submit(std::unique_ptr<Operation> operation, int64_t term) {
 }
 
 Status TabletPeer::SubmitUpdateTransaction(
-    std::unique_ptr<UpdateTxnOperation> operation, int64_t term) {
+    std::unique_ptr<UpdateTxnOperation>& operation, int64_t term) {
   if (!operation->tablet_is_set()) {
     auto tablet = VERIFY_RESULT(shared_tablet_safe());
     operation->SetTablet(tablet);

src/yb/tablet/tablet_peer.h

@@ -220,8 +220,9 @@ class TabletPeer : public std::enable_shared_from_this<TabletPeer>,
   std::unique_ptr<UpdateTxnOperation> CreateUpdateTransaction(
       std::shared_ptr<LWTransactionStatePB> request) override;
 
+  // `operation` is moved from in the event of success, and left alive in event of failure.
   Status SubmitUpdateTransaction(
-      std::unique_ptr<UpdateTxnOperation> operation, int64_t term) override;
+      std::unique_ptr<UpdateTxnOperation>& operation, int64_t term) override;
 
   HybridTime SafeTimeForTransactionParticipant() override;
   Result<HybridTime> WaitForSafeTime(HybridTime safe_time, CoarseTimePoint deadline) override;

src/yb/tablet/transaction_coordinator.cc

@@ -1730,7 +1730,7 @@ class TransactionCoordinator::Impl : public TransactionStateContext,
 
     for (auto& update : actions->updates) {
       auto submit_status =
-          context_.SubmitUpdateTransaction(std::move(update), actions->leader_term);
+          context_.SubmitUpdateTransaction(update, actions->leader_term);
       if (!submit_status.ok()) {
         LOG_WITH_PREFIX(DFATAL)
             << "Could not submit transaction status update operation: "

src/yb/tablet/transaction_coordinator.h

@@ -69,7 +69,7 @@ class TransactionCoordinatorContext {
   virtual std::unique_ptr<UpdateTxnOperation> CreateUpdateTransaction(
       std::shared_ptr<LWTransactionStatePB> request) = 0;
   virtual Status SubmitUpdateTransaction(
-      std::unique_ptr<UpdateTxnOperation> operation, int64_t term) = 0;
+      std::unique_ptr<UpdateTxnOperation>& operation, int64_t term) = 0;
 
   server::Clock& clock() const {
     return *clock_ptr();

src/yb/tablet/transaction_participant.cc

@@ -1882,7 +1882,7 @@ class TransactionParticipant::Impl
   void SubmitUpdateTransaction(
       std::unique_ptr<tablet::UpdateTxnOperation> operation, int64_t term) {
     WARN_NOT_OK(
-        participant_context_.SubmitUpdateTransaction(std::move(operation), term),
+        participant_context_.SubmitUpdateTransaction(operation, term),
         Format("Could not submit transaction status update operation: $0", operation->ToString()));
   }
 

src/yb/tablet/transaction_participant_context.h

@@ -43,7 +43,7 @@ class TransactionParticipantContext {
   virtual void UpdateClock(HybridTime hybrid_time) = 0;
   virtual bool IsLeader() = 0;
   virtual Status SubmitUpdateTransaction(
-      std::unique_ptr<UpdateTxnOperation> state, int64_t term) = 0;
+      std::unique_ptr<UpdateTxnOperation>& state, int64_t term) = 0;
 
   // Returns hybrid time that lower than any future transaction apply record.
   virtual HybridTime SafeTimeForTransactionParticipant() = 0;