SERVER-39441 Write the new 'prepareTransaction' command on primary

21 files changed, 345 insertions, 144 deletions

diff --git a/src/mongo/db/auth/auth_op_observer.h b/src/mongo/db/auth/auth_op_observer.h
index a8afc4baefc..a31e09cdfe9 100644
--- a/src/mongo/db/auth/auth_op_observer.h
+++ b/src/mongo/db/auth/auth_op_observer.h
@@ -166,7 +166,7 @@ public:
         const std::vector<repl::ReplOperation>& statements) noexcept final {}
 
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) final {}
 
     void onTransactionAbort(OperationContext* opCtx,
diff --git a/src/mongo/db/catalog/uuid_catalog.h b/src/mongo/db/catalog/uuid_catalog.h
index 89dc3bac15a..d302e04354f 100644
--- a/src/mongo/db/catalog/uuid_catalog.h
+++ b/src/mongo/db/catalog/uuid_catalog.h
@@ -160,7 +160,7 @@ public:
         Timestamp commitTimestamp,
         const std::vector<repl::ReplOperation>& statements) noexcept override {}
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) override {}
     void onTransactionAbort(OperationContext* opCtx,
                             boost::optional<OplogSlot> abortOplogEntryOpTime) override {}
diff --git a/src/mongo/db/free_mon/free_mon_op_observer.h b/src/mongo/db/free_mon/free_mon_op_observer.h
index 9aec975e8ec..0a2ca1ca1e7 100644
--- a/src/mongo/db/free_mon/free_mon_op_observer.h
+++ b/src/mongo/db/free_mon/free_mon_op_observer.h
@@ -167,7 +167,7 @@ public:
         const std::vector<repl::ReplOperation>& statements) noexcept final {}
 
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) final {}
 
     void onTransactionAbort(OperationContext* opCtx,
diff --git a/src/mongo/db/op_observer.h b/src/mongo/db/op_observer.h
index a6664937df6..dadbe435cc5 100644
--- a/src/mongo/db/op_observer.h
+++ b/src/mongo/db/op_observer.h
@@ -309,12 +309,13 @@ public:
      * The onTransactionPrepare method is called when an atomic transaction is prepared. It must be
      * called when a transaction is active.
      *
-     * The 'prepareOpTime' is passed in to be used as the OpTime of the oplog entry.
+     * 'reservedSlots' is a list of oplog slots reserved for the oplog entries in a transaction. The
+     * last reserved slot represents the prepareOpTime used for the prepare oplog entry.
      *
      * The 'statements' are the list of CRUD operations to be applied in this transaction.
      */
     virtual void onTransactionPrepare(OperationContext* opCtx,
-                                      const OplogSlot& prepareOpTime,
+                                      const std::vector<OplogSlot>& reservedSlots,
                                       std::vector<repl::ReplOperation>& statements) = 0;
 
     /**
diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp
index 568d70a504e..88ce25e15fb 100644
--- a/src/mongo/db/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer_impl.cpp
@@ -1087,11 +1087,15 @@ OpTimeBundle logReplOperationForTransaction(OperationContext* opCtx,
     return times;
 }
 
+// This function expects that the size of 'oplogSlots' be at least as big as the size of 'stmts'. If
+// there are more oplog slots than statements, then only the first n slots are used, where n is the
+// size of 'stmts'.
 void logOplogEntriesForTransaction(OperationContext* opCtx,
                                    const std::vector<repl::ReplOperation>& stmts,
                                    const std::vector<OplogSlot>& oplogSlots) {
     invariant(!stmts.empty());
-    invariant(stmts.size() == oplogSlots.size());
+    invariant(stmts.size() <= oplogSlots.size());
+
     OperationSessionInfo sessionInfo;
     sessionInfo.setSessionId(*opCtx->getLogicalSessionId());
     sessionInfo.setTxnNumber(*opCtx->getTxnNumber());
@@ -1246,6 +1250,7 @@ void OpObserverImpl::onUnpreparedTransactionCommit(
         auto oplogSlots = repl::getNextOpTimes(opCtx, statements.size() + 1);
         auto commitSlot = oplogSlots.back();
         oplogSlots.pop_back();
+        invariant(oplogSlots.size() == statements.size());
         logOplogEntriesForTransaction(opCtx, statements, oplogSlots);
         commitOpTime = logCommitForUnpreparedTransaction(
             opCtx, statements.size() /* stmtId */, oplogSlots.back().opTime, commitSlot);
@@ -1275,9 +1280,48 @@ void OpObserverImpl::onPreparedTransactionCommit(
     shardObserveTransactionCommit(opCtx, statements, commitOplogEntryOpTime.opTime, true);
 }
 
