SERVER-41942: Clear partialTxnList after committing a prepared transaction during initial sync

2 files changed, 117 insertions, 0 deletions

diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp
index 7431107d29d..4781dc71500 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -1245,6 +1245,7 @@ void SyncTail::_fillWriterVectors(OperationContext* opCtx,
                 auto commitOplogEntryOpTime = op.getOpTime();
                 derivedOps->emplace_back(readTransactionOperationsFromOplogChain(
                     opCtx, prevOplogEntry, partialTxnList, commitOplogEntryOpTime.getTimestamp()));
+                partialTxnList.clear();
             }
 
             _fillWriterVectors(opCtx, &derivedOps->back(), writerVectors, derivedOps, nullptr);
diff --git a/src/mongo/db/repl/sync_tail_test.cpp b/src/mongo/db/repl/sync_tail_test.cpp
index 95c981f6ead..0c11fb1432b 100644
--- a/src/mongo/db/repl/sync_tail_test.cpp
+++ b/src/mongo/db/repl/sync_tail_test.cpp
@@ -3352,6 +3352,83 @@ TEST_F(IdempotencyTestTxns, CommitUnpreparedTransactionWithPartialTxnOps) {
     ASSERT_TRUE(docExists(_opCtx.get(), nss, doc2));
 }
 
