[BACKPORT 2025.1][yugabyte#27559] DocDB: Table locks: Disable reuse of docdb txn incase of abort when object locking feature is enabled

Summary:
Original commit: 7d02f94 / D44857
Timed out requests at the Object lock Manager are resumed organically by the poller thread once the deadline is exceeded. So this could mean that a ysql statement could time out, execute a finish transaction rpc and could initiate a new ysql transaction before the poller resumes the obsolete timed out lock waiter.

Given that we try re-using docdb transactions wherever possible when object locking is enabled, the above in combination with transaction re-use could lead to undesired issues. Here's how the OLM works in brief:
1. While serving new incoming lock requests, shared `TrackedTransactionLockEntry` is created if one doesn't exists, keyed against txn id, stored in `txn_locks_`
2. `PrepareAcquire` is executed which validates whether the lock request should be tried.
3. moves on to acquire the lock if available, else enters the wait queue.
4. When waiters are resumed, goto step 2.
5. When serving a release all request, remove entry from `txn_locks_`, release acquired locks, and let obsolete waiting locks be resumed by the poller thread.

Here's the brief working of the object lock tracker code.
1. For incoming lock requests, instrument the lock in the corresponding map key against `<txn, subtxn>` with state set to `WAITING` and invoke the OLM.
2. When OLM executes the lock callback, tap into it, try finding the map with key `<txn, subtxn>`, and if exists, change the state of the lock entry in the map accordingly.

Consider the following scenario:
1. ysql starts read only `ysql_txnA`, issues a lock request. this is associated with `docdb_txnA`, and the lock request is forwarded to the OLM.
2. lock request enters the wait-queue
3. YSQL detects timeout, cancels the request, issues a finish txn call. `docdb_txnA` doesn't get consumed since it didn't write any intents nor the txn itself failed (failed heartbeats). OLM erases `TrackedTransactionLockEntry` keyed against `docdb_txnA`.
4. YSQL start a new `ysql_txnB`, issues a lock request. docdb re-uses `docdb_txnA` issues the lock request, OLM creates a new entry for the same transaction id, and moves on to acquire the lock.
5. The poller might now realize that the earlier waiting request timed out, and try resuming it.

This results in issues with observability, since at some point the OLM has state corresponding to different ysql transactions stored under the same docdb transaction id. As a consequence, it results in a segv with the lock tracking code in the above scenario as follows:
- Step 1 creates a map for key `<docdb_txnA, 1>`, and inserts `key(lock)`.
- Step 3 erases the map for key `<docdb_txnA, 1>`.
- Step 4 creates a new map for key `<docdb_txnA, 1>`, and inserts `key(lock_new)`.
- Step 5 tries to new access entry `key(lock)` in the new map which doesn't exists, resulting in a segv
```
* thread #1, name = 'yb-tserver', stop reason = signal SIGSEGV
  * frame #0: 0x0000aaaaead1b198 yb-tserver`std::__1::__function::__func<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status), std::__1::allocator<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)>, void (yb::Status const&)>::operator()(yb::Status const&) [inlined] yb::tserver::ObjectLockTracker::UntrackLock(this=<unavailable>, lock_context=0x000010f1bbe20800) at ts_local_lock_manager.cc:134:25
    frame #1: 0x0000aaaaead1b0c8 yb-tserver`std::__1::__function::__func<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status), std::__1::allocator<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)>, void (yb::Status const&)>::operator()(yb::Status const&) at ts_local_lock_manager.cc:104:7
    frame #2: 0x0000aaaaead1b0b4 yb-tserver`std::__1::__function::__func<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status), std::__1::allocator<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)>, void (yb::Status const&)>::operator()(yb::Status const&) [inlined] yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(this=0x000010f1b5ad4cd0, status=Status @ 0x0000ffef78896640)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)::operator()(yb::Status) const at ts_local_lock_manager.cc:324:46
    frame #3: 0x0000aaaaead1b034 yb-tserver`std::__1::__function::__func<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status), std::__1::allocator<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)>, void (yb::Status const&)>::operator()(yb::Status const&) [inlined] decltype(std::declval<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)&>()(std::declval<yb::Status const&>())) std::__1::__invoke[abi:ne190100]<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)&, yb::Status const&>(__f=<unavailable>, __args=<unavailable>) at invoke.h:149:25
    frame #4: 0x0000aaaaead1b01c yb-tserver`std::__1::__function::__func<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status), std::__1::allocator<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)>, void (yb::Status const&)>::operator()(yb::Status const&) [inlined] void std::__1::__invoke_void_return_wrapper<void, true>::__call[abi:ne190100]<yb::tserver::TSLocalLockManager::Impl::PrepareAndExecuteAcquire(yb::tserver::AcquireObjectLockRequestPB const&, std::__1::chrono::time_point<yb::CoarseMonoClock, std::__1::chrono::duration<long long, std::__1::ratio<1l, 1000000000l>>>, std::__1::function<void (yb::Status const&)>&, yb::StronglyTypedBool<yb::tserver::WaitForBootstrap_Tag>)::'lambda'(yb::Status)&, yb::Status const&>(__args=<unavailable>, __args=<unavailable>) at invoke.h:224:5
