[yugabyte#26880] DocDB: Prevent deadlock between txn load and compaction that causes peer to enter a stuck state

Summary:
**Problem**
Background compactions using a filter (for instance, tables where cdc is enabled) could deadlock with the transaction loader causing the peer to enter into a stuck state. This is because compaction can wait for the loader to finish with `RunningTransactionContext::mutex_` held, and the loader could wait on the same mutex causing a deadlock.

**Issue seen**
In one of the internal clusters, we saw 2 peers stuck with the following traces unable to process any consensus operations.

The peer gets stuck trying to process a replicated op and holds up `ReplicaState::update_lock_`, and waits for the loader thread to complete.
```
    @     0x7fcfb39f27a9  __pthread_cond_timedwait
    @     0x555e3409bcd4  yb::tablet::TransactionLoader::WaitLoaded()
    @     0x555e33ff760e  yb::tablet::Tablet::ApplyKeyValueRowOperations()
    @     0x555e33ff6dce  yb::tablet::Tablet::ApplyOperation()
    @     0x555e33ff69f2  yb::tablet::Tablet::ApplyRowOperations()
    @     0x555e33fafeed  yb::tablet::WriteOperation::DoReplicated()
    @     0x555e33fa129d  yb::tablet::Operation::Replicated()
    @     0x555e33fa3690  yb::tablet::OperationDriver::ReplicationFinished()
    @     0x555e32f29082  yb::consensus::ConsensusRound::NotifyReplicationFinished()
    @     0x555e32f81091  yb::consensus::ReplicaState::ApplyPendingOperationsUnlocked()
    @     0x555e32f8039a  yb::consensus::ReplicaState::AdvanceCommittedOpIdUnlocked()
    @     0x555e32f65d49  yb::consensus::RaftConsensus::UpdateReplica()
    @     0x555e32f41e4f  yb::consensus::RaftConsensus::Update()
    @     0x555e3437f846  yb::tserver::ConsensusServiceImpl::UpdateConsensus()
    @     0x555e32fce4f8  std::__1::__function::__func<>::operator()()
    @     0x555e32fcf0fe  yb::consensus::ConsensusServiceIf::Handle()

Total number of threads: 2
```

The loader thread waits on `RunningTransactionContext::mutex_`
```
    @     0x7fcfb39f582a  __lll_lock_wait
    @     0x7fcfb39eead8  __GI___pthread_mutex_lock
    @     0x555e340a9003  yb::tablet::TransactionParticipant::Impl::LoadTransaction()
    @     0x555e340984ce  yb::tablet::TransactionLoader::Executor::Execute()
    @     0x555e347d9cd8  yb::Thread::SuperviseThread()
    @     0x7fcfb39ec1c9  start_thread
    @     0x7fcfb3c3de72  __GI___clone

Total number of threads: 2
```

and `RunningTransactionContext::mutex_` is held by the background compaction thread that is waiting for the loader to complete, causing a deadlock.
```
    @     0x7fcfb39f27a9  __pthread_cond_timedwait
    @     0x555e3409bcd4  yb::tablet::TransactionLoader::WaitLoaded()
    @     0x555e340ae872  yb::tablet::TransactionParticipant::Cleanup()
    @     0x555e330db3bf  yb::docdb::(anonymous namespace)::DocDBIntentsCompactionFilter::CompactionFinished()
    @     0x555e33c79d65  rocksdb::CompactionJob::ProcessKeyValueCompaction()
    @     0x555e33c76f6a  rocksdb::CompactionJob::Run()
    @     0x555e33cadea9  rocksdb::DBImpl::BackgroundCompaction()
    @     0x555e33cabd95  rocksdb::DBImpl::BackgroundCallCompaction()
    @     0x555e347a847b  yb::(anonymous namespace)::PriorityThreadPoolWorker::Run()
    @     0x555e347d9cd8  yb::Thread::SuperviseThread()
    @     0x7fcfb39ec1c9  start_thread
    @     0x7fcfb3c3de72  __GI___clone

Total number of threads: 2
```

**Fix**
This revision addresses the issue by executing `WaitLoaded` outside scope of `RunningTransactionContext::mutex_`
Jira: DB-16294

Test Plan:
Jenkins
./yb_build.sh --cxx-test cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestCompactionDoesntDeadlockWithTxnLoader

The test fails without the changes.

Reviewers: sergei, esheng

Reviewed By: sergei

Subscribers: ybase

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

Author: basavaraj29

src/yb/integration-tests/cdcsdk_ysql-test.cc

@@ -2724,6 +2724,62 @@ TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestIntentCountPersistencyAfterCo
   ASSERT_EQ(final_record_size, 0);
 }
 
+TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestCompactionDoesntDeadlockWithTxnLoader)) {
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_retryable_request_timeout_secs) = 0;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_update_min_cdc_indices_interval_secs) = 1;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_cdc_state_checkpoint_update_interval_ms) = 0;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_aborted_intent_cleanup_ms) = 1000;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_save_index_into_wal_segments) = true;
+
+  ASSERT_OK(SetUpWithParams(1, 1, false));
+  const uint32_t num_tablets = 1;
+  auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName, num_tablets));
+
+  google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
+  ASSERT_OK(test_client()->GetTablets(table, 0, &tablets, /* partition_list_version =*/nullptr));
+  ASSERT_EQ(tablets.size(), num_tablets);
+
+  TableId table_id = ASSERT_RESULT(GetTableId(&test_cluster_, kNamespaceName, kTableName));
+  xrepl::StreamId stream_id = ASSERT_RESULT(CreateDBStream(IMPLICIT));
+
+  auto resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
+  ASSERT_FALSE(resp.has_error());
+
+  // Insert some records in transaction.
+  ASSERT_OK(WriteRowsHelper(0 /* start */, 10 /* end */, &test_cluster_, true));
+  ASSERT_OK(WriteRowsHelper(10 /* start */, 20 /* end */, &test_cluster_, true));
+  ASSERT_OK(FlushTable(table.table_id()));
+
+  GetChangesResponsePB change_resp_1;
+  ASSERT_OK(WaitForGetChangesToFetchRecords(&change_resp_1, stream_id, tablets, 20));
+  ASSERT_OK(WaitForPostApplyMetadataWritten(2 /* expected_num_transactions */));
+
+  ASSERT_OK(FlushTable(table.table_id()));
+  auto* tablet_server = test_cluster()->mini_tablet_server(0);
+
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_TEST_skip_process_apply) = true;
+
+  yb::SyncPoint::GetInstance()->LoadDependency({
+    {"TransactionParticipant::Impl::Cleanup", "TransactionLoader::Executor::Start"}});
+  yb::SyncPoint::GetInstance()->ClearTrace();
+  yb::SyncPoint::GetInstance()->EnableProcessing();
+  ASSERT_OK(tablet_server->Restart());
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(FLAGS_aborted_intent_cleanup_ms));
+  auto status_future = std::async(std::launch::async, [&]() -> Status {
+    return tablet_server->CompactTablets(docdb::SkipFlush::kFalse);
+  });
+
+  // Compaction shouldn't deadlock with the txn loader thread.
+  CHECK_OK(
+      WaitFor([&]() {
+        return status_future.wait_for(0s) == std::future_status::ready;
+      },
+      15s * kTimeMultiplier,
+      "Compaction taking longer, probably deadlocked with txn load."));
+  ASSERT_OK(status_future.get());
+}
+
 // https://github.com/yugabyte/yugabyte-db/issues/19385
 TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestLogGCForNewTablesAddedAfterCreateStream)) {
   ANNOTATE_UNPROTECTED_WRITE(FLAGS_update_min_cdc_indices_interval_secs) = 100000;

src/yb/integration-tests/cdcsdk_ysql_test_base.h

@@ -135,6 +135,9 @@ DECLARE_bool(cdc_use_byte_threshold_for_vwal_changes);
 DECLARE_bool(ysql_enable_pg_export_snapshot);
 DECLARE_bool(ysql_yb_enable_consistent_replication_from_hash_range);
 DECLARE_uint64(cdcsdk_update_restart_time_interval_secs);
+DECLARE_int32(retryable_request_timeout_secs);
+DECLARE_bool(save_index_into_wal_segments);
+DECLARE_bool(TEST_skip_process_apply);
 
 namespace yb {
 

src/yb/tablet/transaction_loader.cc

@@ -33,6 +33,7 @@
 #include "yb/util/operation_counter.h"
 #include "yb/util/pb_util.h"
 #include "yb/util/scope_exit.h"
+#include "yb/util/sync_point.h"
 #include "yb/util/thread.h"
 
 using namespace std::literals;
@@ -108,6 +109,7 @@ class TransactionLoader::Executor {
     });
 
     LOG_WITH_PREFIX(INFO) << "Load transactions start";
+    DEBUG_ONLY_TEST_SYNC_POINT("TransactionLoader::Executor::Start");
 
     status = LoadPendingApplies();
     if (!status.ok()) {
@@ -432,6 +434,17 @@ Status TransactionLoader::WaitLoaded(const TransactionId& id) NO_THREAD_SAFETY_A
   return load_status_;
 }
 
+Status TransactionLoader::WaitLoaded(const TransactionIdApplyOpIdMap& txns) {
+  if (txns.empty() || RSTATUS_DCHECK_RESULT(Completed())) {
+    return Status::OK();
+  }
+  const auto& max_txn = std::max_element(
+      txns.begin(), txns.end(), [](const auto& lhs, const auto& rhs) {
+        return lhs.first < rhs.first;
+      });
+  return WaitLoaded(max_txn->first);
+}
+
 // Disable thread safety analysis because std::unique_lock is used.
 Status TransactionLoader::WaitAllLoaded() NO_THREAD_SAFETY_ANALYSIS {
   // WaitAllLoaded is only invoked when opening a tablet.

src/yb/tablet/transaction_loader.h

@@ -87,6 +87,7 @@ class TransactionLoader {
   }
 
   Status WaitLoaded(const TransactionId& id);
+  Status WaitLoaded(const TransactionIdApplyOpIdMap& txns);
   Status WaitAllLoaded();
 
   std::optional<docdb::ApplyStateWithCommitInfo> GetPendingApply(const TransactionId& id) const

src/yb/tablet/transaction_participant.cc

@@ -804,15 +804,17 @@ class TransactionParticipant::Impl
   }
 
   Status Cleanup(TransactionIdApplyOpIdMap&& txns, TransactionStatusManager* status_manager) {
+    DEBUG_ONLY_TEST_SYNC_POINT("TransactionParticipant::Impl::Cleanup");
+    // Execute WaitLoaded outside of this->mutex_, else there's possibility of a deadlock since
+    // the loader needs this->mutex_ in TransactionLoaderContext::LoadTransaction to finish load.
+    RETURN_NOT_OK(loader_.WaitLoaded(txns));
     TransactionIdSet set;
     {
       std::lock_guard lock(mutex_);
       const OpId& cdcsdk_checkpoint_op_id = GetLatestCheckPointUnlocked();
 
       if (cdcsdk_checkpoint_op_id != OpId::Max()) {
         for (const auto& [transaction_id, apply_op_id] : txns) {
-          RETURN_NOT_OK(loader_.WaitLoaded(transaction_id));
-
           const OpId* apply_record_op_id = &apply_op_id;
           if (!apply_op_id.valid()) {
             // Apply op id is unknown -- may be from before upgrade to version that writes