+repl::OpTime logPrepareTransaction(OperationContext* opCtx,
+                                   StmtId stmtId,
+                                   const repl::OpTime& prevOpTime,
+                                   const OplogSlot oplogSlot) {
+    const NamespaceString cmdNss{"admin", "$cmd"};
+
+    const auto wallClockTime = getWallClockTimeForOpLog(opCtx);
+
+    OperationSessionInfo sessionInfo;
+    repl::OplogLink oplogLink;
+    PrepareTransactionOplogObject cmdObj;
+    sessionInfo.setSessionId(*opCtx->getLogicalSessionId());
+    sessionInfo.setTxnNumber(*opCtx->getTxnNumber());
+    oplogLink.prevOpTime = prevOpTime;
+
+    const auto oplogOpTime = logOperation(opCtx,
+                                          "c",
+                                          cmdNss,
+                                          {} /* uuid */,
+                                          cmdObj.toBSON(),
+                                          nullptr /* o2 */,
+                                          false /* fromMigrate */,
+                                          wallClockTime,
+                                          sessionInfo,
+                                          stmtId,
+                                          oplogLink,
+                                          false /* prepare */,
+                                          false /* inTxn */,
+                                          oplogSlot);
+    invariant(oplogSlot.opTime == oplogOpTime);
+
+    onWriteOpCompleted(
+        opCtx, cmdNss, {stmtId}, oplogOpTime, wallClockTime, DurableTxnStateEnum::kPrepared);
+
+    return oplogSlot.opTime;
+}
+
 void OpObserverImpl::onTransactionPrepare(OperationContext* opCtx,
-                                          const OplogSlot& prepareOpTime,
+                                          const std::vector<OplogSlot>& reservedSlots,
                                           std::vector<repl::ReplOperation>& statements) {
+    invariant(!reservedSlots.empty());
+    const auto prepareOpTime = reservedSlots.back();
     invariant(opCtx->getTxnNumber());
     invariant(!prepareOpTime.opTime.isNull());
 
@@ -1291,30 +1335,45 @@ void OpObserverImpl::onTransactionPrepare(OperationContext* opCtx,
         // transaction.
         // We write an empty 'applyOps' entry if there were no writes to choose a prepare timestamp
         // and allow this transaction to be continued on failover.
-        {
-            TransactionParticipant::SideTransactionBlock sideTxn(opCtx);
+        TransactionParticipant::SideTransactionBlock sideTxn(opCtx);
 
-            writeConflictRetry(
-                opCtx, "onTransactionPrepare", NamespaceString::kRsOplogNamespace.ns(), [&] {
+        writeConflictRetry(
+            opCtx, "onTransactionPrepare", NamespaceString::kRsOplogNamespace.ns(), [&] {
 
-                    // Writes to the oplog only require a Global intent lock.
-                    Lock::GlobalLock globalLock(opCtx, MODE_IX);
+                // Writes to the oplog only require a Global intent lock.
+                Lock::GlobalLock globalLock(opCtx, MODE_IX);
 
-                    WriteUnitOfWork wuow(opCtx);
-                    logApplyOpsForTransaction(opCtx, statements, prepareOpTime);
-                    wuow.commit();
-                });
-        }
+                WriteUnitOfWork wuow(opCtx);
+                logApplyOpsForTransaction(opCtx, statements, prepareOpTime);
+                wuow.commit();
+            });
     } else {
-        // It is possible that the transaction resulted in no changes.  In that case, we should
-        // not write any operations other than the prepare oplog entry.
-        if (!statements.empty()) {
-            // Reserve all the optimes in advance, so we only need to get the opTime mutex once.
-            // TODO (SERVER-39441): Reserve an extra slot here for the prepare oplog entry.
-            TransactionParticipant::SideTransactionBlock sideTxn(opCtx);
-            auto oplogSlots = repl::getNextOpTimes(opCtx, statements.size());
-            logOplogEntriesForTransaction(opCtx, statements, oplogSlots);
-        }
+        // We should have reserved an extra oplog slot for the prepare oplog entry.
+        invariant(reservedSlots.size() == statements.size() + 1);
+        TransactionParticipant::SideTransactionBlock sideTxn(opCtx);
+
+        // prevOpTime is a null OpTime if there were no operations written other than the prepare
+        // oplog entry.
+        repl::OpTime prevOpTime;
+
+        writeConflictRetry(
+            opCtx, "onTransactionPrepare", NamespaceString::kRsOplogNamespace.ns(), [&] {
+                // Writes to the oplog only require a Global intent lock.
+                Lock::GlobalLock globalLock(opCtx, MODE_IX);
+
+                WriteUnitOfWork wuow(opCtx);
+                // It is possible that the transaction resulted in no changes, In that case, we
+                // should not write any operations other than the prepare oplog entry.
+                if (!statements.empty()) {
+                    logOplogEntriesForTransaction(opCtx, statements, reservedSlots);
+
+                    // The prevOpTime is the OpTime of the second last entry in the reserved slots.
+                    prevOpTime = reservedSlots.rbegin()[1].opTime;
+                }
+                logPrepareTransaction(
+                    opCtx, statements.size() /* stmtId */, prevOpTime, prepareOpTime);
+                wuow.commit();
+            });
     }
 }
 
diff --git a/src/mongo/db/op_observer_impl.h b/src/mongo/db/op_observer_impl.h
index 86b47b05b31..2b0f55c9c58 100644
--- a/src/mongo/db/op_observer_impl.h
+++ b/src/mongo/db/op_observer_impl.h
@@ -145,8 +145,8 @@ public:
         Timestamp commitTimestamp,
         const std::vector<repl::ReplOperation>& statements) noexcept final;
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
-                              std::vector<repl::ReplOperation>& statments) final;
+                              const std::vector<OplogSlot>& reservedSlots,
+                              std::vector<repl::ReplOperation>& statements) final;
     void onTransactionAbort(OperationContext* opCtx,
                             boost::optional<OplogSlot> abortOplogEntryOpTime) final;
     void onReplicationRollback(OperationContext* opCtx, const RollbackObserverInfo& rbInfo) final;
diff --git a/src/mongo/db/op_observer_impl_test.cpp b/src/mongo/db/op_observer_impl_test.cpp
index 549bceec71b..ef26aa3d60f 100644
--- a/src/mongo/db/op_observer_impl_test.cpp
+++ b/src/mongo/db/op_observer_impl_test.cpp
@@ -746,7 +746,7 @@ TEST_F(OpObserverTransactionTest, TransactionalPrepareTest) {
         txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
         opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
         opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+            opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
     }
 
     auto oplogEntryObj = getSingleOplogEntry(opCtx());
