diff options
Diffstat (limited to 'src')
-rw-r--r-- | src/mongo/db/catalog/collection_impl.cpp | 4 | ||||
-rw-r--r-- | src/mongo/db/op_observer.h | 1 | ||||
-rw-r--r-- | src/mongo/db/op_observer_impl.cpp | 22 | ||||
-rw-r--r-- | src/mongo/db/op_observer_impl_test.cpp | 50 | ||||
-rw-r--r-- | src/mongo/db/repl/oplog.cpp | 32 | ||||
-rw-r--r-- | src/mongo/db/repl/oplog.h | 9 | ||||
-rw-r--r-- | src/mongo/db/transaction_participant.cpp | 18 | ||||
-rw-r--r-- | src/mongo/db/transaction_participant.h | 4 |
8 files changed, 65 insertions, 75 deletions
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp index 27e5de9aacc..fa0f897d75f 100644 --- a/src/mongo/db/catalog/collection_impl.cpp +++ b/src/mongo/db/catalog/collection_impl.cpp @@ -524,7 +524,7 @@ Status CollectionImpl::_insertDocuments(OperationContext* opCtx, for (auto it = begin; it != end; it++) { records.emplace_back(Record{RecordId(), RecordData(it->doc.objdata(), it->doc.objsize())}); - timestamps.emplace_back(it->oplogSlot.opTime.getTimestamp()); + timestamps.emplace_back(it->oplogSlot.getTimestamp()); } Status status = _recordStore->insertRecords(opCtx, &records, timestamps); if (!status.isOK()) @@ -538,7 +538,7 @@ Status CollectionImpl::_insertDocuments(OperationContext* opCtx, invariant(RecordId::min() < loc); invariant(loc < RecordId::max()); - BsonRecord bsonRecord = {loc, Timestamp(it->oplogSlot.opTime.getTimestamp()), &(it->doc)}; + BsonRecord bsonRecord = {loc, Timestamp(it->oplogSlot.getTimestamp()), &(it->doc)}; bsonRecords.push_back(bsonRecord); } diff --git a/src/mongo/db/op_observer.h b/src/mongo/db/op_observer.h index 729908a287e..85847e698ef 100644 --- a/src/mongo/db/op_observer.h +++ b/src/mongo/db/op_observer.h @@ -42,7 +42,6 @@ namespace mongo { struct InsertStatement; class OperationContext; -struct OplogSlot; namespace repl { class OpTime; diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp index 460bb3fd792..e00a44f1ac4 100644 --- a/src/mongo/db/op_observer_impl.cpp +++ b/src/mongo/db/op_observer_impl.cpp @@ -1034,7 +1034,7 @@ OpTimeBundle logApplyOpsForTransaction(OperationContext* opCtx, try { // We are only given an oplog slot for prepared transactions. - auto prepare = !prepareOplogSlot.opTime.isNull(); + auto prepare = !prepareOplogSlot.isNull(); if (prepare) { // TODO: SERVER-36814 Remove "prepare" field on applyOps. applyOpsBuilder.append("prepare", true); @@ -1129,7 +1129,7 @@ void logOplogEntriesForTransaction(OperationContext* opCtx, stmtId = 0; const NamespaceString cmdNss{"admin", "$cmd"}; auto oplogSlot = oplogSlots.begin(); - const auto startOpTime = oplogSlot->opTime; + const auto startOpTime = *oplogSlot; for (const auto& stmt : stmts) { bool isStartOfTxn = prevWriteOpTime.writeOpTime.isNull(); prevWriteOpTime = logReplOperationForTransaction( @@ -1200,7 +1200,7 @@ void logCommitOrAbortForPreparedTransaction(OperationContext* opCtx, false /* prepare */, false /* inTxn */, oplogSlot); - invariant(oplogSlot.opTime.isNull() || oplogSlot.opTime == oplogOpTime); + invariant(oplogSlot.isNull() || oplogSlot == oplogOpTime); onWriteOpCompleted(opCtx, cmdNss, @@ -1248,7 +1248,7 @@ repl::OpTime logCommitForUnpreparedTransaction(OperationContext* opCtx, oplogSlot); // In the present implementation, a reserved oplog slot is always provided. However that // is not enforced at this level. - invariant(oplogSlot.opTime.isNull() || oplogSlot.opTime == oplogOpTime); + invariant(oplogSlot.isNull() || oplogSlot == oplogOpTime); onWriteOpCompleted(opCtx, cmdNss, @@ -1258,7 +1258,7 @@ repl::OpTime logCommitForUnpreparedTransaction(OperationContext* opCtx, DurableTxnStateEnum::kCommitted, boost::none /* startOpTime */); - return oplogSlot.opTime; + return oplogSlot; } } // namespace @@ -1289,7 +1289,7 @@ void OpObserverImpl::onUnpreparedTransactionCommit( logOplogEntriesForTransaction(opCtx, statements, oplogSlots); commitOpTime = logCommitForUnpreparedTransaction(opCtx, statements.size() /* stmtId */, - oplogSlots.back().opTime, + oplogSlots.back(), commitSlot, statements.size()); } @@ -1349,7 +1349,7 @@ repl::OpTime logPrepareTransaction(OperationContext* opCtx, false /* prepare */, false /* inTxn */, oplogSlot); - invariant(oplogSlot.opTime == oplogOpTime); + invariant(oplogSlot == oplogOpTime); onWriteOpCompleted(opCtx, cmdNss, @@ -1359,7 +1359,7 @@ repl::OpTime logPrepareTransaction(OperationContext* opCtx, DurableTxnStateEnum::kPrepared, startOpTime /* startOpTime */); - return oplogSlot.opTime; + return oplogSlot; } void OpObserverImpl::onTransactionPrepare(OperationContext* opCtx, @@ -1368,7 +1368,7 @@ void OpObserverImpl::onTransactionPrepare(OperationContext* opCtx, invariant(!reservedSlots.empty()); const auto prepareOpTime = reservedSlots.back(); invariant(opCtx->getTxnNumber()); - invariant(!prepareOpTime.opTime.isNull()); + invariant(!prepareOpTime.isNull()); // Don't write oplog entry on secondaries. if (!opCtx->writesAreReplicated()) { @@ -1413,10 +1413,10 @@ void OpObserverImpl::onTransactionPrepare(OperationContext* opCtx, logOplogEntriesForTransaction(opCtx, statements, reservedSlots); // The prevOpTime is the OpTime of the second last entry in the reserved slots. - prevOpTime = reservedSlots.rbegin()[1].opTime; + prevOpTime = reservedSlots.rbegin()[1]; } auto startTxnSlot = reservedSlots.front(); - const auto startOpTime = startTxnSlot.opTime; + const auto startOpTime = startTxnSlot; logPrepareTransaction( opCtx, statements.size() /* stmtId */, prevOpTime, prepareOpTime, startOpTime); wuow.commit(); diff --git a/src/mongo/db/op_observer_impl_test.cpp b/src/mongo/db/op_observer_impl_test.cpp index bca0c067b88..979e1f2169e 100644 --- a/src/mongo/db/op_observer_impl_test.cpp +++ b/src/mongo/db/op_observer_impl_test.cpp @@ -763,8 +763,8 @@ TEST_F(OpObserverTransactionTest, TransactionalPrepareTest) { { WriteUnitOfWork wuow(opCtx()); OplogSlot slot = repl::getNextOpTime(opCtx()); - txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime); - opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp()); + txnParticipant.transitionToPreparedforTest(opCtx(), slot); + opCtx()->recoveryUnit()->setPrepareTimestamp(slot.getTimestamp()); opObserver().onTransactionPrepare( opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); } @@ -838,8 +838,8 @@ TEST_F(OpObserverTransactionTest, TransactionalPreparedCommitTest) { opObserver().onInserts(opCtx(), nss, uuid, insert.begin(), insert.end(), false); const auto prepareSlot = repl::getNextOpTime(opCtx()); - txnParticipant.transitionToPreparedforTest(opCtx(), prepareSlot.opTime); - prepareTimestamp = prepareSlot.opTime.getTimestamp(); + txnParticipant.transitionToPreparedforTest(opCtx(), prepareSlot); + prepareTimestamp = prepareSlot.getTimestamp(); opObserver().onTransactionPrepare( opCtx(), {prepareSlot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); @@ -910,7 +910,7 @@ TEST_F(OpObserverTransactionTest, TransactionalPreparedAbortTest) { opObserver().onInserts(opCtx(), nss, uuid, insert.begin(), insert.end(), false); const auto prepareSlot = repl::getNextOpTime(opCtx()); - txnParticipant.transitionToPreparedforTest(opCtx(), prepareSlot.opTime); + txnParticipant.transitionToPreparedforTest(opCtx(), prepareSlot); opObserver().onTransactionPrepare( opCtx(), {prepareSlot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); abortSlot = repl::getNextOpTime(opCtx()); @@ -989,8 +989,8 @@ TEST_F(OpObserverTransactionTest, PreparingEmptyTransactionLogsEmptyApplyOps) { { WriteUnitOfWork wuow(opCtx()); OplogSlot slot = repl::getNextOpTime(opCtx()); - txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime); - opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp()); + txnParticipant.transitionToPreparedforTest(opCtx(), slot); + opCtx()->recoveryUnit()->setPrepareTimestamp(slot.getTimestamp()); opObserver().onTransactionPrepare( opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); } @@ -1014,11 +1014,11 @@ TEST_F(OpObserverTransactionTest, PreparingTransactionWritesToTransactionTable) { WriteUnitOfWork wuow(opCtx()); OplogSlot slot = repl::getNextOpTime(opCtx()); - txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime); - prepareOpTime = slot.opTime; + txnParticipant.transitionToPreparedforTest(opCtx(), slot); + prepareOpTime = slot; opObserver().onTransactionPrepare( opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); - opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp()); + opCtx()->recoveryUnit()->setPrepareTimestamp(slot.getTimestamp()); } ASSERT_EQ(prepareOpTime.getTimestamp(), opCtx()->recoveryUnit()->getPrepareTimestamp()); @@ -1048,10 +1048,10 @@ TEST_F(OpObserverTransactionTest, AbortingPreparedTransactionWritesToTransaction { WriteUnitOfWork wuow(opCtx()); OplogSlot slot = repl::getNextOpTime(opCtx()); - opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp()); + opCtx()->recoveryUnit()->setPrepareTimestamp(slot.getTimestamp()); opObserver().onTransactionPrepare( opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); - txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime); + txnParticipant.transitionToPreparedforTest(opCtx(), slot); abortSlot = repl::getNextOpTime(opCtx()); } @@ -1116,15 +1116,15 @@ TEST_F(OpObserverTransactionTest, CommittingPreparedTransactionWritesToTransacti { WriteUnitOfWork wuow(opCtx()); OplogSlot slot = repl::getNextOpTime(opCtx()); - prepareOpTime = slot.opTime; - opCtx()->recoveryUnit()->setPrepareTimestamp(slot.opTime.getTimestamp()); + prepareOpTime = slot; + opCtx()->recoveryUnit()->setPrepareTimestamp(slot.getTimestamp()); opObserver().onTransactionPrepare( opCtx(), {slot}, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); - txnParticipant.transitionToPreparedforTest(opCtx(), slot.opTime); + txnParticipant.transitionToPreparedforTest(opCtx(), slot); } OplogSlot commitSlot = repl::getNextOpTime(opCtx()); - repl::OpTime commitOpTime = commitSlot.opTime; + repl::OpTime commitOpTime = commitSlot; ASSERT_LTE(prepareOpTime, commitOpTime); // Mimic committing the transaction. @@ -1578,7 +1578,7 @@ TEST_F(OpObserverMultiEntryTransactionTest, txnParticipant.unstashTransactionResources(opCtx(), "prepareTransaction"); repl::OpTime prepareOpTime; auto reservedSlots = repl::getNextOpTimes(opCtx(), 1); - prepareOpTime = reservedSlots.back().opTime; + prepareOpTime = reservedSlots.back(); opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp()); opObserver().onTransactionPrepare( opCtx(), reservedSlots, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); @@ -1623,7 +1623,7 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalInsertPrepareTest) { repl::OpTime prepareOpTime; auto reservedSlots = repl::getNextOpTimes(opCtx(), 5); - prepareOpTime = reservedSlots.back().opTime; + prepareOpTime = reservedSlots.back(); txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime); opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp()); opObserver().onTransactionPrepare( @@ -1715,7 +1715,7 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalUpdatePrepareTest) { repl::OpTime prepareOpTime; auto reservedSlots = repl::getNextOpTimes(opCtx(), 3); - prepareOpTime = reservedSlots.back().opTime; + prepareOpTime = reservedSlots.back(); txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime); opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp()); opObserver().onTransactionPrepare( @@ -1796,9 +1796,9 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalDeletePrepareTest) { repl::OpTime prepareOpTime; auto reservedSlots = repl::getNextOpTimes(opCtx(), 3); - prepareOpTime = reservedSlots.back().opTime; + prepareOpTime = reservedSlots.back(); txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime); - prepareOpTime = reservedSlots.back().opTime; + prepareOpTime = reservedSlots.back(); opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp()); opObserver().onTransactionPrepare( opCtx(), reservedSlots, txnParticipant.retrieveCompletedTransactionOperations(opCtx())); @@ -1864,7 +1864,7 @@ TEST_F(OpObserverMultiEntryTransactionTest, CommitPreparedTest) { repl::OpTime prepareOpTime; auto reservedSlots = repl::getNextOpTimes(opCtx(), 2); - prepareOpTime = reservedSlots.back().opTime; + prepareOpTime = reservedSlots.back(); txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime); opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp()); @@ -1918,7 +1918,7 @@ TEST_F(OpObserverMultiEntryTransactionTest, CommitPreparedTest) { "commitTransaction" << 1 << "commitTimestamp" << commitTimestamp << "prepared" << true); ASSERT_BSONOBJ_EQ(oExpected, o); - assertTxnRecord(txnNum(), commitSlot.opTime, DurableTxnStateEnum::kCommitted); + assertTxnRecord(txnNum(), commitSlot, DurableTxnStateEnum::kCommitted); // startTimestamp should no longer be set once the transaction has been committed. assertTxnRecordStartOpTime(boost::none); } @@ -1940,7 +1940,7 @@ TEST_F(OpObserverMultiEntryTransactionTest, AbortPreparedTest) { repl::OpTime prepareOpTime; auto reservedSlots = repl::getNextOpTimes(opCtx(), 2); - prepareOpTime = reservedSlots.back().opTime; + prepareOpTime = reservedSlots.back(); txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime); opCtx()->recoveryUnit()->setPrepareTimestamp(prepareOpTime.getTimestamp()); @@ -1984,7 +1984,7 @@ TEST_F(OpObserverMultiEntryTransactionTest, AbortPreparedTest) { auto oExpected = BSON("abortTransaction" << 1); ASSERT_BSONOBJ_EQ(oExpected, o); - assertTxnRecord(txnNum(), abortSlot.opTime, DurableTxnStateEnum::kAborted); + assertTxnRecord(txnNum(), abortSlot, DurableTxnStateEnum::kAborted); // startOpTime should no longer be set once a transaction has been aborted. assertTxnRecordStartOpTime(boost::none); } diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp index f421603e99f..129f7ab16db 100644 --- a/src/mongo/db/repl/oplog.cpp +++ b/src/mongo/db/repl/oplog.cpp @@ -137,9 +137,6 @@ struct LocalOplogInfo { // Synchronizes the section where a new Timestamp is generated and when it is registered in the // storage engine. stdx::mutex newOpMutex; - - // Used to generate "h" fields in pv0. Synchronized by newOpMutex. - PseudoRandom hashGenerator{std::unique_ptr<SecureRandom>(SecureRandom::create())->nextInt64()}; }; const auto localOplogInfo = ServiceContext::declareDecoration<LocalOplogInfo>(); @@ -185,8 +182,7 @@ void _getNextOpTimes(OperationContext* opCtx, } for (std::size_t i = 0; i < count; i++) { - slotsOut[i].opTime = {Timestamp(ts.asULL() + i), term}; - slotsOut[i].hash = oplogInfo.hashGenerator.nextInt64(); + slotsOut[i] = {Timestamp(ts.asULL() + i), term}; } } @@ -449,7 +445,6 @@ OplogDocWriter _logOpWriter(OperationContext* opCtx, const BSONObj* o2, bool fromMigrate, OpTime optime, - long long hashNew, Date_t wallTime, const OperationSessionInfo& sessionInfo, StmtId statementId, @@ -461,7 +456,10 @@ OplogDocWriter _logOpWriter(OperationContext* opCtx, b.append("ts", optime.getTimestamp()); if (optime.getTerm() != -1) b.append("t", optime.getTerm()); - b.append("h", hashNew); + + // Always write zero hash instead of using FCV to gate this for retryable writes + // and change stream, who expect to be able to read oplog across FCV's. + b.append("h", 0LL); b.append("v", OplogEntry::kOplogVersion); b.append("op", opstr); b.append("ns", nss.ns()); @@ -608,7 +606,7 @@ OpTime logOp(OperationContext* opCtx, auto const oplog = oplogInfo.oplog; OplogSlot slot; WriteUnitOfWork wuow(opCtx); - if (oplogSlot.opTime.isNull()) { + if (oplogSlot.isNull()) { _getNextOpTimes(opCtx, oplog, 1, &slot); } else { slot = oplogSlot; @@ -621,8 +619,7 @@ OpTime logOp(OperationContext* opCtx, obj, o2, fromMigrate, - slot.opTime, - slot.hash, + slot, wallClockTime, sessionInfo, statementId, @@ -630,10 +627,10 @@ OpTime logOp(OperationContext* opCtx, prepare, inTxn); const DocWriter* basePtr = &writer; - auto timestamp = slot.opTime.getTimestamp(); - _logOpsInner(opCtx, nss, &basePtr, ×tamp, 1, oplog, slot.opTime, wallClockTime); + auto timestamp = slot.getTimestamp(); + _logOpsInner(opCtx, nss, &basePtr, ×tamp, 1, oplog, slot, wallClockTime); wuow.commit(); - return slot.opTime; + return slot; } std::vector<OpTime> logInsertOps(OperationContext* opCtx, @@ -686,7 +683,7 @@ std::vector<OpTime> logInsertOps(OperationContext* opCtx, // Make a mutable copy. auto insertStatementOplogSlot = begin[i].oplogSlot; // Fetch optime now, if not already fetched. - if (insertStatementOplogSlot.opTime.isNull()) { + if (insertStatementOplogSlot.isNull()) { _getNextOpTimes(opCtx, oplog, 1, &insertStatementOplogSlot); } // Only 'applyOps' oplog entries can be prepared. @@ -698,17 +695,16 @@ std::vector<OpTime> logInsertOps(OperationContext* opCtx, begin[i].doc, NULL, fromMigrate, - insertStatementOplogSlot.opTime, - insertStatementOplogSlot.hash, + insertStatementOplogSlot, wallClockTime, sessionInfo, begin[i].stmtId, oplogLink, prepare, false /* inTxn */)); - oplogLink.prevOpTime = insertStatementOplogSlot.opTime; + oplogLink.prevOpTime = insertStatementOplogSlot; timestamps[i] = oplogLink.prevOpTime.getTimestamp(); - opTimes.push_back(insertStatementOplogSlot.opTime); + opTimes.push_back(insertStatementOplogSlot); } MONGO_FAIL_POINT_BLOCK(sleepBetweenInsertOpTimeGenerationAndLogOp, customWait) { diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h index 694d5c38152..d9fa2c8bf74 100644 --- a/src/mongo/db/repl/oplog.h +++ b/src/mongo/db/repl/oplog.h @@ -51,12 +51,7 @@ class OperationContext; class OperationSessionInfo; class Session; -struct OplogSlot { - OplogSlot() {} - OplogSlot(repl::OpTime opTime, std::int64_t hash) : opTime(opTime), hash(hash) {} - repl::OpTime opTime; - std::int64_t hash = 0; -}; +using OplogSlot = repl::OpTime; struct InsertStatement { public: @@ -67,7 +62,7 @@ public: InsertStatement(StmtId statementId, BSONObj toInsert, OplogSlot os) : stmtId(statementId), oplogSlot(os), doc(toInsert) {} InsertStatement(BSONObj toInsert, Timestamp ts, long long term) - : oplogSlot(repl::OpTime(ts, term), 0), doc(toInsert) {} + : oplogSlot(repl::OpTime(ts, term)), doc(toInsert) {} StmtId stmtId = kUninitializedStmtId; OplogSlot oplogSlot; diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp index f3620219982..59aa73e47c7 100644 --- a/src/mongo/db/transaction_participant.cpp +++ b/src/mongo/db/transaction_participant.cpp @@ -1017,7 +1017,7 @@ Timestamp TransactionParticipant::Participant::prepareTransaction( std::vector<OplogSlot> reservedSlots; if (prepareOptime) { // On secondary, we just prepare the transaction and discard the buffered ops. - prepareOplogSlot = OplogSlot(*prepareOptime, 0); + prepareOplogSlot = OplogSlot(*prepareOptime); stdx::lock_guard<Client> lk(*opCtx->getClient()); o(lk).prepareOpTime = *prepareOptime; reservedSlots.push_back(prepareOplogSlot); @@ -1045,18 +1045,18 @@ Timestamp TransactionParticipant::Participant::prepareTransaction( { stdx::lock_guard<Client> lk(*opCtx->getClient()); - o(lk).prepareOpTime = prepareOplogSlot.opTime; + o(lk).prepareOpTime = prepareOplogSlot; } if (MONGO_FAIL_POINT(hangAfterReservingPrepareTimestamp)) { // This log output is used in js tests so please leave it. log() << "transaction - hangAfterReservingPrepareTimestamp fail point " "enabled. Blocking until fail point is disabled. Prepare OpTime: " - << prepareOplogSlot.opTime; + << prepareOplogSlot; MONGO_FAIL_POINT_PAUSE_WHILE_SET(hangAfterReservingPrepareTimestamp); } } - opCtx->recoveryUnit()->setPrepareTimestamp(prepareOplogSlot.opTime.getTimestamp()); + opCtx->recoveryUnit()->setPrepareTimestamp(prepareOplogSlot.getTimestamp()); opCtx->getWriteUnitOfWork()->prepare(); opCtx->getServiceContext()->getOpObserver()->onTransactionPrepare( opCtx, reservedSlots, completedTransactionOperations); @@ -1068,7 +1068,7 @@ Timestamp TransactionParticipant::Participant::prepareTransaction( << "been set to: " << p().oldestOplogEntryOpTime->toString()); // Keep track of the OpTime from the first oplog entry written by this transaction. - p().oldestOplogEntryOpTime = prepareOplogSlot.opTime; + p().oldestOplogEntryOpTime = prepareOplogSlot; // Maintain the OpTime of the oldest active oplog entry for this transaction. We currently // only write an oplog entry for an in progress transaction when it is in the prepare state @@ -1092,7 +1092,7 @@ Timestamp TransactionParticipant::Participant::prepareTransaction( const bool unlocked = opCtx->lockState()->unlockRSTLforPrepare(); invariant(unlocked); - return prepareOplogSlot.opTime.getTimestamp(); + return prepareOplogSlot.getTimestamp(); } void TransactionParticipant::Participant::addTransactionOperation( @@ -1245,12 +1245,12 @@ void TransactionParticipant::Participant::commitPreparedTransaction( invariant(!commitOplogEntryOpTime); oplogSlotReserver.emplace(opCtx); commitOplogSlot = oplogSlotReserver->getLastSlot(); - invariant(commitOplogSlot.opTime.getTimestamp() >= commitTimestamp, + invariant(commitOplogSlot.getTimestamp() >= commitTimestamp, str::stream() << "Commit oplog entry must be greater than or equal to commit " "timestamp due to causal consistency. commit timestamp: " << commitTimestamp.toBSON() << ", commit oplog entry optime: " - << commitOplogSlot.opTime.toBSON()); + << commitOplogSlot.toBSON()); } else { // We always expect a non-null commitOplogEntryOpTime to be passed in on secondaries // in order to set the finishOpTime. @@ -1260,7 +1260,7 @@ void TransactionParticipant::Participant::commitPreparedTransaction( // If commitOplogEntryOpTime is a nullopt, then we grab the OpTime from the commitOplogSlot // which will only be set if we are primary. Otherwise, the commitOplogEntryOpTime must have // been passed in during secondary oplog application. - auto commitOplogSlotOpTime = commitOplogEntryOpTime.value_or(commitOplogSlot.opTime); + auto commitOplogSlotOpTime = commitOplogEntryOpTime.value_or(commitOplogSlot); opCtx->recoveryUnit()->setDurableTimestamp(commitOplogSlotOpTime.getTimestamp()); _commitStorageTransaction(opCtx); diff --git a/src/mongo/db/transaction_participant.h b/src/mongo/db/transaction_participant.h index 3cf7735cc94..0c16db50160 100644 --- a/src/mongo/db/transaction_participant.h +++ b/src/mongo/db/transaction_participant.h @@ -822,13 +822,13 @@ private: */ OplogSlot getLastSlot() { invariant(!_oplogSlots.empty()); - invariant(!_oplogSlots.back().opTime.isNull()); + invariant(!_oplogSlots.back().isNull()); return getSlots().back(); } std::vector<OplogSlot>& getSlots() { invariant(!_oplogSlots.empty()); - invariant(!_oplogSlots.back().opTime.isNull()); + invariant(!_oplogSlots.back().isNull()); return _oplogSlots; } |