```

This revision address the issue by burning/consuming the docdb transaction in case of YSQL transaction abort. This would force the `ysql_txnB` above to be associated with a new `docdb_txnB` and wouldn't lead to any observability issues/mixed state.

Note that there isn't any problem with reusing docdb transactions across ysql read-only transactions that move on to commit successfully. This is because if the ysql transaction moves on to commit, it implies that all the object locks were granted => no waiting object locks, and hence all corresponding entries of the respective docdb transaction would be released at the OLM in-line with the commit.
Jira: DB-17123

Test Plan:
Jenkins

./yb_build.sh --cxx-test pg_object_locks-test --gtest_filter PgObjectLocksTestRF1.TestDisableReuseAbortedPlainTxn
./yb_build.sh release --java-test 'org.yb.pgsql.TestPgRegressPgAuth#schedule'

Reviewers: #db-approvers, amitanand, yyan, rthallam

Reviewed By: amitanand

Subscribers: yql, ybase, svc_phabricator

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

Author: basavaraj29

src/yb/tserver/pg_client_session.cc

@@ -1062,6 +1062,10 @@ class TransactionProvider {
     return txn_meta_for_release;
   }
 
+  bool HasNextTxnForPlain() const {
+    return next_plain_ != nullptr;
+  }
+
  private:
   struct BuildStrategy {
     bool is_ddl = false;
@@ -1796,17 +1800,27 @@ class PgClientSession::Impl {
       rpc::RpcContext* context) {
     saved_priority_.reset();
     const bool is_ddl = req.has_ddl_mode();
+    const bool is_commit = req.commit();
     const bool ddl_use_regular_transaction_block =
         is_ddl && req.ddl_mode().use_regular_transaction_block();
     const auto kind = GetSessionKindBasedOnDDLOptions(is_ddl, ddl_use_regular_transaction_block);
     const auto deadline = context->GetClientDeadline();
     auto& txn = GetSessionData(kind).transaction;
     if (!txn) {
       VLOG_WITH_PREFIX_AND_FUNC(2)
-          << "ddl: " << is_ddl
+          << "ddl: " << is_ddl << ", " << (is_commit ? "commit" : "abort")
           << ", ddl_use_regular_transaction_block: " << ddl_use_regular_transaction_block
           << ", no running distributed transaction";
-      return ReleaseObjectLocksIfNecessary(txn, kind, deadline);
+      if (is_commit || is_ddl || !IsObjectLockingEnabled()) {
+        return ReleaseObjectLocksIfNecessary(txn, kind, deadline);
+      }
+      // When object locking is enabled, prevent re-use of plain docdb txn is case of abort.
+      if (!transaction_provider_.HasNextTxnForPlain()) {
+        return Status::OK();
+      }
+      txn = transaction_provider_.Take<PgClientSessionKind::kPlain>(
+          client::ForceGlobalTransaction::kFalse, deadline).first;
+      VLOG_WITH_PREFIX_AND_FUNC(1) << "Consuming re-usable kPlain txn " << txn->id();
     }
 
     client::YBTransactionPtr txn_value;
@@ -3283,7 +3297,7 @@ class PgClientSession::Impl {
       return Status::OK();
     }
     if (!txn && kind != PgClientSessionKind::kPlain) {
-      // Release of object locks would be handled by master's ddl verification task.
+      // kDdl might not have a txn when this function is invoked on Shutdown.
       return Status::OK();
     }
 

src/yb/yql/pgwrapper/pg_object_locks-test.cc

@@ -836,4 +836,30 @@ TEST_F_EX(
   }
 }
 
+TEST_F_EX(PgObjectLocksTestRF1, TestDisableReuseAbortedPlainTxn, TestWithTransactionalDDL) {
+  constexpr auto kStatementTimeoutMs = 1000 * kTimeMultiplier;
+  // Restart the cluster_ with the poll interval set to 5x of statment timeout
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_olm_poll_interval_ms) = 5 * kStatementTimeoutMs;
+  google::SetVLOGLevel("pg_client_session*", 1);
+  ASSERT_OK(cluster_->RestartSync());
+  Init();
+
+  auto conn1 = ASSERT_RESULT(Connect());
+  ASSERT_OK(conn1.Execute("CREATE TABLE test(k INT PRIMARY KEY, v INT)"));
+  ASSERT_OK(conn1.Execute("CREATE TABLE test1(k INT)"));
+
+  ASSERT_OK(conn1.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn1.Execute("LOCK TABLE test in ACCESS EXCLUSIVE mode"));
+
+  auto conn2 = ASSERT_RESULT(Connect());
+  // TODO(#26792): Change to LOCK_TIMEOUT once GH is addressed.
+  ASSERT_OK(conn2.ExecuteFormat("SET statement_timeout=$0", kStatementTimeoutMs));
+  ASSERT_OK(conn2.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  auto log_waiter = StringWaiterLogSink("Consuming re-usable kPlain txn");
+  // Would timeout before the waiting lock at the OLM is released.
+  ASSERT_NOK(conn2.Execute("LOCK TABLE test in ACCESS SHARE mode"));
+  ASSERT_OK(conn2.RollbackTransaction());
+  ASSERT_OK(log_waiter.WaitFor(MonoDelta::FromMilliseconds(5 * kTimeMultiplier)));
+}
+
 }  // namespace yb::pgwrapper