diff options
-rw-r--r-- | src/mongo/db/repl/bgsync.cpp | 39 | ||||
-rw-r--r-- | src/mongo/db/repl/bgsync.h | 10 | ||||
-rw-r--r-- | src/mongo/db/repl/rs_rollback.cpp | 25 | ||||
-rw-r--r-- | src/mongo/db/repl/rs_rollback.h | 5 | ||||
-rw-r--r-- | src/mongo/db/repl/rs_rollback_test.cpp | 89 |
5 files changed, 103 insertions, 65 deletions
diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp index 186903d0924..cd03141f6ab 100644 --- a/src/mongo/db/repl/bgsync.cpp +++ b/src/mongo/db/repl/bgsync.cpp @@ -266,12 +266,13 @@ void BackgroundSync::_produce(OperationContext* opCtx) { } } + auto storageInterface = StorageInterface::get(opCtx); // find a target to sync from the last optime fetched OpTime lastOpTimeFetched; HostAndPort source; SyncSourceResolverResponse syncSourceResp; { - const OpTime minValidSaved = StorageInterface::get(opCtx)->getMinValid(opCtx); + const OpTime minValidSaved = storageInterface->getMinValid(opCtx); stdx::lock_guard<stdx::mutex> lock(_mutex); const auto requiredOpTime = (minValidSaved > _lastOpTimeFetched) ? minValidSaved : OpTime(); @@ -358,9 +359,8 @@ void BackgroundSync::_produce(OperationContext* opCtx) { // Set the applied point if unset. This is most likely the first time we've established a sync // source since stepping down or otherwise clearing the applied point. We need to set this here, // before the OplogWriter gets a chance to append to the oplog. - if (StorageInterface::get(opCtx)->getAppliedThrough(opCtx).isNull()) { - StorageInterface::get(opCtx)->setAppliedThrough(opCtx, - _replCoord->getMyLastAppliedOpTime()); + if (storageInterface->getAppliedThrough(opCtx).isNull()) { + storageInterface->setAppliedThrough(opCtx, _replCoord->getMyLastAppliedOpTime()); } // "lastFetched" not used. Already set in _enqueueDocuments. @@ -473,7 +473,11 @@ void BackgroundSync::_produce(OperationContext* opCtx) { } } - _rollback(opCtx, source, syncSourceResp.rbid, getConnection); + OplogInterfaceLocal localOplog(opCtx, rsOplogName); + RollbackSourceImpl rollbackSource(getConnection, source, rsOplogName); + _rollback( + opCtx, localOplog, rollbackSource, syncSourceResp.rbid, _replCoord, storageInterface); + // Reset the producer to clear the sync source and the last optime fetched. stop(true); startProducerIfStopped(); @@ -611,9 +615,11 @@ void BackgroundSync::consume(OperationContext* opCtx) { } void BackgroundSync::_rollback(OperationContext* opCtx, - const HostAndPort& source, + const OplogInterface& localOplog, + const RollbackSource& rollbackSource, boost::optional<int> requiredRBID, - stdx::function<DBClientBase*()> getConnection) { + ReplicationCoordinator* replCoord, + StorageInterface* storageInterface) { if (MONGO_FAIL_POINT(rollbackHangBeforeStart)) { // This log output is used in js tests so please leave it. log() << "rollback - rollbackHangBeforeStart fail point " @@ -637,19 +643,16 @@ void BackgroundSync::_rollback(OperationContext* opCtx, { log() << "rollback 0"; Lock::GlobalWrite globalWrite(opCtx->lockState()); - if (!_replCoord->setFollowerMode(MemberState::RS_ROLLBACK)) { - log() << "Cannot transition from " << _replCoord->getMemberState().toString() << " to " + if (!replCoord->setFollowerMode(MemberState::RS_ROLLBACK)) { + log() << "Cannot transition from " << replCoord->getMemberState().toString() << " to " << MemberState(MemberState::RS_ROLLBACK).toString(); return; } } try { - auto status = syncRollback(opCtx, - OplogInterfaceLocal(opCtx, rsOplogName), - RollbackSourceImpl(getConnection, source, rsOplogName), - requiredRBID, - _replCoord); + auto status = syncRollback( + opCtx, localOplog, rollbackSource, requiredRBID, replCoord, storageInterface); // Abort only when syncRollback detects we are in a unrecoverable state. // WARNING: these statuses sometimes have location codes which are lost with uassertStatusOK @@ -668,8 +671,8 @@ void BackgroundSync::_rollback(OperationContext* opCtx, invariant(ex.getCode() != ErrorCodes::UnrecoverableRollbackError); warning() << "rollback cannot complete at this time (retrying later): " << redact(ex) - << " appliedThrough=" << _replCoord->getMyLastAppliedOpTime() - << " minvalid=" << StorageInterface::get(opCtx)->getMinValid(opCtx); + << " appliedThrough=" << replCoord->getMyLastAppliedOpTime() + << " minvalid=" << storageInterface->getMinValid(opCtx); // Sleep a bit to allow upstream node to coalesce, if that was the cause of the failure. If // we failed in a way that will keep failing, but wasn't flagged as a fatal failure, this @@ -697,10 +700,10 @@ void BackgroundSync::_rollback(OperationContext* opCtx, fassertFailedNoTrace(40276); } - if (!_replCoord->setFollowerMode(MemberState::RS_RECOVERING)) { + if (!replCoord->setFollowerMode(MemberState::RS_RECOVERING)) { severe() << "Failed to transition into " << MemberState(MemberState::RS_RECOVERING) << "; expected to be in state " << MemberState(MemberState::RS_ROLLBACK) - << " but found self in " << _replCoord->getMemberState(); + << " but found self in " << replCoord->getMemberState(); fassertFailedNoTrace(40364); } } diff --git a/src/mongo/db/repl/bgsync.h b/src/mongo/db/repl/bgsync.h index 85069066cdd..f83d327e4e4 100644 --- a/src/mongo/db/repl/bgsync.h +++ b/src/mongo/db/repl/bgsync.h @@ -51,8 +51,11 @@ class OperationContext; namespace repl { +class OplogInterface; class ReplicationCoordinator; class ReplicationCoordinatorExternalState; +class RollbackSource; +class StorageInterface; class BackgroundSync { MONGO_DISALLOW_COPYING(BackgroundSync); @@ -163,12 +166,13 @@ private: /** * Executes a rollback. - * 'getConnection' returns a connection to the sync source. */ void _rollback(OperationContext* opCtx, - const HostAndPort& source, + const OplogInterface& localOplog, + const RollbackSource& rollbackSource, boost::optional<int> requiredRBID, - stdx::function<DBClientBase*()> getConnection); + ReplicationCoordinator* replCoord, + StorageInterface* storageInterface); // restart syncing void start(OperationContext* opCtx); diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp index 8859ea6fba7..db7cef394bd 100644 --- a/src/mongo/db/repl/rs_rollback.cpp +++ b/src/mongo/db/repl/rs_rollback.cpp @@ -337,7 +337,8 @@ namespace { */ void checkRbidAndUpdateMinValid(OperationContext* opCtx, const int rbid, - const RollbackSource& rollbackSource) { + const RollbackSource& rollbackSource, + StorageInterface* storageInterface) { // It is important that the steps are performed in order to avoid racing with upstream rollbacks // // 1) Get the last doc in their oplog. @@ -357,8 +358,8 @@ void checkRbidAndUpdateMinValid(OperationContext* opCtx, // online until we get to that point in freshness. OpTime minValid = fassertStatusOK(28774, OpTime::parseFromOplogEntry(newMinValidDoc)); log() << "Setting minvalid to " << minValid; - StorageInterface::get(opCtx)->setAppliedThrough(opCtx, {}); // Use top of oplog. - StorageInterface::get(opCtx)->setMinValid(opCtx, minValid); + storageInterface->setAppliedThrough(opCtx, {}); // Use top of oplog. + storageInterface->setMinValid(opCtx, minValid); if (MONGO_FAIL_POINT(rollbackHangThenFailAfterWritingMinValid)) { // This log output is used in js tests so please leave it. @@ -376,7 +377,8 @@ void checkRbidAndUpdateMinValid(OperationContext* opCtx, void syncFixUp(OperationContext* opCtx, const FixUpInfo& fixUpInfo, const RollbackSource& rollbackSource, - ReplicationCoordinator* replCoord) { + ReplicationCoordinator* replCoord, + StorageInterface* storageInterface) { // fetch all first so we needn't handle interruption in a fancy way unsigned long long totalSize = 0; @@ -415,7 +417,7 @@ void syncFixUp(OperationContext* opCtx, } log() << "rollback 3.5"; - checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource); + checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource, storageInterface); // update them log() << "rollback 4 n:" << goodVersions.size(); @@ -519,7 +521,7 @@ void syncFixUp(OperationContext* opCtx, // we did more reading from primary, so check it again for a rollback (which would mess // us up), and make minValid newer. log() << "rollback 4.2"; - checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource); + checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource, storageInterface); } log() << "rollback 4.6"; @@ -788,7 +790,8 @@ Status _syncRollback(OperationContext* opCtx, const OplogInterface& localOplog, const RollbackSource& rollbackSource, boost::optional<int> requiredRBID, - ReplicationCoordinator* replCoord) { + ReplicationCoordinator* replCoord, + StorageInterface* storageInterface) { invariant(!opCtx->lockState()->isLocked()); FixUpInfo how; @@ -834,7 +837,7 @@ Status _syncRollback(OperationContext* opCtx, log() << "rollback 3 fixup"; try { ON_BLOCK_EXIT([&] { replCoord->incrementRollbackID(); }); - syncFixUp(opCtx, how, rollbackSource, replCoord); + syncFixUp(opCtx, how, rollbackSource, replCoord, storageInterface); } catch (const RSFatalException& e) { return Status(ErrorCodes::UnrecoverableRollbackError, e.what(), 18753); } @@ -858,7 +861,8 @@ Status syncRollback(OperationContext* opCtx, const OplogInterface& localOplog, const RollbackSource& rollbackSource, boost::optional<int> requiredRBID, - ReplicationCoordinator* replCoord) { + ReplicationCoordinator* replCoord, + StorageInterface* storageInterface) { invariant(opCtx); invariant(replCoord); @@ -866,7 +870,8 @@ Status syncRollback(OperationContext* opCtx, DisableDocumentValidation validationDisabler(opCtx); UnreplicatedWritesBlock replicationDisabler(opCtx); - Status status = _syncRollback(opCtx, localOplog, rollbackSource, requiredRBID, replCoord); + Status status = + _syncRollback(opCtx, localOplog, rollbackSource, requiredRBID, replCoord, storageInterface); log() << "rollback finished" << rsLog; return status; diff --git a/src/mongo/db/repl/rs_rollback.h b/src/mongo/db/repl/rs_rollback.h index 58dd4a27d4f..c5589003ec4 100644 --- a/src/mongo/db/repl/rs_rollback.h +++ b/src/mongo/db/repl/rs_rollback.h @@ -44,6 +44,7 @@ namespace repl { class OplogInterface; class ReplicationCoordinator; class RollbackSource; +class StorageInterface; /** * Initiates the rollback process. @@ -65,6 +66,7 @@ class RollbackSource; * provides oplog; and * supports fetching documents and copying collections. * @param replCoord Used to track the rollback ID and to change the follower state + * @param storageInterface Used to update minValid. * * If requiredRBID is supplied, we error if the upstream node has a different RBID (ie it rolled * back) after fetching any information from it. @@ -77,7 +79,8 @@ Status syncRollback(OperationContext* opCtx, const OplogInterface& localOplog, const RollbackSource& rollbackSource, boost::optional<int> requiredRBID, - ReplicationCoordinator* replCoord); + ReplicationCoordinator* replCoord, + StorageInterface* storageInterface); /** * This namespace contains internal details of the rollback system. It is only exposed in a header diff --git a/src/mongo/db/repl/rs_rollback_test.cpp b/src/mongo/db/repl/rs_rollback_test.cpp index 22e7a677fea..2e8c199c277 100644 --- a/src/mongo/db/repl/rs_rollback_test.cpp +++ b/src/mongo/db/repl/rs_rollback_test.cpp @@ -134,6 +134,8 @@ protected: // Owned by service context ReplicationCoordinator* _coordinator; + repl::StorageInterfaceMock _storageInterface; + private: void setUp() override; void tearDown() override; @@ -147,11 +149,10 @@ void RSRollbackTest::setUp() { auto serviceContext = getServiceContext(); ReplicationCoordinator::set(serviceContext, std::unique_ptr<ReplicationCoordinator>(_coordinator)); - StorageInterface::set(serviceContext, stdx::make_unique<StorageInterfaceMock>()); setOplogCollectionName(); - repl::StorageInterface::get(_opCtx.get())->setAppliedThrough(_opCtx.get(), OpTime{}); - repl::StorageInterface::get(_opCtx.get())->setMinValid(_opCtx.get(), OpTime{}); + _storageInterface.setAppliedThrough(_opCtx.get(), OpTime{}); + _storageInterface.setMinValid(_opCtx.get(), OpTime{}); } void RSRollbackTest::tearDown() { @@ -161,16 +162,15 @@ void RSRollbackTest::tearDown() { } TEST_F(RSRollbackTest, InconsistentMinValid) { - repl::StorageInterface::get(_opCtx.get()) - ->setAppliedThrough(_opCtx.get(), OpTime(Timestamp(Seconds(0), 0), 0)); - repl::StorageInterface::get(_opCtx.get()) - ->setMinValid(_opCtx.get(), OpTime(Timestamp(Seconds(1), 0), 0)); + _storageInterface.setAppliedThrough(_opCtx.get(), OpTime(Timestamp(Seconds(0), 0), 0)); + _storageInterface.setMinValid(_opCtx.get(), OpTime(Timestamp(Seconds(1), 0), 0)); auto status = syncRollback(_opCtx.get(), OplogInterfaceMock(kEmptyMockOperations), RollbackSourceMock(std::unique_ptr<OplogInterface>( new OplogInterfaceMock(kEmptyMockOperations))), {}, - _coordinator); + _coordinator, + &_storageInterface); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18752, status.location()); } @@ -187,7 +187,8 @@ TEST_F(RSRollbackTest, OplogStartMissing) { operation, }))), {}, - _coordinator) + _coordinator, + &_storageInterface) .code()); } @@ -200,7 +201,8 @@ TEST_F(RSRollbackTest, NoRemoteOpLog) { RollbackSourceMock(std::unique_ptr<OplogInterface>( new OplogInterfaceMock(kEmptyMockOperations))), {}, - _coordinator); + _coordinator, + &_storageInterface); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18752, status.location()); } @@ -222,7 +224,8 @@ TEST_F(RSRollbackTest, RemoteGetRollbackIdThrows) { RollbackSourceLocal(std::unique_ptr<OplogInterface>( new OplogInterfaceMock(kEmptyMockOperations))), {}, - _coordinator), + _coordinator, + &_storageInterface), UserException, ErrorCodes::UnknownError); } @@ -245,7 +248,8 @@ TEST_F(RSRollbackTest, RemoteGetRollbackIdDiffersFromRequiredRBID) { RollbackSourceLocal(std::unique_ptr<OplogInterface>( new OplogInterfaceMock(kEmptyMockOperations))), {1}, - _coordinator), + _coordinator, + &_storageInterface), UserException, ErrorCodes::Error(40362)); } @@ -262,7 +266,8 @@ TEST_F(RSRollbackTest, BothOplogsAtCommonPoint) { operation, }))), {}, - _coordinator)); + _coordinator, + &_storageInterface)); } /** @@ -296,6 +301,7 @@ Collection* _createCollection(OperationContext* opCtx, */ int _testRollbackDelete(OperationContext* opCtx, ReplicationCoordinator* coordinator, + StorageInterface* storageInterface, const BSONObj& documentAtSource) { auto commonOperation = std::make_pair(BSON("ts" << Timestamp(Seconds(1), 0) << "h" << 1LL), RecordId(1)); @@ -330,7 +336,8 @@ int _testRollbackDelete(OperationContext* opCtx, OplogInterfaceMock({deleteOperation, commonOperation}), rollbackSource, {}, - coordinator)); + coordinator, + storageInterface)); ASSERT_TRUE(rollbackSource.called); Lock::DBLock dbLock(opCtx->lockState(), "test", MODE_S); @@ -346,14 +353,16 @@ int _testRollbackDelete(OperationContext* opCtx, TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionDoesNotExist) { createOplog(_opCtx.get()); - ASSERT_EQUALS(-1, _testRollbackDelete(_opCtx.get(), _coordinator, BSONObj())); + ASSERT_EQUALS(-1, + _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj())); } TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsNonCapped) { createOplog(_opCtx.get()); _createCollection(_opCtx.get(), "test.t", CollectionOptions()); - _testRollbackDelete(_opCtx.get(), _coordinator, BSONObj()); - ASSERT_EQUALS(0, _testRollbackDelete(_opCtx.get(), _coordinator, BSONObj())); + _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj()); + ASSERT_EQUALS(0, + _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj())); } TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsCapped) { @@ -361,15 +370,16 @@ TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsCapped) { CollectionOptions options; options.capped = true; _createCollection(_opCtx.get(), "test.t", options); - ASSERT_EQUALS(0, _testRollbackDelete(_opCtx.get(), _coordinator, BSONObj())); + ASSERT_EQUALS(0, + _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj())); } TEST_F(RSRollbackTest, RollbackDeleteRestoreDocument) { createOplog(_opCtx.get()); _createCollection(_opCtx.get(), "test.t", CollectionOptions()); BSONObj doc = BSON("_id" << 0 << "a" << 1); - _testRollbackDelete(_opCtx.get(), _coordinator, doc); - ASSERT_EQUALS(1, _testRollbackDelete(_opCtx.get(), _coordinator, doc)); + _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, doc); + ASSERT_EQUALS(1, _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, doc)); } TEST_F(RSRollbackTest, RollbackInsertDocumentWithNoId) { @@ -405,7 +415,8 @@ TEST_F(RSRollbackTest, RollbackInsertDocumentWithNoId) { OplogInterfaceMock({insertDocumentOperation, commonOperation}), rollbackSource, {}, - _coordinator); + _coordinator, + &_storageInterface); stopCapturingLogMessages(); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18752, status.location()); @@ -469,7 +480,8 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommand) { OplogInterfaceMock({insertDocumentOperation, insertDocumentOperation, commonOperation}), rollbackSource, {}, - _coordinator)); + _coordinator, + &_storageInterface)); stopCapturingLogMessages(); ASSERT_EQUALS(1, countLogLinesContaining("rollback drop index: collection: test.t. index: a_1")); @@ -529,7 +541,8 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandIndexNotInCatalog) { OplogInterfaceMock({insertDocumentOperation, commonOperation}), rollbackSource, {}, - _coordinator)); + _coordinator, + &_storageInterface)); stopCapturingLogMessages(); ASSERT_EQUALS(1, countLogLinesContaining("rollback drop index: collection: test.t. index: a_1")); @@ -577,7 +590,8 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandMissingNamespace) { OplogInterfaceMock({insertDocumentOperation, commonOperation}), rollbackSource, {}, - _coordinator); + _coordinator, + &_storageInterface); stopCapturingLogMessages(); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18752, status.location()); @@ -624,7 +638,8 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandInvalidNamespace) { OplogInterfaceMock({insertDocumentOperation, commonOperation}), rollbackSource, {}, - _coordinator); + _coordinator, + &_storageInterface); stopCapturingLogMessages(); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18752, status.location()); @@ -669,7 +684,8 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandMissingIndexName) { OplogInterfaceMock({insertDocumentOperation, commonOperation}), rollbackSource, {}, - _coordinator); + _coordinator, + &_storageInterface); stopCapturingLogMessages(); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18752, status.location()); @@ -705,7 +721,8 @@ TEST_F(RSRollbackTest, RollbackUnknownCommand) { commonOperation, }))), {}, - _coordinator); + _coordinator, + &_storageInterface); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18751, status.location()); } @@ -741,7 +758,8 @@ TEST_F(RSRollbackTest, RollbackDropCollectionCommand) { OplogInterfaceMock({dropCollectionOperation, commonOperation}), rollbackSource, {}, - _coordinator)); + _coordinator, + &_storageInterface)); ASSERT_TRUE(rollbackSource.called); } @@ -779,7 +797,8 @@ TEST_F(RSRollbackTest, RollbackDropCollectionCommandFailsIfRBIDChangesWhileSynci OplogInterfaceMock({dropCollectionOperation, commonOperation}), rollbackSource, {0}, - _coordinator), + _coordinator, + &_storageInterface), DBException, 40365); ASSERT(rollbackSource.copyCollectionCalled); @@ -903,7 +922,8 @@ TEST_F(RSRollbackTest, RollbackApplyOpsCommand) { OplogInterfaceMock({applyOpsOperation, commonOperation}), rollbackSource, {}, - _coordinator)); + _coordinator, + &_storageInterface)); ASSERT_EQUALS(4U, rollbackSource.searchedIds.size()); ASSERT_EQUALS(1U, rollbackSource.searchedIds.count(1)); ASSERT_EQUALS(1U, rollbackSource.searchedIds.count(2)); @@ -943,7 +963,8 @@ TEST_F(RSRollbackTest, RollbackCreateCollectionCommand) { OplogInterfaceMock({createCollectionOperation, commonOperation}), rollbackSource, {}, - _coordinator)); + _coordinator, + &_storageInterface)); { Lock::DBLock dbLock(_opCtx->lockState(), "test", MODE_S); auto db = dbHolder().get(_opCtx.get(), "test"); @@ -986,7 +1007,8 @@ TEST_F(RSRollbackTest, RollbackCollectionModificationCommand) { OplogInterfaceMock({collectionModificationOperation, commonOperation}), rollbackSource, {}, - _coordinator)); + _coordinator, + &_storageInterface)); stopCapturingLogMessages(); ASSERT_TRUE(rollbackSource.called); for (const auto& message : getCapturedLogMessages()) { @@ -1027,7 +1049,8 @@ TEST_F(RSRollbackTest, RollbackCollectionModificationCommandInvalidCollectionOpt OplogInterfaceMock({collectionModificationOperation, commonOperation}), rollbackSource, {}, - _coordinator); + _coordinator, + &_storageInterface); ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code()); ASSERT_EQUALS(18753, status.location()); } |