@@ -821,7 +821,7 @@ TEST_F(OpObserverTransactionTest, TransactionalPreparedCommitTest) {
         txnParticipant.transitionToPreparedforTest(opCtx(), prepareSlot.opTime);
         prepareTimestamp = prepareSlot.opTime.getTimestamp();
         opObserver().onTransactionPrepare(
-            opCtx(), prepareSlot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+            opCtx(), {prepareSlot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
 
         commitSlot = repl::getNextOpTime(opCtx());
     }
@@ -892,7 +892,7 @@ TEST_F(OpObserverTransactionTest, TransactionalPreparedAbortTest) {
         const auto prepareSlot = repl::getNextOpTime(opCtx());
         txnParticipant.transitionToPreparedforTest(opCtx(), prepareSlot.opTime);
         opObserver().onTransactionPrepare(
-            opCtx(), prepareSlot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+            opCtx(), {prepareSlot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
         abortSlot = repl::getNextOpTime(opCtx());
     }
 
@@ -972,7 +972,7 @@ TEST_F(OpObserverTransactionTest, PreparingEmptyTransactionLogsEmptyApplyOps) {
         txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
         opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
         opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+            opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
     }
 
     auto oplogEntryObj = getSingleOplogEntry(opCtx());
@@ -997,7 +997,7 @@ TEST_F(OpObserverTransactionTest, PreparingTransactionWritesToTransactionTable)
         txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
         prepareOpTime = slot.opTime;
         opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+            opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
         opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
     }
 
@@ -1030,7 +1030,7 @@ TEST_F(OpObserverTransactionTest, AbortingPreparedTransactionWritesToTransaction
         OplogSlot slot = repl::getNextOpTime(opCtx());
         opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
         opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+            opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
         txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
         abortSlot = repl::getNextOpTime(opCtx());
     }
@@ -1099,7 +1099,7 @@ TEST_F(OpObserverTransactionTest, CommittingPreparedTransactionWritesToTransacti
         prepareOpTime = slot.opTime;
         opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
         opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+            opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
         txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
     }
 
@@ -1557,19 +1557,24 @@ TEST_F(OpObserverMultiEntryTransactionTest,
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "prepareTransaction");
     repl::OpTime prepareOpTime;
-    {
-        WriteUnitOfWork wuow(opCtx());
-        OplogSlot slot = repl::getNextOpTime(opCtx());
-        txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
-        prepareOpTime = slot.opTime;
-        opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
-        opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
-    }
-    getNOplogEntries(opCtx(), 0);
+    auto reservedSlots = repl::getNextOpTimes(opCtx(), 1);
+    prepareOpTime = reservedSlots.back().opTime;
+    opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp());
+    opObserver().onTransactionPrepare(
+        opCtx(), reservedSlots, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+
+    auto oplogEntryObjs = getNOplogEntries(opCtx(), 1);
+    auto prepareEntryObj = oplogEntryObjs.back();
+    const auto prepareOplogEntry = assertGet(OplogEntry::parse(prepareEntryObj));
+    checkSessionAndTransactionFields(prepareEntryObj, 0);
 
-    // TODO (SERVER-39441): Make sure the prepare oplog entry is written and the durable transaction
-    // state is kPrepared.
+    ASSERT_EQ(prepareOpTime.getTimestamp(), opCtx()->recoveryUnit()->getPrepareTimestamp());
+    ASSERT_BSONOBJ_EQ(BSON("prepareTransaction" << 1), prepareOplogEntry.getObject());
+    txnParticipant.stashTransactionResources(opCtx());
+    assertTxnRecord(txnNum(), prepareOpTime, DurableTxnStateEnum::kPrepared);
+    txnParticipant.unstashTransactionResources(opCtx(), "abortTransaction");
+
+    ASSERT_EQ(prepareOpTime, txnParticipant.getLastWriteOpTime());
 }
 
 TEST_F(OpObserverMultiEntryTransactionTest, TransactionalInsertPrepareTest) {
@@ -1580,7 +1585,6 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalInsertPrepareTest) {
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "insert");
 
-    WriteUnitOfWork wuow(opCtx());
     AutoGetCollection autoColl1(opCtx(), nss1, MODE_IX);
     AutoGetCollection autoColl2(opCtx(), nss2, MODE_IX);
 
@@ -1594,26 +1598,31 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalInsertPrepareTest) {
     opObserver().onInserts(opCtx(), nss1, uuid1, inserts1.begin(), inserts1.end(), false);
     opObserver().onInserts(opCtx(), nss2, uuid2, inserts2.begin(), inserts2.end(), false);
 
-    {
-        WriteUnitOfWork wuow(opCtx());
-        OplogSlot slot = repl::getNextOpTime(opCtx());
-        txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
-        opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
-        opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
-    }
-    // TODO (SERVER-39441): Account for prepare oplog entry.
-    auto oplogEntryObjs = getNOplogEntries(opCtx(), 4);
+    repl::OpTime prepareOpTime;
+    auto reservedSlots = repl::getNextOpTimes(opCtx(), 5);
+    prepareOpTime = reservedSlots.back().opTime;
+    txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime);
+    opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp());
+    opObserver().onTransactionPrepare(
+        opCtx(), reservedSlots, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
+    auto oplogEntryObjs = getNOplogEntries(opCtx(), 5);
     StmtId expectedStmtId = 0;
     std::vector<OplogEntry> oplogEntries;
     mongo::repl::OpTime expectedPrevWriteOpTime;
     for (const auto& oplogEntryObj : oplogEntryObjs) {
-        checkSessionAndTransactionFields(oplogEntryObj, expectedStmtId++);
+        checkSessionAndTransactionFields(oplogEntryObj, expectedStmtId);
         oplogEntries.push_back(assertGet(OplogEntry::parse(oplogEntryObj)));
         const auto& oplogEntry = oplogEntries.back();
-        ASSERT_TRUE(oplogEntry.isCrudOpType());
-        ASSERT(oplogEntry.getOpType() == repl::OpTypeEnum::kInsert);
-        ASSERT_TRUE(oplogEntry.getInTxn());
+        if (expectedStmtId++ < 4) {
+            ASSERT_TRUE(oplogEntry.isCrudOpType());
+            ASSERT(oplogEntry.getOpType() == repl::OpTypeEnum::kInsert);
+            ASSERT_TRUE(oplogEntry.getInTxn());
+        } else {
+            ASSERT_EQ("admin.$cmd"_sd, oplogEntry.getNss().toString());
+            ASSERT_TRUE(oplogEntry.isCommand());
+            ASSERT_TRUE(OplogEntry::CommandType::kPrepareTransaction ==
+                        oplogEntry.getCommandType());
+        }
         ASSERT(!oplogEntry.getPrepare());
         ASSERT_TRUE(oplogEntry.getPrevWriteOpTimeInTransaction());
         ASSERT_EQ(expectedPrevWriteOpTime, *oplogEntry.getPrevWriteOpTimeInTransaction());
@@ -1640,8 +1649,14 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalInsertPrepareTest) {
     ASSERT_BSONOBJ_EQ(BSON("_id" << 3), oplogEntries[3].getObject());
     ASSERT_FALSE(oplogEntries[3].getObject2());
 
-    // TODO (SERVER-39441): Test that the last oplog entry has the "prepareTransaction" command and
-    // the durable transaction state is kPrepared.
+    ASSERT_EQ(prepareOpTime.getTimestamp(), opCtx()->recoveryUnit()->getPrepareTimestamp());
+    ASSERT_BSONOBJ_EQ(BSON("prepareTransaction" << 1), oplogEntries[4].getObject());
+    ASSERT_FALSE(oplogEntries[4].getObject2());
+
+    ASSERT_EQ(prepareOpTime, txnParticipant.getLastWriteOpTime());
+    txnParticipant.stashTransactionResources(opCtx());
+    assertTxnRecord(txnNum(), prepareOpTime, DurableTxnStateEnum::kPrepared);
+    txnParticipant.unstashTransactionResources(opCtx(), "abortTransaction");
 }
 
 TEST_F(OpObserverMultiEntryTransactionTest, TransactionalUpdatePrepareTest) {
@@ -1670,33 +1685,37 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalUpdatePrepareTest) {
     updateArgs2.criteria = BSON("_id" << 1);
     OplogUpdateEntryArgs update2(std::move(updateArgs2), nss2, uuid2);
 
-    WriteUnitOfWork wuow(opCtx());
     AutoGetCollection autoColl1(opCtx(), nss1, MODE_IX);
     AutoGetCollection autoColl2(opCtx(), nss2, MODE_IX);
     opObserver().onUpdate(opCtx(), update1);
     opObserver().onUpdate(opCtx(), update2);
 
-    {
-        WriteUnitOfWork wuow(opCtx());
-        OplogSlot slot = repl::getNextOpTime(opCtx());
-        txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
-        opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
-        opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
-    }
+    repl::OpTime prepareOpTime;
+    auto reservedSlots = repl::getNextOpTimes(opCtx(), 3);
+    prepareOpTime = reservedSlots.back().opTime;
+    txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime);
+    opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp());
+    opObserver().onTransactionPrepare(
+        opCtx(), reservedSlots, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
 
-    // TODO (SERVER-39441): Account for prepare oplog entry.
-    auto oplogEntryObjs = getNOplogEntries(opCtx(), 2);
+    auto oplogEntryObjs = getNOplogEntries(opCtx(), 3);
     StmtId expectedStmtId = 0;
     std::vector<OplogEntry> oplogEntries;
     mongo::repl::OpTime expectedPrevWriteOpTime;
     for (const auto& oplogEntryObj : oplogEntryObjs) {
-        checkSessionAndTransactionFields(oplogEntryObj, expectedStmtId++);
+        checkSessionAndTransactionFields(oplogEntryObj, expectedStmtId);
         oplogEntries.push_back(assertGet(OplogEntry::parse(oplogEntryObj)));
         const auto& oplogEntry = oplogEntries.back();
-        ASSERT_TRUE(oplogEntry.isCrudOpType());
-        ASSERT(oplogEntry.getOpType() == repl::OpTypeEnum::kUpdate);
-        ASSERT_TRUE(oplogEntry.getInTxn());
+        if (expectedStmtId++ < 2) {
+            ASSERT_TRUE(oplogEntry.isCrudOpType());
+            ASSERT(oplogEntry.getOpType() == repl::OpTypeEnum::kUpdate);
+            ASSERT_TRUE(oplogEntry.getInTxn());
+        } else {
+            ASSERT_EQ("admin.$cmd"_sd, oplogEntry.getNss().toString());
+            ASSERT_TRUE(oplogEntry.isCommand());
+            ASSERT_TRUE(OplogEntry::CommandType::kPrepareTransaction ==
+                        oplogEntry.getCommandType());
+        }
         ASSERT(!oplogEntry.getPrepare());
         ASSERT_TRUE(oplogEntry.getPrevWriteOpTimeInTransaction());
         ASSERT_EQ(expectedPrevWriteOpTime, *oplogEntry.getPrevWriteOpTimeInTransaction());
@@ -1719,8 +1738,15 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalUpdatePrepareTest) {
     ASSERT_TRUE(oplogEntries[1].getObject2());
     ASSERT_BSONOBJ_EQ(*oplogEntries[1].getObject2(), BSON("_id" << 1));
 
-    // TODO (SERVER-39441): Test that the last oplog entry has the "prepareTransaction" command and
-    // the durable transaction state is kPrepared.
+    ASSERT_EQ(prepareOpTime.getTimestamp(), opCtx()->recoveryUnit()->getPrepareTimestamp());
+    ASSERT_BSONOBJ_EQ(BSON("prepareTransaction" << 1), oplogEntries[2].getObject());
+    ASSERT_FALSE(oplogEntries[2].getObject2());
+
+    ASSERT_EQ(prepareOpTime, txnParticipant.getLastWriteOpTime());
+
+    txnParticipant.stashTransactionResources(opCtx());
+    assertTxnRecord(txnNum(), prepareOpTime, DurableTxnStateEnum::kPrepared);
+    txnParticipant.unstashTransactionResources(opCtx(), "abortTransaction");
 }
 
 TEST_F(OpObserverMultiEntryTransactionTest, TransactionalDeletePrepareTest) {
@@ -1732,7 +1758,6 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalDeletePrepareTest) {
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "delete");
 
-    WriteUnitOfWork wuow(opCtx());
     AutoGetCollection autoColl1(opCtx(), nss1, MODE_IX);
     AutoGetCollection autoColl2(opCtx(), nss2, MODE_IX);
     opObserver().aboutToDelete(opCtx(),
@@ -1746,27 +1771,33 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalDeletePrepareTest) {
                                           << "y"));
     opObserver().onDelete(opCtx(), nss2, uuid2, 0, false, boost::none);
 
