SERVER-41317 Push commit transaction's check for a majority-committed prepare down into the TransactionParticipant

4 files changed, 44 insertions, 19 deletions

diff --git a/src/mongo/db/commands/txn_cmds.cpp b/src/mongo/db/commands/txn_cmds.cpp
index 997f5c83104..d18a85a4064 100644
--- a/src/mongo/db/commands/txn_cmds.cpp
+++ b/src/mongo/db/commands/txn_cmds.cpp
@@ -52,7 +52,6 @@ MONGO_FAIL_POINT_DEFINE(participantReturnNetworkErrorForAbortAfterExecutingAbort
 MONGO_FAIL_POINT_DEFINE(participantReturnNetworkErrorForCommitAfterExecutingCommitLogic);
 MONGO_FAIL_POINT_DEFINE(hangBeforeCommitingTxn);
 MONGO_FAIL_POINT_DEFINE(hangBeforeAbortingTxn);
-MONGO_FAIL_POINT_DEFINE(skipCommitTxnCheckPrepareMajorityCommitted);
 // TODO SERVER-39704: Remove this fail point once the router can safely retry within a transaction
 // on stale version and snapshot errors.
 MONGO_FAIL_POINT_DEFINE(dontRemoveTxnCoordinatorOnAbort);
@@ -122,13 +121,6 @@ public:
 
         auto optionalCommitTimestamp = cmd.getCommitTimestamp();
         if (optionalCommitTimestamp) {
-            const auto replCoord = repl::ReplicationCoordinator::get(opCtx);
-            uassert(ErrorCodes::InvalidOptions,
-                    "commitTransaction for a prepared transaction cannot be run before its prepare "
-                    "oplog entry has been majority committed",
-                    replCoord->getLastCommittedOpTime().getTimestamp() >=
-                            txnParticipant.getPrepareOpTime().getTimestamp() ||
-                        MONGO_FAIL_POINT(skipCommitTxnCheckPrepareMajorityCommitted));
             // commitPreparedTransaction will throw if the transaction is not prepared.
             txnParticipant.commitPreparedTransaction(opCtx, optionalCommitTimestamp.get(), {});
         } else {
diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp
index 3b963c9d7b8..0ccc127e94d 100644
--- a/src/mongo/db/transaction_participant.cpp
+++ b/src/mongo/db/transaction_participant.cpp
@@ -81,6 +81,8 @@ MONGO_FAIL_POINT_DEFINE(hangAfterSettingPrepareStartTime);
 
 MONGO_FAIL_POINT_DEFINE(hangBeforeReleasingTransactionOplogHole);
 
+MONGO_FAIL_POINT_DEFINE(skipCommitTxnCheckPrepareMajorityCommitted);
+
 const auto getTransactionParticipant = Session::declareDecoration<TransactionParticipant>();
 
 // The command names that are allowed in a prepared transaction.
@@ -1260,8 +1262,10 @@ void TransactionParticipant::Participant::commitPreparedTransaction(
     // transaction was prepared, we dropped the RSTL. We do not need to reacquire the PBWM because
     // if we're not the primary we will uassert anyways.
     repl::ReplicationStateTransitionLockGuard rstl(opCtx, MODE_IX);
+
+    const auto replCoord = repl::ReplicationCoordinator::get(opCtx);
+
     if (opCtx->writesAreReplicated()) {
-        auto replCoord = repl::ReplicationCoordinator::get(opCtx);
         uassert(ErrorCodes::NotMaster,
                 "Not primary so we cannot commit a prepared transaction",
                 replCoord->canAcceptWritesForDatabase(opCtx, "admin"));
@@ -1272,9 +1276,20 @@ void TransactionParticipant::Participant::commitPreparedTransaction(
             o().txnState.isPrepared());
     uassert(
         ErrorCodes::InvalidOptions, "'commitTimestamp' cannot be null", !commitTimestamp.isNull());
+
+    const auto prepareTimestamp = o().prepareOpTime.getTimestamp();
+
     uassert(ErrorCodes::InvalidOptions,
             "'commitTimestamp' must be greater than or equal to 'prepareTimestamp'",
-            commitTimestamp >= o().prepareOpTime.getTimestamp());
+            commitTimestamp >= prepareTimestamp);
+
+    if (!commitOplogEntryOpTime) {
+        uassert(ErrorCodes::InvalidOptions,
+                "commitTransaction for a prepared transaction cannot be run before its prepare "
+                "oplog entry has been majority committed",
+                replCoord->getLastCommittedOpTime().getTimestamp() >= prepareTimestamp ||
+                    MONGO_FAIL_POINT(skipCommitTxnCheckPrepareMajorityCommitted));
+    }
 
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
diff --git a/src/mongo/db/transaction_participant_test.cpp b/src/mongo/db/transaction_participant_test.cpp
index 8dbdd4aa6f5..f699bceb432 100644
--- a/src/mongo/db/transaction_participant_test.cpp
+++ b/src/mongo/db/transaction_participant_test.cpp
@@ -57,6 +57,7 @@
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/clock_source_mock.h"
+#include "mongo/util/fail_point_service.h"
 #include "mongo/util/net/socket_utils.h"
 #include "mongo/util/tick_source_mock.h"
 
@@ -258,6 +259,12 @@ protected:
 
         opCtx()->setLogicalSessionId(_sessionId);
         opCtx()->setTxnNumber(_txnNumber);
+
+        // Normally, committing a transaction is supposed to usassert if the corresponding prepare
+        // has not been majority committed. We excempt our unit tests from this expectation.
+        setGlobalFailPoint("skipCommitTxnCheckPrepareMajorityCommitted",
+                           BSON("mode"
+                                << "alwaysOn"));
     }
 
     void tearDown() override {
@@ -267,6 +274,10 @@ protected:
         SessionCatalog::get(opCtx()->getServiceContext())->reset_forTest();
 
         MockReplCoordServerFixture::tearDown();
+
+        setGlobalFailPoint("skipCommitTxnCheckPrepareMajorityCommitted",
+                           BSON("mode"
+                                << "off"));
     }
 
     SessionCatalog* catalog() {
@@ -1901,7 +1912,7 @@ TEST_F(TransactionsMetricsTest, SingleTransactionStatsDurationShouldBeSetUponCom
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "commitTransaction");
     // The transaction machinery cannot store an empty locker.
-    Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow);
+    { Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow); }
 
     // Advance the clock.
     tickSource->advance(Microseconds(100));
@@ -1919,7 +1930,7 @@ TEST_F(TransactionsMetricsTest, SingleTransactionStatsPreparedDurationShouldBeSe
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "commitTransaction");
     // The transaction machinery cannot store an empty locker.
-    Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow);
+    { Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow); }
 
     // Advance the clock.
     tickSource->advance(Microseconds(10));
