SERVER-35814 ensure that prepareTransaction waits for writeConcern

4 files changed, 41 insertions, 0 deletions

diff --git a/src/mongo/db/s/txn_two_phase_commit_cmds.cpp b/src/mongo/db/s/txn_two_phase_commit_cmds.cpp
index 99ed8e1ae5a..72c45624050 100644
--- a/src/mongo/db/s/txn_two_phase_commit_cmds.cpp
+++ b/src/mongo/db/s/txn_two_phase_commit_cmds.cpp
@@ -82,9 +82,30 @@ public:
                 "Transaction isn't in progress",
                 txnParticipant->inMultiDocumentTransaction());
 
+        if (txnParticipant->transactionIsPrepared()) {
+            auto& replClient = repl::ReplClientInfo::forClient(opCtx->getClient());
+            auto prepareOpTime = txnParticipant->getPrepareOpTime();
+            // Set the client optime to be prepareOpTime if it's not already later than
+            // prepareOpTime.
+            // This ensures that we wait for writeConcern and that prepareOpTime will be committed.
+            if (prepareOpTime > replClient.getLastOp()) {
+                replClient.setLastOp(prepareOpTime);
+            }
+
+            invariant(opCtx->recoveryUnit()->getPrepareTimestamp() == prepareOpTime.getTimestamp(),
+                      str::stream() << "recovery unit prepareTimestamp: "
+                                    << opCtx->recoveryUnit()->getPrepareTimestamp().toString()
+                                    << " participant prepareOpTime: "
+                                    << prepareOpTime.toString());
+
+            result.append("prepareTimestamp", prepareOpTime.getTimestamp());
+            return true;
+        }
+
         // Add prepareTimestamp to the command response.
         auto timestamp = txnParticipant->prepareTransaction(opCtx);
         result.append("prepareTimestamp", timestamp);
+
         return true;
     }
 } prepareTransactionCmd;
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index fd185ca00fe..517d9c934d6 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -501,6 +501,7 @@ bool runCommandImpl(OperationContext* opCtx,
                     !txnParticipant->inMultiDocumentTransaction() ||
                     invocation->definition()->getName() == "commitTransaction" ||
                     invocation->definition()->getName() == "abortTransaction" ||
+                    invocation->definition()->getName() == "prepareTransaction" ||
                     invocation->definition()->getName() == "doTxn");
 
         auto lastOpBeforeRun = repl::ReplClientInfo::forClient(opCtx->getClient()).getLastOp();
diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp
index d4567780bcd..ba01744fff4 100644
--- a/src/mongo/db/transaction_participant.cpp
+++ b/src/mongo/db/transaction_participant.cpp
@@ -542,6 +542,10 @@ Timestamp TransactionParticipant::prepareTransaction(OperationContext* opCtx) {
     OplogSlotReserver oplogSlotReserver(opCtx);
     const auto prepareOplogSlot = oplogSlotReserver.getReservedOplogSlot();
     const auto prepareTimestamp = prepareOplogSlot.opTime.getTimestamp();
+    invariant(_prepareOpTime.isNull(),
+              str::stream() << "This transaction has already reserved a prepareOpTime at: "
+                            << _prepareOpTime.toString());
+    _prepareOpTime = prepareOplogSlot.opTime;
 
     if (MONGO_FAIL_POINT(hangAfterReservingPrepareTimestamp)) {
         // This log output is used in js tests so please leave it.
@@ -862,6 +866,7 @@ void TransactionParticipant::_abortTransactionOnSession(WithLock wl) {
     _transactionOperationBytes = 0;
     _transactionOperations.clear();
     _txnState.transitionTo(wl, TransactionState::kAborted);
+    _prepareOpTime = repl::OpTime();
     _speculativeTransactionReadOpTime = repl::OpTime();
 
     _getSession()->unlockTxnNumber();
@@ -1292,6 +1297,7 @@ void TransactionParticipant::_setNewTxnNumber(WithLock wl, const TxnNumber& txnN
         stdx::lock_guard<stdx::mutex> lm(_metricsMutex);
         _transactionMetricsObserver.resetSingleTransactionStats(txnNumber);
     }
+    _prepareOpTime = repl::OpTime();
     _speculativeTransactionReadOpTime = repl::OpTime();
     _multikeyPathInfo.clear();
     _autoCommit = boost::none;
diff --git a/src/mongo/db/transaction_participant.h b/src/mongo/db/transaction_participant.h
index 476cef53a61..e6bcbf5e237 100644
--- a/src/mongo/db/transaction_participant.h
+++ b/src/mongo/db/transaction_participant.h
@@ -255,6 +255,11 @@ public:
         return _txnState.isAborted(lk);
     }
 
+    bool transactionIsPrepared() const {
+        stdx::lock_guard<stdx::mutex> lk(_mutex);
+        return _txnState.isPrepared(lk);
+    }
+
     /**
      * Returns true if we are in an active multi-document transaction or if the transaction has
      * been aborted. This is used to cover the case where a transaction has been aborted, but the
@@ -292,6 +297,11 @@ public:
         return _speculativeTransactionReadOpTime;
     }
 
+    repl::OpTime getPrepareOpTime() const {
+        stdx::lock_guard<stdx::mutex> lk(_mutex);
+        return _prepareOpTime;
+    }
+
     const Locker* getTxnResourceStashLockerForTest() const {
         stdx::lock_guard<stdx::mutex> lk(_mutex);
         invariant(_txnResourceStash);
@@ -625,6 +635,9 @@ private:
     // transaction. Currently only needed for diagnostics reporting.
     boost::optional<bool> _autoCommit;
 
+    // Track the prepareOpTime, the OpTime of the 'prepare' oplog entry for a transaction.
+    repl::OpTime _prepareOpTime;
+
     // The OpTime a speculative transaction is reading from and also the earliest opTime it
     // should wait for write concern for on commit.
     repl::OpTime _speculativeTransactionReadOpTime;