-    {
-        WriteUnitOfWork wuow(opCtx());
-        OplogSlot slot = repl::getNextOpTime(opCtx());
-        txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime);
-        opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp());
-        opObserver().onTransactionPrepare(
-            opCtx(), slot, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
-    }
+    repl::OpTime prepareOpTime;
+    auto reservedSlots = repl::getNextOpTimes(opCtx(), 3);
+    prepareOpTime = reservedSlots.back().opTime;
+    txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime);
+    prepareOpTime = reservedSlots.back().opTime;
+    opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp());
+    opObserver().onTransactionPrepare(
+        opCtx(), reservedSlots, txnParticipant.retrieveCompletedTransactionOperations(opCtx()));
 
-    // TODO (SERVER-39441): Account for prepare oplog entry.
-    auto oplogEntryObjs = getNOplogEntries(opCtx(), 2);
+    auto oplogEntryObjs = getNOplogEntries(opCtx(), 3);
     StmtId expectedStmtId = 0;
     std::vector<OplogEntry> oplogEntries;
     mongo::repl::OpTime expectedPrevWriteOpTime;
     for (const auto& oplogEntryObj : oplogEntryObjs) {
-        checkSessionAndTransactionFields(oplogEntryObj, expectedStmtId++);
+        checkSessionAndTransactionFields(oplogEntryObj, expectedStmtId);
         oplogEntries.push_back(assertGet(OplogEntry::parse(oplogEntryObj)));
         const auto& oplogEntry = oplogEntries.back();
-        ASSERT_TRUE(oplogEntry.isCrudOpType());
-        ASSERT(oplogEntry.getOpType() == repl::OpTypeEnum::kDelete);
-        ASSERT_TRUE(oplogEntry.getInTxn());
+        if (expectedStmtId++ < 2) {
+            ASSERT_TRUE(oplogEntry.isCrudOpType());
+            ASSERT(oplogEntry.getOpType() == repl::OpTypeEnum::kDelete);
+            ASSERT_TRUE(oplogEntry.getInTxn());
+        } else {
+            ASSERT_EQ("admin.$cmd"_sd, oplogEntry.getNss().toString());
+            ASSERT_TRUE(oplogEntry.isCommand());
+            ASSERT_TRUE(OplogEntry::CommandType::kPrepareTransaction ==
+                        oplogEntry.getCommandType());
+        }
         ASSERT(!oplogEntry.getPrepare());
         ASSERT_TRUE(oplogEntry.getPrevWriteOpTimeInTransaction());
         ASSERT_EQ(expectedPrevWriteOpTime, *oplogEntry.getPrevWriteOpTimeInTransaction());
@@ -1783,8 +1814,14 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalDeletePrepareTest) {
     ASSERT_BSONOBJ_EQ(oplogEntries[1].getObject(), BSON("_id" << 1));
     ASSERT_FALSE(oplogEntries[1].getObject2());
 
-    // TODO (SERVER-39441): Test that the last oplog entry has the "prepareTransaction" command and
-    // the durable transaction state is kPrepared.
+    ASSERT_EQ(prepareOpTime.getTimestamp(), opCtx()->recoveryUnit()->getPrepareTimestamp());
+    ASSERT_BSONOBJ_EQ(BSON("prepareTransaction" << 1), oplogEntries[2].getObject());
+    ASSERT_FALSE(oplogEntries[2].getObject2());
+
+    ASSERT_EQ(prepareOpTime, txnParticipant.getLastWriteOpTime());
+    txnParticipant.stashTransactionResources(opCtx());
+    assertTxnRecord(txnNum(), prepareOpTime, DurableTxnStateEnum::kPrepared);
+    txnParticipant.unstashTransactionResources(opCtx(), "abortTransaction");
 }
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/op_observer_noop.h b/src/mongo/db/op_observer_noop.h
index b4290305eca..6c72a04aa25 100644
--- a/src/mongo/db/op_observer_noop.h
+++ b/src/mongo/db/op_observer_noop.h
@@ -144,7 +144,7 @@ public:
         Timestamp commitTimestamp,
         const std::vector<repl::ReplOperation>& statements) noexcept override{};
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) override{};
     void onTransactionAbort(OperationContext* opCtx,
                             boost::optional<OplogSlot> abortOplogEntryOpTime) override{};