@@ -1978,7 +1989,7 @@ TEST_F(TransactionsMetricsTest, SingleTransactionStatsDurationShouldKeepIncreasi
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "commitTransaction");
     // The transaction machinery cannot store an empty locker.
-    Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow);
+    { Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow); }
 
     tickSource->advance(Microseconds(100));
 
@@ -2010,7 +2021,7 @@ TEST_F(TransactionsMetricsTest,
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "commitTransaction");
     // The transaction machinery cannot store an empty locker.
-    Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow);
+    { Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow); }
 
     // Prepare the transaction and extend the duration in the prepared state.
     const auto prepareTimestamp = txnParticipant.prepareTransaction(opCtx(), {});
@@ -2044,7 +2055,7 @@ TEST_F(TransactionsMetricsTest, SingleTransactionStatsDurationShouldKeepIncreasi
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "insert");
     // The transaction machinery cannot store an empty locker.
-    Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow);
+    { Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow); }
 
     tickSource->advance(Microseconds(100));
 
@@ -2076,7 +2087,7 @@ TEST_F(TransactionsMetricsTest,
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "abortTransaction");
     // The transaction machinery cannot store an empty locker.
-    Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow);
+    { Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow); }
 
     // Prepare the transaction and extend the duration in the prepared state.
     txnParticipant.prepareTransaction(opCtx(), {});
@@ -2759,7 +2770,7 @@ TEST_F(TransactionsMetricsTest, LastClientInfoShouldUpdateUponCommit) {
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "insert");
     // The transaction machinery cannot store an empty locker.
-    Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow);
+    { Lock::GlobalLock lk(opCtx(), MODE_IX, Date_t::now(), Lock::InterruptBehavior::kThrow); }
     txnParticipant.commitUnpreparedTransaction(opCtx());
 
     // LastClientInfo should have been set.
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index 03f0dd3249b..d7e10b856e6 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -3052,7 +3052,10 @@ public:
 
         txnParticipant.unstashTransactionResources(_opCtx, "commitTransaction");
 
-        txnParticipant.commitPreparedTransaction(_opCtx, commitEntryTs, {});
+        {
+            FailPointEnableBlock failPointBlock("skipCommitTxnCheckPrepareMajorityCommitted");
+            txnParticipant.commitPreparedTransaction(_opCtx, commitEntryTs, {});
+        }
 
         txnParticipant.stashTransactionResources(_opCtx);
         {
@@ -3325,7 +3328,11 @@ public:
         }
         txnParticipant.unstashTransactionResources(_opCtx, "commitTransaction");
 
-        txnParticipant.commitPreparedTransaction(_opCtx, commitEntryTs, {});
+        {
+            FailPointEnableBlock failPointBlock("skipCommitTxnCheckPrepareMajorityCommitted");
+            txnParticipant.commitPreparedTransaction(_opCtx, commitEntryTs, {});
+        }
+
         assertNoStartOpTime();
 
         txnParticipant.stashTransactionResources(_opCtx);