+++ b/jstests/replsets/stable_timestamp_oplog_hole.js @@ -0,0 +1,68 @@ +/** + * Test that stable timestamp can advance after a oplog hole is released via an abort. + * + * @tags: [requires_persistence, uses_transactions] + */ + +(function() { +"use strict"; +load("jstests/libs/check_log.js"); +load("jstests/core/txns/libs/prepare_helpers.js"); + +const replTest = new ReplSetTest({nodes: 1}); +replTest.startSet(); +replTest.initiate(); + +let primary = replTest.getPrimary(); + +const dbName = "test"; +const collName = "stable_timestamp_oplog_hole"; +let testDB = primary.getDB(dbName); +const testColl = testDB.getCollection(collName); + +assert.commandWorked(testColl.insert({_id: 1})); + +assert.commandWorked(testDB.adminCommand( + {configureFailPoint: 'hangAndFailUnpreparedCommitAfterReservingOplogSlot', mode: 'alwaysOn'})); + +// Run a transaction in a parallel shell. +function transactionFn() { + const name = 'stable_timestamp_oplog_hole'; + const dbName = 'test'; + const collName = name; + const session = db.getMongo().startSession(); + const sessionDB = session.getDatabase(dbName); + + session.startTransaction(); + sessionDB[collName].update({_id: 1}, {_id: 1, a: 1}); + assert.commandFailedWithCode(session.commitTransaction_forTesting(), 51260); +} +const joinTransaction = startParallelShell(transactionFn, replTest.ports[0]); + +jsTestLog("Waiting to hang unprepared commit after reserving oplog slot."); +checkLog.contains(primary, "hangAndFailUnpreparedCommitAfterReservingOplogSlot fail point enabled"); + +// Run a write with {w: "majority"} in a parallel shell. +function majorityWriteFn() { + const dbName = 'test'; + const collName = 'stable_timestamp_oplog_hole'; + const testDB = db.getMongo().getDB(dbName); + const testColl = testDB.getCollection(collName); + + assert.commandWorked(testColl.insert({_id: 2}, {writeConcern: {w: "majority", wtimeout: 20000}})); +} +const joinMajorityWrite = startParallelShell(majorityWriteFn, replTest.ports[0]); + +jsTestLog("Waiting for majority write to advance lastApplied."); +sleep(5 * 1000); +assert.commandWorked(testDB.adminCommand( + {configureFailPoint: 'hangAndFailUnpreparedCommitAfterReservingOplogSlot', mode: 'off'})); + +jsTestLog("Joining the transaction."); +joinTransaction(); + +jsTestLog("Joining the majority write."); +joinMajorityWrite(); + +replTest.stopSet(); +}()); diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp index 73d31fd799..e4270bd907 100644 --- a/src/mongo/db/op_observer_impl.cpp +++ b/src/mongo/db/op_observer_impl.cpp @@ -27,6 +27,8 @@ * it in the license file. */ +#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplication + #include "mongo/platform/basic.h" #include "mongo/db/op_observer_impl.h" @@ -62,6 +64,7 @@ #include "mongo/scripting/engine.h" #include "mongo/util/assert_util.h" #include "mongo/util/fail_point.h" +#include "mongo/util/log.h" namespace mongo { using repl::MutableOplogEntry; @@ -70,6 +73,7 @@ using repl::OplogEntry; namespace { MONGO_FAIL_POINT_DEFINE(failCollectionUpdates); +MONGO_FAIL_POINT_DEFINE(hangAndFailUnpreparedCommitAfterReservingOplogSlot); const auto documentKeyDecoration = OperationContext::declareDecoration<BSONObj>(); @@ -1009,6 +1013,14 @@ void OpObserverImpl::onUnpreparedTransactionCommit( auto oplogSlots = repl::getNextOpTimes(opCtx, statements.size()); invariant(oplogSlots.size() == statements.size()); + if (MONGO_unlikely(hangAndFailUnpreparedCommitAfterReservingOplogSlot.shouldFail())) { + log() << "hangAndFailUnpreparedCommitAfterReservingOplogSlot fail point enabled."; + hangAndFailUnpreparedCommitAfterReservingOplogSlot.pauseWhileSet(opCtx); + uasserted(51260, + str::stream() + << "hangAndFailUnpreparedCommitAfterReservingOplogSlot fail point enabled"); + } + // Log in-progress entries for the transaction along with the implicit commit. int numOplogEntries = logOplogEntriesForTransaction(opCtx, statements, oplogSlots, false); commitOpTime = oplogSlots[numOplogEntries - 1];