diff --git a/src/mongo/db/op_observer_registry.h b/src/mongo/db/op_observer_registry.h
index 946ee8893e9..d1c0713ab7d 100644
--- a/src/mongo/db/op_observer_registry.h
+++ b/src/mongo/db/op_observer_registry.h
@@ -277,11 +277,11 @@ public:
     }
 
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) override {
         ReservedTimes times{opCtx};
         for (auto& observer : _observers) {
-            observer->onTransactionPrepare(opCtx, prepareOpTime, statements);
+            observer->onTransactionPrepare(opCtx, reservedSlots, statements);
         }
     }
 
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 06dbcdc93a0..6838330d601 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -818,17 +818,6 @@ void createOplog(OperationContext* opCtx) {
     createOplog(opCtx, localOplogInfo(opCtx->getServiceContext()).oplogName, isReplSet);
 }
 
-MONGO_REGISTER_SHIM(GetNextOpTimeClass::getNextOpTime)(OperationContext* opCtx)->OplogSlot {
-    // The local oplog collection pointer must already be established by this point.
-    // We can't establish it here because that would require locking the local database, which would
-    // be a lock order violation.
-    auto oplog = localOplogInfo(opCtx->getServiceContext()).oplog;
-    invariant(oplog);
-    OplogSlot os;
-    _getNextOpTimes(opCtx, oplog, 1, &os);
-    return os;
-}
-
 OplogSlot getNextOpTimeNoPersistForTesting(OperationContext* opCtx) {
     auto oplog = localOplogInfo(opCtx->getServiceContext()).oplog;
     invariant(oplog);
@@ -838,7 +827,8 @@ OplogSlot getNextOpTimeNoPersistForTesting(OperationContext* opCtx) {
     return os;
 }
 
-std::vector<OplogSlot> getNextOpTimes(OperationContext* opCtx, std::size_t count) {
+MONGO_REGISTER_SHIM(GetNextOpTimeClass::getNextOpTimes)
+(OperationContext* opCtx, std::size_t count)->std::vector<OplogSlot> {
     // The local oplog collection pointer must already be established by this point.
     // We can't establish it here because that would require locking the local database, which would
     // be a lock order violation.
@@ -850,7 +840,6 @@ std::vector<OplogSlot> getNextOpTimes(OperationContext* opCtx, std::size_t count
     return oplogSlots;
 }
 
-
 // -------------------------------------
 
 namespace {
diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h
index d04dab93a78..86be44cc30d 100644
--- a/src/mongo/db/repl/oplog.h
+++ b/src/mongo/db/repl/oplog.h
@@ -267,15 +267,21 @@ void createIndexForApplyOps(OperationContext* opCtx,
 // workaround.
 struct GetNextOpTimeClass {
     /**
-     * Allocates optimes for new entries in the oplog.  Returns an OplogSlot or a vector of
-     * OplogSlots, which contain the new optimes along with their terms and newly calculated hash
-     * fields.
+     * Allocates optimes for new entries in the oplog.  Returns a vector of OplogSlots, which
+     * contain the new optimes along with their terms and newly calculated hash fields.
      */
-    static MONGO_DECLARE_SHIM((OperationContext * opCtx)->OplogSlot) getNextOpTime;
+    static MONGO_DECLARE_SHIM((OperationContext * opCtx, std::size_t count)->std::vector<OplogSlot>)
+        getNextOpTimes;
+};
+
+inline std::vector<OplogSlot> getNextOpTimes(OperationContext* opCtx, std::size_t count) {
+    return GetNextOpTimeClass::getNextOpTimes(opCtx, count);
 };
 
 inline OplogSlot getNextOpTime(OperationContext* opCtx) {
-    return GetNextOpTimeClass::getNextOpTime(opCtx);
+    auto slots = getNextOpTimes(opCtx, 1);
+    invariant(slots.size() == 1);
+    return slots.back();
 }
 
 /**
@@ -290,7 +296,6 @@ inline OplogSlot getNextOpTime(OperationContext* opCtx) {
  * prepare generates an oplog entry in a separate unit of work.
  */
 OplogSlot getNextOpTimeNoPersistForTesting(OperationContext* opCtx);
-std::vector<OplogSlot> getNextOpTimes(OperationContext* opCtx, std::size_t count);
 
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/oplog_entry.cpp b/src/mongo/db/repl/oplog_entry.cpp
index 41c6df4eee3..9b97b178976 100644
--- a/src/mongo/db/repl/oplog_entry.cpp
+++ b/src/mongo/db/repl/oplog_entry.cpp
@@ -77,6 +77,8 @@ OplogEntry::CommandType parseCommandType(const BSONObj& objectField) {
         return OplogEntry::CommandType::kCommitTransaction;
     } else if (commandString == "abortTransaction") {
         return OplogEntry::CommandType::kAbortTransaction;
+    } else if (commandString == "prepareTransaction") {
+        return OplogEntry::CommandType::kPrepareTransaction;
     } else {
         uasserted(ErrorCodes::BadValue,
                   str::stream() << "Unknown oplog entry command type: " << commandString
diff --git a/src/mongo/db/repl/oplog_entry.h b/src/mongo/db/repl/oplog_entry.h
index c385e813552..de984061e6b 100644
--- a/src/mongo/db/repl/oplog_entry.h
+++ b/src/mongo/db/repl/oplog_entry.h
@@ -62,6 +62,7 @@ public:
         kDropIndexes,
         kCommitTransaction,
         kAbortTransaction,
+        kPrepareTransaction,
     };
 
     // Current oplog version, should be the value of the v field in all oplog entries.
diff --git a/src/mongo/db/repl/oplog_shim.cpp b/src/mongo/db/repl/oplog_shim.cpp
index 24840a3aae9..f6efbe31683 100644
--- a/src/mongo/db/repl/oplog_shim.cpp
+++ b/src/mongo/db/repl/oplog_shim.cpp
@@ -31,6 +31,6 @@
 
 namespace mongo {
 namespace repl {
-MONGO_DEFINE_SHIM(GetNextOpTimeClass::getNextOpTime);
+MONGO_DEFINE_SHIM(GetNextOpTimeClass::getNextOpTimes);
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/s/config_server_op_observer.h b/src/mongo/db/s/config_server_op_observer.h
index 2ab3ea5e89b..08e1dfd0cf4 100644
--- a/src/mongo/db/s/config_server_op_observer.h
+++ b/src/mongo/db/s/config_server_op_observer.h
@@ -167,7 +167,7 @@ public:
         const std::vector<repl::ReplOperation>& statements) noexcept override {}
 
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) override {}
 
     void onTransactionAbort(OperationContext* opCtx,
diff --git a/src/mongo/db/s/shard_server_op_observer.h b/src/mongo/db/s/shard_server_op_observer.h
index 8703db1dfce..e39089dbb32 100644
--- a/src/mongo/db/s/shard_server_op_observer.h
+++ b/src/mongo/db/s/shard_server_op_observer.h
@@ -168,7 +168,7 @@ public:
         const std::vector<repl::ReplOperation>& statements) noexcept override {}
 
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) override {}
 
     void onTransactionAbort(OperationContext* opCtx,
diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp
index 1f2301e3765..4cf82057024 100644
--- a/src/mongo/db/transaction_participant.cpp
+++ b/src/mongo/db/transaction_participant.cpp
@@ -551,7 +551,8 @@ void TransactionParticipant::Participant::_setSpeculativeTransactionReadTimestam
     o(lk).transactionMetricsObserver.onChooseReadTimestamp(timestamp);
 }
 
-TransactionParticipant::OplogSlotReserver::OplogSlotReserver(OperationContext* opCtx)
+TransactionParticipant::OplogSlotReserver::OplogSlotReserver(OperationContext* opCtx,
+                                                             int numSlotsToReserve)
     : _opCtx(opCtx) {
     // Stash the transaction on the OperationContext on the stack. At the end of this function it
     // will be unstashed onto the OperationContext.
@@ -559,7 +560,7 @@ TransactionParticipant::OplogSlotReserver::OplogSlotReserver(OperationContext* o
 
     // Begin a new WUOW and reserve a slot in the oplog.
     WriteUnitOfWork wuow(opCtx);
-    _oplogSlot = repl::getNextOpTime(opCtx);
+    _oplogSlots = repl::getNextOpTimes(opCtx, numSlotsToReserve);
 
     // Release the WUOW state since this WUOW is no longer in use.
     wuow.release();
@@ -936,12 +937,13 @@ Timestamp TransactionParticipant::Participant::prepareTransaction(
         opCtx->checkForInterrupt();
         o(lk).txnState.transitionTo(TransactionState::kPrepared);
     }
-
+    std::vector<OplogSlot> reservedSlots;
     if (prepareOptime) {
         // On secondary, we just prepare the transaction and discard the buffered ops.
         prepareOplogSlot = OplogSlot(*prepareOptime, 0);
         stdx::lock_guard<Client> lk(*opCtx->getClient());
         o(lk).prepareOpTime = *prepareOptime;
+        reservedSlots.push_back(prepareOplogSlot);
     } else {
         // On primary, we reserve an optime, prepare the transaction and write the oplog entry.
         //
@@ -950,8 +952,16 @@ Timestamp TransactionParticipant::Participant::prepareTransaction(
         // being prepared. When the OplogSlotReserver goes out of scope and is destroyed, the
         // storage-transaction it uses to keep the hole open will abort and the slot (and
         // corresponding oplog hole) will vanish.
-        oplogSlotReserver.emplace(opCtx);
-        prepareOplogSlot = oplogSlotReserver->getReservedOplogSlot();
+        if (!gUseMultipleOplogEntryFormatForTransactions) {
+            oplogSlotReserver.emplace(opCtx);
+        } else {
+            const auto numSlotsToReserve = retrieveCompletedTransactionOperations(opCtx).size();
+            // Reserve an extra slot here for the prepare oplog entry.
+            oplogSlotReserver.emplace(opCtx, numSlotsToReserve + 1);
+            invariant(oplogSlotReserver->getSlots().size() >= 1);
+        }
+        prepareOplogSlot = oplogSlotReserver->getLastSlot();
+        reservedSlots = oplogSlotReserver->getSlots();
         invariant(o().prepareOpTime.isNull(),
                   str::stream() << "This transaction has already reserved a prepareOpTime at: "
                                 << o().prepareOpTime.toString());
@@ -971,9 +981,8 @@ Timestamp TransactionParticipant::Participant::prepareTransaction(
     }
     opCtx->recoveryUnit()->setPrepareTimestamp(prepareOplogSlot.opTime.getTimestamp());
     opCtx->getWriteUnitOfWork()->prepare();
-
     opCtx->getServiceContext()->getOpObserver()->onTransactionPrepare(
-        opCtx, prepareOplogSlot, retrieveCompletedTransactionOperations(opCtx));
+        opCtx, reservedSlots, retrieveCompletedTransactionOperations(opCtx));
 
     abortGuard.dismiss();
 
@@ -1149,7 +1158,7 @@ void TransactionParticipant::Participant::commitPreparedTransaction(
         if (opCtx->writesAreReplicated()) {
             invariant(!commitOplogEntryOpTime);
             oplogSlotReserver.emplace(opCtx);
-            commitOplogSlot = oplogSlotReserver->getReservedOplogSlot();
+            commitOplogSlot = oplogSlotReserver->getLastSlot();
             invariant(commitOplogSlot.opTime.getTimestamp() >= commitTimestamp,
                       str::stream() << "Commit oplog entry must be greater than or equal to commit "
                                        "timestamp due to causal consistency. commit timestamp: "
@@ -1334,7 +1343,7 @@ void TransactionParticipant::Participant::_abortActiveTransaction(
     boost::optional<OplogSlot> abortOplogSlot;
     if (o().txnState.isPrepared() && opCtx->writesAreReplicated()) {
         oplogSlotReserver.emplace(opCtx);
-        abortOplogSlot = oplogSlotReserver->getReservedOplogSlot();
+        abortOplogSlot = oplogSlotReserver->getLastSlot();
     }
 
     // Clean up the transaction resources on the opCtx even if the transaction resources on the
diff --git a/src/mongo/db/transaction_participant.h b/src/mongo/db/transaction_participant.h
index a3673ef544d..6d05089b284 100644
--- a/src/mongo/db/transaction_participant.h
+++ b/src/mongo/db/transaction_participant.h
@@ -796,22 +796,29 @@ private:
      */
     class OplogSlotReserver {
     public:
-        OplogSlotReserver(OperationContext* opCtx);
+        OplogSlotReserver(OperationContext* opCtx, int numSlotsToReserve = 1);
         ~OplogSlotReserver();
 
         /**
-         * Returns the oplog slot reserved at construction.
+         * Returns the latest oplog slot reserved at construction.
          */
-        OplogSlot getReservedOplogSlot() const {
-            invariant(!_oplogSlot.opTime.isNull());
-            return _oplogSlot;
+        OplogSlot getLastSlot() {
+            invariant(!_oplogSlots.empty());
+            invariant(!_oplogSlots.back().opTime.isNull());
+            return getSlots().back();
+        }
+
+        std::vector<OplogSlot>& getSlots() {
+            invariant(!_oplogSlots.empty());
+            invariant(!_oplogSlots.back().opTime.isNull());
+            return _oplogSlots;
         }
 
     private:
         OperationContext* _opCtx;
         std::unique_ptr<Locker> _locker;
         std::unique_ptr<RecoveryUnit> _recoveryUnit;
-        OplogSlot _oplogSlot;
+        std::vector<OplogSlot> _oplogSlots;
     };
 
     friend std::ostream& operator<<(std::ostream& s, TransactionState txnState) {
diff --git a/src/mongo/db/transaction_participant_retryable_writes_test.cpp b/src/mongo/db/transaction_participant_retryable_writes_test.cpp
index b78c62518a5..6f51d4ddf7c 100644
--- a/src/mongo/db/transaction_participant_retryable_writes_test.cpp
+++ b/src/mongo/db/transaction_participant_retryable_writes_test.cpp
@@ -87,10 +87,10 @@ repl::OplogEntry makeOplogEntry(repl::OpTime opTime,
 class OpObserverMock : public OpObserverNoop {
 public:
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) override {
         ASSERT_TRUE(opCtx->lockState()->inAWriteUnitOfWork());
-        OpObserverNoop::onTransactionPrepare(opCtx, prepareOpTime, statements);
+        OpObserverNoop::onTransactionPrepare(opCtx, reservedSlots, statements);
 
         uassert(ErrorCodes::OperationFailed,
                 "onTransactionPrepare() failed",
diff --git a/src/mongo/db/transaction_participant_test.cpp b/src/mongo/db/transaction_participant_test.cpp
index 02f9cd938d8..bd92dd240a9 100644
--- a/src/mongo/db/transaction_participant_test.cpp
+++ b/src/mongo/db/transaction_participant_test.cpp
@@ -93,7 +93,7 @@ repl::OplogEntry makeOplogEntry(repl::OpTime opTime,
 class OpObserverMock : public OpObserverNoop {
 public:
     void onTransactionPrepare(OperationContext* opCtx,
-                              const OplogSlot& prepareOpTime,
+                              const std::vector<OplogSlot>& reservedSlots,
                               std::vector<repl::ReplOperation>& statements) override;
 
     bool onTransactionPrepareThrowsException = false;
@@ -136,10 +136,10 @@ public:
 };
 
 void OpObserverMock::onTransactionPrepare(OperationContext* opCtx,
-                                          const OplogSlot& prepareOpTime,
+                                          const std::vector<OplogSlot>& reservedSlots,
                                           std::vector<repl::ReplOperation>& statements) {
     ASSERT_TRUE(opCtx->lockState()->inAWriteUnitOfWork());
-    OpObserverNoop::onTransactionPrepare(opCtx, prepareOpTime, statements);
+    OpObserverNoop::onTransactionPrepare(opCtx, reservedSlots, statements);
 
     uassert(ErrorCodes::OperationFailed,
             "onTransactionPrepare() failed",
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index d68a567f1e5..9cdf8834649 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -2902,6 +2902,96 @@ protected:
     Timestamp firstOplogEntryTs, secondOplogEntryTs;
 };
 
+class PreparedMultiOplogEntryTransaction : public MultiDocumentTransactionTest {
+public:
+    PreparedMultiOplogEntryTransaction()
+        : MultiDocumentTransactionTest("preparedMultiOplogEntryTransaction") {
+        gUseMultipleOplogEntryFormatForTransactions = true;
+        const auto currentTime = _clock->getClusterTime();
+        firstOplogEntryTs = currentTime.addTicks(1).asTimestamp();
+        secondOplogEntryTs = currentTime.addTicks(2).asTimestamp();
+        prepareEntryTs = currentTime.addTicks(3).asTimestamp();
+        commitEntryTs = currentTime.addTicks(4).asTimestamp();
+    }
+
+    ~PreparedMultiOplogEntryTransaction() {
+        gUseMultipleOplogEntryFormatForTransactions = false;
+    }
+
+    void run() {
+        auto txnParticipant = TransactionParticipant::get(_opCtx);
+        ASSERT(txnParticipant);
+        unittest::log() << "PrepareTS: " << prepareEntryTs;
+        logTimestamps();
+
+        const auto prepareFilter = BSON("ts" << prepareEntryTs);
+        const auto commitFilter = BSON("ts" << commitEntryTs);
+        {
+            AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS, LockMode::MODE_IS);
+            auto coll = autoColl.getCollection();
+            assertDocumentAtTimestamp(coll, presentTs, BSONObj());
+            assertDocumentAtTimestamp(coll, beforeTxnTs, BSONObj());
+            assertDocumentAtTimestamp(coll, firstOplogEntryTs, BSONObj());
+            assertDocumentAtTimestamp(coll, secondOplogEntryTs, BSONObj());
+            assertDocumentAtTimestamp(coll, prepareEntryTs, BSONObj());
+            assertDocumentAtTimestamp(coll, commitEntryTs, BSONObj());
+
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, presentTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, beforeTxnTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, firstOplogEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, secondOplogEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, prepareEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, commitEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, nullTs, false);
+
+            assertOplogDocumentExistsAtTimestamp(commitFilter, presentTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, beforeTxnTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, prepareEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, commitEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, nullTs, false);
+        }
+        txnParticipant.unstashTransactionResources(_opCtx, "insert");
+        const BSONObj doc2 = BSON("_id" << 2 << "TestValue" << 2);
+        {
+            AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IX, LockMode::MODE_IX);
+            insertDocument(autoColl.getCollection(), InsertStatement(doc2));
+        }
+        txnParticipant.prepareTransaction(_opCtx, {});
+
+        txnParticipant.stashTransactionResources(_opCtx);
+        {
+            AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS, LockMode::MODE_IS);
+            auto coll = autoColl.getCollection();
+            const BSONObj query1 = BSON("_id" << 1);
+            const BSONObj query2 = BSON("_id" << 2);
+            assertDocumentAtTimestamp(coll, presentTs, BSONObj());
+            assertDocumentAtTimestamp(coll, beforeTxnTs, BSONObj());
+            assertDocumentAtTimestamp(coll, prepareEntryTs, BSONObj());
+            assertDocumentAtTimestamp(coll, commitEntryTs, BSONObj());
+            assertDocumentAtTimestamp(coll, nullTs, BSONObj());
+
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, presentTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, beforeTxnTs, false);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, prepareEntryTs, true);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, commitEntryTs, true);
+            assertOplogDocumentExistsAtTimestamp(prepareFilter, nullTs, true);
+
+            // We haven't committed the prepared transaction
+            assertOplogDocumentExistsAtTimestamp(commitFilter, presentTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, beforeTxnTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, prepareEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, commitEntryTs, false);
+            assertOplogDocumentExistsAtTimestamp(commitFilter, nullTs, false);
+        }
+
+        // TODO (SERVER-39442): Commit the prepared transaction and assert existence of oplogs at
+        // commitTimestamp.
+    }
+
+protected:
+    Timestamp firstOplogEntryTs, secondOplogEntryTs, prepareEntryTs;
+};
+
 class PreparedMultiDocumentTransaction : public MultiDocumentTransactionTest {
 public:
     PreparedMultiDocumentTransaction()
@@ -3145,6 +3235,7 @@ public:
         add<CreateCollectionWithSystemIndex>();
         add<MultiDocumentTransaction>();
         add<MultiOplogEntryTransaction>();
+        add<PreparedMultiOplogEntryTransaction>();
         add<PreparedMultiDocumentTransaction>();
         add<AbortedPreparedMultiDocumentTransaction>();
     }