+TEST_F(IdempotencyTestTxns, CommitTwoUnpreparedTransactionsWithPartialTxnOpsAtOnce) {
+    createCollectionWithUuid(_opCtx.get(), NamespaceString::kSessionTransactionsTableNamespace);
+    auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
+    auto lsid = makeLogicalSessionId(_opCtx.get());
+    TxnNumber txnNum1(1);
+    TxnNumber txnNum2(2);
+
+    auto partialOp1 = partialTxn(
+        lsid, txnNum1, StmtId(0), OpTime(), BSON_ARRAY(makeInsertApplyOpsEntry(nss, uuid, doc)));
+    auto commitOp1 =
+        commitUnprepared(lsid, txnNum1, StmtId(1), BSONArray(), partialOp1.getOpTime());
+
+    // The second transaction (with a different transaction number) in the same session.
+    auto partialOp2 = partialTxn(
+        lsid, txnNum2, StmtId(0), OpTime(), BSON_ARRAY(makeInsertApplyOpsEntry(nss, uuid, doc2)));
+    auto commitOp2 =
+        commitUnprepared(lsid, txnNum2, StmtId(1), BSONArray(), partialOp2.getOpTime());
+
+    ASSERT_OK(ReplicationCoordinator::get(getGlobalServiceContext())
+                  ->setFollowerMode(MemberState::RS_RECOVERING));
+
+    // This also tests that we clear the partialTxnList for the session after applying the commit of
+    // the first transaction. Otherwise, saving operations from the second transaction to the same
+    // partialTxnList as the first transaction will trigger an invariant because of the mismatching
+    // transaction numbers.
+    testOpsAreIdempotent({partialOp1, commitOp1, partialOp2, commitOp2});
+
+    // The transaction table should only contain the second transaction of the session.
+    repl::checkTxnTable(_opCtx.get(),
+                        lsid,
+                        txnNum2,
+                        commitOp2.getOpTime(),
+                        *commitOp2.getWallClockTime(),
+                        boost::none,
+                        DurableTxnStateEnum::kCommitted);
+    ASSERT_TRUE(docExists(_opCtx.get(), nss, doc));
+    ASSERT_TRUE(docExists(_opCtx.get(), nss, doc2));
+}
+
+TEST_F(IdempotencyTestTxns, CommitAndAbortTwoTransactionsWithPartialTxnOpsAtOnce) {
+    createCollectionWithUuid(_opCtx.get(), NamespaceString::kSessionTransactionsTableNamespace);
+    auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
+    auto lsid = makeLogicalSessionId(_opCtx.get());
+    TxnNumber txnNum1(1);
+    TxnNumber txnNum2(2);
+
+    auto partialOp1 = partialTxn(
+        lsid, txnNum1, StmtId(0), OpTime(), BSON_ARRAY(makeInsertApplyOpsEntry(nss, uuid, doc)));
+    auto abortOp1 = abortPrepared(lsid, txnNum1, StmtId(1), partialOp1.getOpTime());
+
+    // The second transaction (with a different transaction number) in the same session.
+    auto partialOp2 = partialTxn(
+        lsid, txnNum2, StmtId(0), OpTime(), BSON_ARRAY(makeInsertApplyOpsEntry(nss, uuid, doc2)));
+    auto commitOp2 =
+        commitUnprepared(lsid, txnNum2, StmtId(1), BSONArray(), partialOp2.getOpTime());
+
+    ASSERT_OK(ReplicationCoordinator::get(getGlobalServiceContext())
+                  ->setFollowerMode(MemberState::RS_RECOVERING));
+
+    // This also tests that we clear the partialTxnList for the session after applying the abort of
+    // the first transaction. Otherwise, saving operations from the second transaction to the same
+    // partialTxnList as the first transaction will trigger an invariant because of the mismatching
+    // transaction numbers.
+    testOpsAreIdempotent({partialOp1, abortOp1, partialOp2, commitOp2});
+
+    // The transaction table should only contain the second transaction of the session.
+    repl::checkTxnTable(_opCtx.get(),
+                        lsid,
+                        txnNum2,
+                        commitOp2.getOpTime(),
+                        *commitOp2.getWallClockTime(),
+                        boost::none,
+                        DurableTxnStateEnum::kCommitted);
+    ASSERT_FALSE(docExists(_opCtx.get(), nss, doc));
+    ASSERT_TRUE(docExists(_opCtx.get(), nss, doc2));
+}
+
 TEST_F(IdempotencyTestTxns, CommitUnpreparedTransactionWithPartialTxnOpsAndDataPartiallyApplied) {
     createCollectionWithUuid(_opCtx.get(), NamespaceString::kSessionTransactionsTableNamespace);
     auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
@@ -3471,6 +3548,45 @@ TEST_F(IdempotencyTestTxns, CommitPreparedTransactionWithPartialTxnOps) {
     ASSERT_TRUE(docExists(_opCtx.get(), nss, doc2));
 }
 
+TEST_F(IdempotencyTestTxns, CommitTwoPreparedTransactionsWithPartialTxnOpsAtOnce) {
+    createCollectionWithUuid(_opCtx.get(), NamespaceString::kSessionTransactionsTableNamespace);
+    auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
+    auto lsid = makeLogicalSessionId(_opCtx.get());
+    TxnNumber txnNum1(1);
+    TxnNumber txnNum2(2);
+
+    auto partialOp1 = partialTxn(
+        lsid, txnNum1, StmtId(0), OpTime(), BSON_ARRAY(makeInsertApplyOpsEntry(nss, uuid, doc)));
+    auto prepareOp1 = prepare(lsid, txnNum1, StmtId(1), BSONArray(), partialOp1.getOpTime());
+    auto commitOp1 = commitPrepared(lsid, txnNum1, StmtId(2), prepareOp1.getOpTime());
+
+    // The second transaction (with a different transaction number) in the same session.
+    auto partialOp2 = partialTxn(
+        lsid, txnNum2, StmtId(0), OpTime(), BSON_ARRAY(makeInsertApplyOpsEntry(nss, uuid, doc2)));
+    auto prepareOp2 = prepare(lsid, txnNum2, StmtId(1), BSONArray(), partialOp2.getOpTime());
+    auto commitOp2 = commitPrepared(lsid, txnNum2, StmtId(2), prepareOp2.getOpTime());
+
+    ASSERT_OK(ReplicationCoordinator::get(getGlobalServiceContext())
+                  ->setFollowerMode(MemberState::RS_RECOVERING));
+
+    // This also tests that we clear the partialTxnList for the session after applying the commit of
+    // the first prepared transaction. Otherwise, saving operations from the second transaction to
+    // the same partialTxnList as the first transaction will trigger an invariant because of the
+    // mismatching transaction numbers.
+    testOpsAreIdempotent({partialOp1, prepareOp1, commitOp1, partialOp2, prepareOp2, commitOp2});
+
+    // The transaction table should only contain the second transaction of the session.
+    repl::checkTxnTable(_opCtx.get(),
+                        lsid,
+                        txnNum2,
+                        commitOp2.getOpTime(),
+                        *commitOp2.getWallClockTime(),
+                        boost::none,
+                        DurableTxnStateEnum::kCommitted);
+    ASSERT_TRUE(docExists(_opCtx.get(), nss, doc));
+    ASSERT_TRUE(docExists(_opCtx.get(), nss, doc2));
+}
+
 TEST_F(IdempotencyTestTxns, CommitPreparedTransactionWithPartialTxnOpsAndDataPartiallyApplied) {
     createCollectionWithUuid(_opCtx.get(), NamespaceString::kSessionTransactionsTableNamespace);
     auto uuid = createCollectionWithUuid(_opCtx.get(), nss);