/**
* Copyright (C) 2015 MongoDB Inc.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License, version 3,
* as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*
* As a special exception, the copyright holders give permission to link the
* code of portions of this program with the OpenSSL library under certain
* conditions as described in each individual source file and distribute
* linked combinations including the program with the OpenSSL library. You
* must comply with the GNU Affero General Public License in all respects for
* all of the code used other than as permitted herein. If you modify file(s)
* with this exception, you may extend this exception to your version of the
* file(s), but you are not obligated to do so. If you do not wish to do so,
* delete this exception statement from your version. If you delete this
* exception statement from all source files in the program, then also delete
* it in the license file.
*/
#include "mongo/platform/basic.h"
#include
#include
#include
#include "mongo/client/fetcher.h"
#include "mongo/db/client.h"
#include "mongo/db/json.h"
#include "mongo/db/query/getmore_request.h"
#include "mongo/db/repl/base_cloner_test_fixture.h"
#include "mongo/db/repl/data_replicator_external_state_mock.h"
#include "mongo/db/repl/initial_syncer.h"
#include "mongo/db/repl/member_state.h"
#include "mongo/db/repl/oplog_entry.h"
#include "mongo/db/repl/oplog_fetcher.h"
#include "mongo/db/repl/optime.h"
#include "mongo/db/repl/reporter.h"
#include "mongo/db/repl/storage_interface.h"
#include "mongo/db/repl/storage_interface_mock.h"
#include "mongo/db/repl/sync_source_resolver.h"
#include "mongo/db/repl/sync_source_selector.h"
#include "mongo/db/repl/sync_source_selector_mock.h"
#include "mongo/db/repl/update_position_args.h"
#include "mongo/executor/network_interface_mock.h"
#include "mongo/executor/thread_pool_task_executor_test_fixture.h"
#include "mongo/stdx/mutex.h"
#include "mongo/unittest/task_executor_proxy.h"
#include "mongo/util/concurrency/old_thread_pool.h"
#include "mongo/util/concurrency/thread_name.h"
#include "mongo/util/fail_point_service.h"
#include "mongo/util/mongoutils/str.h"
#include "mongo/util/scopeguard.h"
#include "mongo/unittest/barrier.h"
#include "mongo/unittest/unittest.h"
namespace mongo {
namespace repl {
/**
* Insertion operator for InitialSyncer::State. Formats initial syncer state for output stream.
*/
std::ostream& operator<<(std::ostream& os, const InitialSyncer::State& state) {
switch (state) {
case InitialSyncer::State::kPreStart:
return os << "PreStart";
case InitialSyncer::State::kRunning:
return os << "Running";
case InitialSyncer::State::kShuttingDown:
return os << "ShuttingDown";
case InitialSyncer::State::kComplete:
return os << "Complete";
}
MONGO_UNREACHABLE;
}
} // namespace repl
} // namespace mongo
namespace {
using namespace mongo;
using namespace mongo::repl;
using executor::NetworkInterfaceMock;
using executor::RemoteCommandRequest;
using executor::RemoteCommandResponse;
using unittest::log;
using LockGuard = stdx::lock_guard;
using NetworkGuard = executor::NetworkInterfaceMock::InNetworkGuard;
using ResponseStatus = executor::TaskExecutor::ResponseStatus;
using UniqueLock = stdx::unique_lock;
struct CollectionCloneInfo {
CollectionMockStats stats;
CollectionBulkLoaderMock* loader = nullptr;
Status status{ErrorCodes::NotYetInitialized, ""};
};
class TaskExecutorMock : public unittest::TaskExecutorProxy {
public:
using ShouldFailRequestFn = stdx::function;
TaskExecutorMock(executor::TaskExecutor* executor, ShouldFailRequestFn shouldFailRequest)
: unittest::TaskExecutorProxy(executor), _shouldFailRequest(shouldFailRequest) {}
StatusWith scheduleWork(const CallbackFn& work) {
if (shouldFailScheduleWork) {
return Status(ErrorCodes::OperationFailed, "failed to schedule work");
}
return getExecutor()->scheduleWork(work);
}
StatusWith scheduleWorkAt(Date_t when, const CallbackFn& work) {
if (shouldFailScheduleWorkAt) {
return Status(ErrorCodes::OperationFailed,
str::stream() << "failed to schedule work at " << when.toString());
}
return getExecutor()->scheduleWorkAt(when, work);
}
StatusWith scheduleRemoteCommand(const executor::RemoteCommandRequest& request,
const RemoteCommandCallbackFn& cb) override {
if (_shouldFailRequest(request)) {
return Status(ErrorCodes::OperationFailed, "failed to schedule remote command");
}
return getExecutor()->scheduleRemoteCommand(request, cb);
}
bool shouldFailScheduleWork = false;
bool shouldFailScheduleWorkAt = false;
private:
ShouldFailRequestFn _shouldFailRequest;
};
class InitialSyncerTest : public executor::ThreadPoolExecutorTest, public SyncSourceSelector {
public:
InitialSyncerTest() {}
executor::ThreadPoolMock::Options makeThreadPoolMockOptions() const override;
/**
* clear/reset state
*/
void reset() {
_setMyLastOptime = [this](const OpTime& opTime) { _myLastOpTime = opTime; };
_myLastOpTime = OpTime();
_syncSourceSelector = stdx::make_unique();
}
// SyncSourceSelector
void clearSyncSourceBlacklist() override {
_syncSourceSelector->clearSyncSourceBlacklist();
}
HostAndPort chooseNewSyncSource(const OpTime& ot) override {
return _syncSourceSelector->chooseNewSyncSource(ot);
}
void blacklistSyncSource(const HostAndPort& host, Date_t until) override {
_syncSourceSelector->blacklistSyncSource(host, until);
}
bool shouldChangeSyncSource(const HostAndPort& currentSource,
const rpc::ReplSetMetadata& replMetadata,
boost::optional oqMetadata) override {
return _syncSourceSelector->shouldChangeSyncSource(currentSource, replMetadata, oqMetadata);
}
void scheduleNetworkResponse(std::string cmdName, const BSONObj& obj) {
NetworkInterfaceMock* net = getNet();
if (!net->hasReadyRequests()) {
log() << "The network doesn't have a request to process for this response: " << obj;
}
verifyNextRequestCommandName(cmdName);
scheduleNetworkResponse(net->getNextReadyRequest(), obj);
}
void scheduleNetworkResponse(NetworkInterfaceMock::NetworkOperationIterator noi,
const BSONObj& obj) {
NetworkInterfaceMock* net = getNet();
Milliseconds millis(0);
RemoteCommandResponse response(obj, BSONObj(), millis);
executor::TaskExecutor::ResponseStatus responseStatus(response);
log() << "Sending response for network request:";
log() << " req: " << noi->getRequest().dbname << "." << noi->getRequest().cmdObj;
log() << " resp:" << response;
net->scheduleResponse(noi, net->now(), responseStatus);
}
void scheduleNetworkResponse(std::string cmdName, Status errorStatus) {
NetworkInterfaceMock* net = getNet();
if (!getNet()->hasReadyRequests()) {
log() << "The network doesn't have a request to process for the error: " << errorStatus;
}
verifyNextRequestCommandName(cmdName);
net->scheduleResponse(net->getNextReadyRequest(), net->now(), errorStatus);
}
void processNetworkResponse(std::string cmdName, const BSONObj& obj) {
scheduleNetworkResponse(cmdName, obj);
finishProcessingNetworkResponse();
}
void processNetworkResponse(std::string cmdName, Status errorStatus) {
scheduleNetworkResponse(cmdName, errorStatus);
finishProcessingNetworkResponse();
}
/**
* Schedules and processes a successful response to the network request sent by InitialSyncer's
* last oplog entry fetcher. Also validates the find command arguments in the request.
*/
void processSuccessfulLastOplogEntryFetcherResponse(std::vector docs);
void finishProcessingNetworkResponse() {
getNet()->runReadyNetworkOperations();
if (getNet()->hasReadyRequests()) {
log() << "The network has unexpected requests to process, next req:";
NetworkInterfaceMock::NetworkOperation req = *getNet()->getNextReadyRequest();
log() << req.getDiagnosticString();
}
ASSERT_FALSE(getNet()->hasReadyRequests());
}
InitialSyncer& getInitialSyncer() {
return *_initialSyncer;
}
DataReplicatorExternalStateMock* getExternalState() {
return _externalState;
}
StorageInterface& getStorage() {
return *_storageInterface;
}
OldThreadPool& getDbWorkThreadPool() {
return *_dbWorkThreadPool;
}
protected:
struct StorageInterfaceResults {
bool createOplogCalled = false;
bool insertedOplogEntries = false;
int oplogEntriesInserted = 0;
bool droppedUserDBs = false;
std::vector droppedCollections;
int documentsInsertedCount = 0;
};
stdx::mutex _storageInterfaceWorkDoneMutex; // protects _storageInterfaceWorkDone.
StorageInterfaceResults _storageInterfaceWorkDone;
void setUp() override {
executor::ThreadPoolExecutorTest::setUp();
_storageInterface = stdx::make_unique();
_storageInterface->createOplogFn = [this](OperationContext* opCtx,
const NamespaceString& nss) {
LockGuard lock(_storageInterfaceWorkDoneMutex);
_storageInterfaceWorkDone.createOplogCalled = true;
return Status::OK();
};
_storageInterface->insertDocumentFn =
[this](OperationContext* opCtx, const NamespaceString& nss, const BSONObj& doc) {
LockGuard lock(_storageInterfaceWorkDoneMutex);
++_storageInterfaceWorkDone.documentsInsertedCount;
return Status::OK();
};
_storageInterface->insertDocumentsFn = [this](
OperationContext* opCtx, const NamespaceString& nss, const std::vector& ops) {
LockGuard lock(_storageInterfaceWorkDoneMutex);
_storageInterfaceWorkDone.insertedOplogEntries = true;
++_storageInterfaceWorkDone.oplogEntriesInserted;
return Status::OK();
};
_storageInterface->dropCollFn = [this](OperationContext* opCtx,
const NamespaceString& nss) {
LockGuard lock(_storageInterfaceWorkDoneMutex);
_storageInterfaceWorkDone.droppedCollections.push_back(nss.ns());
return Status::OK();
};
_storageInterface->dropUserDBsFn = [this](OperationContext* opCtx) {
LockGuard lock(_storageInterfaceWorkDoneMutex);
_storageInterfaceWorkDone.droppedUserDBs = true;
return Status::OK();
};
_storageInterface->createCollectionForBulkFn =
[this](const NamespaceString& nss,
const CollectionOptions& options,
const BSONObj idIndexSpec,
const std::vector& secondaryIndexSpecs) {
// Get collection info from map.
const auto collInfo = &_collections[nss];
if (collInfo->stats.initCalled) {
log() << "reusing collection during test which may cause problems, ns:" << nss;
}
(collInfo->loader = new CollectionBulkLoaderMock(&collInfo->stats))
->init(nullptr, secondaryIndexSpecs);
return StatusWith>(
std::unique_ptr(collInfo->loader));
};
_dbWorkThreadPool = stdx::make_unique(1);
Client::initThreadIfNotAlready();
reset();
launchExecutorThread();
_shouldFailRequest = [](const executor::RemoteCommandRequest&) { return false; };
_executorProxy = stdx::make_unique(
&getExecutor(), [this](const executor::RemoteCommandRequest& request) {
return _shouldFailRequest(request);
});
_myLastOpTime = OpTime({3, 0}, 1);
InitialSyncerOptions options;
options.initialSyncRetryWait = Milliseconds(1);
options.getMyLastOptime = [this]() { return _myLastOpTime; };
options.setMyLastOptime = [this](const OpTime& opTime) { _setMyLastOptime(opTime); };
options.getSlaveDelay = [this]() { return Seconds(0); };
options.syncSourceSelector = this;
_options = options;
ThreadPool::Options threadPoolOptions;
threadPoolOptions.poolName = "replication";
threadPoolOptions.minThreads = 1U;
threadPoolOptions.maxThreads = 1U;
threadPoolOptions.onCreateThread = [](const std::string& threadName) {
Client::initThread(threadName.c_str());
};
auto dataReplicatorExternalState = stdx::make_unique();
dataReplicatorExternalState->taskExecutor = _executorProxy.get();
dataReplicatorExternalState->dbWorkThreadPool = &getDbWorkThreadPool();
dataReplicatorExternalState->currentTerm = 1LL;
dataReplicatorExternalState->lastCommittedOpTime = _myLastOpTime;
{
ReplSetConfig config;
ASSERT_OK(config.initialize(BSON("_id"
<< "myset"
<< "version"
<< 1
<< "protocolVersion"
<< 1
<< "members"
<< BSON_ARRAY(BSON("_id" << 0 << "host"
<< "localhost:12345"))
<< "settings"
<< BSON("electionTimeoutMillis" << 10000))));
dataReplicatorExternalState->replSetConfigResult = config;
}
_externalState = dataReplicatorExternalState.get();
_lastApplied = getDetectableErrorStatus();
_onCompletion = [this](const StatusWith& lastApplied) {
_lastApplied = lastApplied;
};
try {
// When creating InitialSyncer, we wrap _onCompletion so that we can override the
// InitialSyncer's callback behavior post-construction.
// See InitialSyncerTransitionsToCompleteWhenFinishCallbackThrowsException.
_initialSyncer = stdx::make_unique(
options,
std::move(dataReplicatorExternalState),
_storageInterface.get(),
[this](const StatusWith& lastApplied) {
_onCompletion(lastApplied);
});
_initialSyncer->setScheduleDbWorkFn_forTest(
[this](const executor::TaskExecutor::CallbackFn& work) {
return getExecutor().scheduleWork(work);
});
} catch (...) {
ASSERT_OK(exceptionToStatus());
}
}
void tearDownExecutorThread() {
if (_executorThreadShutdownComplete) {
return;
}
executor::ThreadPoolExecutorTest::shutdownExecutorThread();
executor::ThreadPoolExecutorTest::joinExecutorThread();
_executorThreadShutdownComplete = true;
}
void tearDown() override {
tearDownExecutorThread();
_initialSyncer.reset();
_dbWorkThreadPool->join();
_dbWorkThreadPool.reset();
_storageInterface.reset();
// tearDown() destroys the task executor which was referenced by the initial syncer.
executor::ThreadPoolExecutorTest::tearDown();
}
/**
* Note: An empty cmdName will skip validation.
*/
void verifyNextRequestCommandName(std::string cmdName) {
const auto net = getNet();
ASSERT_TRUE(net->hasReadyRequests());
if (cmdName != "") {
const NetworkInterfaceMock::NetworkOperationIterator req =
net->getFrontOfUnscheduledQueue();
const BSONObj reqBSON = req->getRequest().cmdObj;
const BSONElement cmdElem = reqBSON.firstElement();
auto reqCmdName = cmdElem.fieldNameStringData();
ASSERT_EQ(cmdName, reqCmdName);
}
}
TaskExecutorMock::ShouldFailRequestFn _shouldFailRequest;
std::unique_ptr _executorProxy;
InitialSyncerOptions _options;
InitialSyncerOptions::SetMyLastOptimeFn _setMyLastOptime;
OpTime _myLastOpTime;
std::unique_ptr _syncSourceSelector;
std::unique_ptr _storageInterface;
std::unique_ptr _dbWorkThreadPool;
std::map _collectionStats;
std::map _collections;
StatusWith _lastApplied = Status(ErrorCodes::NotYetInitialized, "");
InitialSyncer::OnCompletionFn _onCompletion;
private:
DataReplicatorExternalStateMock* _externalState;
std::unique_ptr _initialSyncer;
bool _executorThreadShutdownComplete = false;
};
executor::ThreadPoolMock::Options InitialSyncerTest::makeThreadPoolMockOptions() const {
executor::ThreadPoolMock::Options options;
options.onCreateThread = []() { Client::initThread("InitialSyncerTest"); };
return options;
}
void advanceClock(NetworkInterfaceMock* net, Milliseconds duration) {
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
auto when = net->now() + duration;
ASSERT_EQUALS(when, net->runUntil(when));
}
ServiceContext::UniqueOperationContext makeOpCtx() {
return cc().makeOperationContext();
}
/**
* Generates a replSetGetRBID response.
*/
BSONObj makeRollbackCheckerResponse(int rollbackId) {
return BSON("ok" << 1 << "rbid" << rollbackId);
}
/**
* Generates a cursor response for a Fetcher to consume.
*/
RemoteCommandResponse makeCursorResponse(CursorId cursorId,
const NamespaceString& nss,
std::vector docs,
bool isFirstBatch = true,
int rbid = 1) {
OpTime futureOpTime(Timestamp(1000, 1000), 1000);
rpc::OplogQueryMetadata oqMetadata(futureOpTime, futureOpTime, rbid, 0, 0);
BSONObjBuilder metadataBob;
ASSERT_OK(oqMetadata.writeToMetadata(&metadataBob));
auto metadataObj = metadataBob.obj();
BSONObjBuilder bob;
{
BSONObjBuilder cursorBob(bob.subobjStart("cursor"));
cursorBob.append("id", cursorId);
cursorBob.append("ns", nss.toString());
{
BSONArrayBuilder batchBob(
cursorBob.subarrayStart(isFirstBatch ? "firstBatch" : "nextBatch"));
for (const auto& doc : docs) {
batchBob.append(doc);
}
}
}
bob.append("ok", 1);
return {bob.obj(), metadataObj, Milliseconds(0)};
}
/**
* Generates a listDatabases response for a DatabasesCloner to consume.
*/
BSONObj makeListDatabasesResponse(std::vector databaseNames) {
BSONObjBuilder bob;
{
BSONArrayBuilder databasesBob(bob.subarrayStart("databases"));
for (const auto& name : databaseNames) {
BSONObjBuilder nameBob(databasesBob.subobjStart());
nameBob.append("name", name);
}
}
bob.append("ok", 1);
return bob.obj();
}
/**
* Generates oplog entries with the given number used for the timestamp.
*/
BSONObj makeOplogEntry(int t, const char* opType = "i", int version = OplogEntry::kOplogVersion) {
return BSON("ts" << Timestamp(t, 1) << "h" << static_cast(t) << "ns"
<< "a.a"
<< "v"
<< version
<< "op"
<< opType
<< "o"
<< BSON("_id" << t << "a" << t));
}
void InitialSyncerTest::processSuccessfulLastOplogEntryFetcherResponse(std::vector docs) {
auto net = getNet();
auto request = assertRemoteCommandNameEquals(
"find",
net->scheduleSuccessfulResponse(makeCursorResponse(0LL, _options.localOplogNS, docs)));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
ASSERT_TRUE(request.cmdObj.hasField("sort"));
ASSERT_EQUALS(mongo::BSONType::Object, request.cmdObj["sort"].type());
ASSERT_BSONOBJ_EQ(BSON("$natural" << -1), request.cmdObj.getObjectField("sort"));
net->runReadyNetworkOperations();
}
TEST_F(InitialSyncerTest, InvalidConstruction) {
InitialSyncerOptions options;
options.getMyLastOptime = []() { return OpTime(); };
options.setMyLastOptime = [](const OpTime&) {};
options.getSlaveDelay = []() { return Seconds(0); };
options.syncSourceSelector = this;
auto callback = [](const StatusWith&) {};
// Null task executor in external state.
{
auto dataReplicatorExternalState = stdx::make_unique();
ASSERT_THROWS_CODE_AND_WHAT(
InitialSyncer(
options, std::move(dataReplicatorExternalState), _storageInterface.get(), callback),
UserException,
ErrorCodes::BadValue,
"task executor cannot be null");
}
// Null callback function.
{
auto dataReplicatorExternalState = stdx::make_unique();
dataReplicatorExternalState->taskExecutor = &getExecutor();
ASSERT_THROWS_CODE_AND_WHAT(InitialSyncer(options,
std::move(dataReplicatorExternalState),
_storageInterface.get(),
InitialSyncer::OnCompletionFn()),
UserException,
ErrorCodes::BadValue,
"callback function cannot be null");
}
}
TEST_F(InitialSyncerTest, CreateDestroy) {}
const std::uint32_t maxAttempts = 1U;
TEST_F(InitialSyncerTest, StartupReturnsIllegalOperationIfAlreadyActive) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
ASSERT_FALSE(initialSyncer->isActive());
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(initialSyncer->isActive());
ASSERT_EQUALS(ErrorCodes::IllegalOperation, initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(initialSyncer->isActive());
}
TEST_F(InitialSyncerTest, StartupReturnsShutdownInProgressIfInitialSyncerIsShuttingDown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
ASSERT_FALSE(initialSyncer->isActive());
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(initialSyncer->isActive());
// SyncSourceSelector returns an invalid sync source so InitialSyncer is stuck waiting for
// another sync source in 'Options::syncSourceRetryWait' ms.
ASSERT_OK(initialSyncer->shutdown());
ASSERT_EQUALS(ErrorCodes::ShutdownInProgress, initialSyncer->startup(opCtx.get(), maxAttempts));
}
TEST_F(InitialSyncerTest, StartupReturnsShutdownInProgressIfExecutorIsShutdown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
getExecutor().shutdown();
ASSERT_EQUALS(ErrorCodes::ShutdownInProgress, initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_FALSE(initialSyncer->isActive());
// Cannot startup initial syncer again since it's in the Complete state.
ASSERT_EQUALS(ErrorCodes::ShutdownInProgress, initialSyncer->startup(opCtx.get(), maxAttempts));
}
TEST_F(InitialSyncerTest, ShutdownTransitionsStateToCompleteIfCalledBeforeStartup) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
ASSERT_OK(initialSyncer->shutdown());
ASSERT_EQUALS(ErrorCodes::ShutdownInProgress, initialSyncer->startup(opCtx.get(), maxAttempts));
// Initial syncer is inactive when it's in the Complete state.
ASSERT_FALSE(initialSyncer->isActive());
}
TEST_F(InitialSyncerTest, StartupSetsInitialSyncFlagOnSuccess) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Initial sync flag should not be set before starting.
ASSERT_FALSE(getStorage().getInitialSyncFlag(opCtx.get()));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(initialSyncer->isActive());
// Initial sync flag should be set.
ASSERT_TRUE(getStorage().getInitialSyncFlag(opCtx.get()));
}
TEST_F(InitialSyncerTest, InitialSyncerReturnsCallbackCanceledIfShutdownImmediatelyAfterStartup) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
// This will cancel the _startInitialSyncAttemptCallback() task scheduled by startup().
ASSERT_OK(initialSyncer->shutdown());
// Depending on which InitialSyncer stage (_chooseSyncSource or _rollbackCheckerResetCallback)
// was interrupted by shutdown(), we may have to request the network interface to deliver
// cancellation signals to the InitialSyncer callbacks in for InitialSyncer to run to
// completion.
executor::NetworkInterfaceMock::InNetworkGuard(getNet())->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerRetriesSyncSourceSelectionIfChooseNewSyncSourceReturnsInvalidSyncSource) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Override chooseNewSyncSource() result in SyncSourceSelectorMock before calling startup()
// because InitialSyncer will look for a valid sync source immediately after startup.
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort());
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
// Run first sync source selection attempt.
executor::NetworkInterfaceMock::InNetworkGuard(getNet())->runReadyNetworkOperations();
// InitialSyncer will not drop user databases while looking for a valid sync source.
ASSERT_FALSE(_storageInterfaceWorkDone.droppedUserDBs);
// First sync source selection attempt failed. Update SyncSourceSelectorMock to return valid
// sync source next time chooseNewSyncSource() is called.
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
// Advance clock until the next sync source selection attempt.
advanceClock(getNet(), _options.syncSourceRetryWait);
// DataReplictor drops user databases after obtaining a valid sync source.
ASSERT_TRUE(_storageInterfaceWorkDone.droppedUserDBs);
}
const std::uint32_t chooseSyncSourceMaxAttempts = 10U;
/**
* Advances executor clock so that InitialSyncer exhausts all 'chooseSyncSourceMaxAttempts' (server
* parameter numInitialSyncConnectAttempts) sync source selection attempts.
* If SyncSourceSelectorMock keeps returning an invalid sync source, InitialSyncer will retry every
* '_options.syncSourceRetryWait' ms up to a maximum of 'chooseSyncSourceMaxAttempts' attempts.
*/
void _simulateChooseSyncSourceFailure(executor::NetworkInterfaceMock* net,
Milliseconds syncSourceRetryWait) {
advanceClock(net, int(chooseSyncSourceMaxAttempts - 1) * syncSourceRetryWait);
}
TEST_F(
InitialSyncerTest,
InitialSyncerReturnsInitialSyncOplogSourceMissingIfNoValidSyncSourceCanBeFoundAfterTenFailedChooseSyncSourceAttempts) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Override chooseNewSyncSource() result in SyncSourceSelectorMock before calling startup()
// because InitialSyncer will look for a valid sync source immediately after startup.
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort());
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
_simulateChooseSyncSourceFailure(getNet(), _options.syncSourceRetryWait);
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::InitialSyncOplogSourceMissing, _lastApplied);
}
// Confirms that InitialSyncer keeps retrying initial sync.
// Make every initial sync attempt fail early by having the sync source selector always return an
// invalid sync source.
TEST_F(InitialSyncerTest,
InitialSyncerRetriesInitialSyncUpToMaxAttemptsAndReturnsLastAttemptError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort());
const std::uint32_t initialSyncMaxAttempts = 3U;
ASSERT_OK(initialSyncer->startup(opCtx.get(), initialSyncMaxAttempts));
auto net = getNet();
for (std::uint32_t i = 0; i < initialSyncMaxAttempts; ++i) {
_simulateChooseSyncSourceFailure(net, _options.syncSourceRetryWait);
advanceClock(net, _options.initialSyncRetryWait);
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::InitialSyncOplogSourceMissing, _lastApplied);
// Check number of failed attempts in stats.
auto progress = initialSyncer->getInitialSyncProgress();
unittest::log() << "Progress after " << initialSyncMaxAttempts
<< " failed attempts: " << progress;
ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), int(initialSyncMaxAttempts))
<< progress;
ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), int(initialSyncMaxAttempts))
<< progress;
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsCallbackCanceledIfShutdownWhileRetryingSyncSourceSelection) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort());
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
auto when = net->now() + _options.syncSourceRetryWait / 2;
ASSERT_GREATER_THAN(when, net->now());
ASSERT_EQUALS(when, net->runUntil(when));
}
// This will cancel the _chooseSyncSourceCallback() task scheduled at getNet()->now() +
// '_options.syncSourceRetryWait'.
ASSERT_OK(initialSyncer->shutdown());
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsScheduleErrorIfTaskExecutorFailsToScheduleNextChooseSyncSourceCallback) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort());
_executorProxy->shouldFailScheduleWorkAt = true;
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsScheduleErrorIfTaskExecutorFailsToScheduleNextInitialSyncAttempt) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort());
ASSERT_EQUALS(InitialSyncer::State::kPreStart, initialSyncer->getState_forTest());
ASSERT_OK(initialSyncer->startup(opCtx.get(), 2U));
ASSERT_EQUALS(InitialSyncer::State::kRunning, initialSyncer->getState_forTest());
// Advance clock so that we run all but the last sync source callback.
auto net = getNet();
advanceClock(net, int(chooseSyncSourceMaxAttempts - 2) * _options.syncSourceRetryWait);
// Last choose sync source attempt should now be scheduled. Advance clock so we fail last
// choose sync source attempt which cause the next initial sync attempt to be scheduled.
_executorProxy->shouldFailScheduleWorkAt = true;
advanceClock(net, _options.syncSourceRetryWait);
initialSyncer->join();
ASSERT_EQUALS(InitialSyncer::State::kComplete, initialSyncer->getState_forTest());
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
// This test verifies that the initial syncer will still transition to a complete state even if
// the completion callback function throws an exception.
TEST_F(InitialSyncerTest, InitialSyncerTransitionsToCompleteWhenFinishCallbackThrowsException) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_onCompletion = [this](const StatusWith& lastApplied) {
_lastApplied = lastApplied;
uassert(ErrorCodes::InternalError, "", false);
};
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort());
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_OK(initialSyncer->shutdown());
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
class SharedCallbackState {
MONGO_DISALLOW_COPYING(SharedCallbackState);
public:
explicit SharedCallbackState(bool* sharedCallbackStateDestroyed)
: _sharedCallbackStateDestroyed(sharedCallbackStateDestroyed) {}
~SharedCallbackState() {
*_sharedCallbackStateDestroyed = true;
}
private:
bool* _sharedCallbackStateDestroyed;
};
TEST_F(InitialSyncerTest, InitialSyncerResetsOnCompletionCallbackFunctionPointerUponCompletion) {
bool sharedCallbackStateDestroyed = false;
auto sharedCallbackData = std::make_shared(&sharedCallbackStateDestroyed);
decltype(_lastApplied) lastApplied = getDetectableErrorStatus();
auto dataReplicatorExternalState = stdx::make_unique();
dataReplicatorExternalState->taskExecutor = &getExecutor();
auto initialSyncer = stdx::make_unique(
_options,
std::move(dataReplicatorExternalState),
_storageInterface.get(),
[&lastApplied, sharedCallbackData](const StatusWith& result) {
lastApplied = result;
});
ON_BLOCK_EXIT([this]() { getExecutor().shutdown(); });
auto opCtx = makeOpCtx();
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
sharedCallbackData.reset();
ASSERT_FALSE(sharedCallbackStateDestroyed);
ASSERT_OK(initialSyncer->shutdown());
// Depending on which InitialSyncer stage (_chooseSyncSource or _rollbackCheckerResetCallback)
// was interrupted by shutdown(), we may have to request the network interface to deliver
// cancellation signals to the InitialSyncer callbacks in for InitialSyncer to run to
// completion.
executor::NetworkInterfaceMock::InNetworkGuard(getNet())->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, lastApplied);
// InitialSyncer should reset 'InitialSyncer::_onCompletion' after running callback function
// for the last time before becoming inactive.
// This ensures that we release resources associated with 'InitialSyncer::_onCompletion'.
ASSERT_TRUE(sharedCallbackStateDestroyed);
}
TEST_F(InitialSyncerTest, InitialSyncerRecreatesOplogAndDropsReplicatedDatabases) {
// We are not interested in proceeding beyond the oplog creation stage so we inject a failure
// after setting '_storageInterfaceWorkDone.createOplogCalled' to true.
auto oldCreateOplogFn = _storageInterface->createOplogFn;
_storageInterface->createOplogFn = [oldCreateOplogFn](OperationContext* opCtx,
const NamespaceString& nss) {
oldCreateOplogFn(opCtx, nss);
return Status(ErrorCodes::OperationFailed, "oplog creation failed");
};
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
LockGuard lock(_storageInterfaceWorkDoneMutex);
ASSERT_TRUE(_storageInterfaceWorkDone.droppedUserDBs);
ASSERT_TRUE(_storageInterfaceWorkDone.createOplogCalled);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetRollbackIdScheduleError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// replSetGetRBID is the first remote command to be scheduled by the initial syncer after
// creating the oplog collection.
executor::RemoteCommandRequest request;
_shouldFailRequest = [&request](const executor::RemoteCommandRequest& requestToSend) {
request = requestToSend;
return true;
};
HostAndPort syncSource("localhost", 12345);
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(syncSource);
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
ASSERT_EQUALS("admin", request.dbname);
assertRemoteCommandNameEquals("replSetGetRBID", request);
ASSERT_EQUALS(syncSource, request.target);
}
TEST_F(
InitialSyncerTest,
InitialSyncerReturnsShutdownInProgressIfSchedulingRollbackCheckerFailedDueToExecutorShutdown) {
// The rollback id request is sent immediately after oplog creation. We shut the task executor
// down before returning from createOplog() to make the scheduleRemoteCommand() call for
// replSetGetRBID fail.
auto oldCreateOplogFn = _storageInterface->createOplogFn;
_storageInterface->createOplogFn = [oldCreateOplogFn, this](OperationContext* opCtx,
const NamespaceString& nss) {
auto status = oldCreateOplogFn(opCtx, nss);
getExecutor().shutdown();
return status;
};
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::ShutdownInProgress, _lastApplied);
LockGuard lock(_storageInterfaceWorkDoneMutex);
ASSERT_TRUE(_storageInterfaceWorkDone.createOplogCalled);
}
TEST_F(InitialSyncerTest, InitialSyncerCancelsRollbackCheckerOnShutdown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
HostAndPort syncSource("localhost", 12345);
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(syncSource);
ASSERT_EQUALS(InitialSyncer::State::kPreStart, initialSyncer->getState_forTest());
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_EQUALS(InitialSyncer::State::kRunning, initialSyncer->getState_forTest());
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
ASSERT_TRUE(net->hasReadyRequests());
auto noi = net->getNextReadyRequest();
const auto& request = assertRemoteCommandNameEquals("replSetGetRBID", noi->getRequest());
ASSERT_EQUALS("admin", request.dbname);
ASSERT_EQUALS(syncSource, request.target);
net->blackHole(noi);
}
ASSERT_OK(initialSyncer->shutdown());
// Since we need to request the NetworkInterfaceMock to deliver the cancellation event,
// the InitialSyncer has to be in a pre-completion state (ie. ShuttingDown).
ASSERT_EQUALS(InitialSyncer::State::kShuttingDown, initialSyncer->getState_forTest());
executor::NetworkInterfaceMock::InNetworkGuard(net)->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(InitialSyncer::State::kComplete, initialSyncer->getState_forTest());
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughRollbackCheckerCallbackError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
assertRemoteCommandNameEquals(
"replSetGetRBID",
net->scheduleErrorResponse(
Status(ErrorCodes::OperationFailed, "replSetGetRBID failed at sync source")));
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughLastOplogEntryFetcherScheduleError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// The last oplog entry fetcher is the first component that sends a find command so we reject
// any find commands and save the request for inspection at the end of this test case.
executor::RemoteCommandRequest request;
_shouldFailRequest = [&request](const executor::RemoteCommandRequest& requestToSend) {
request = requestToSend;
return "find" == requestToSend.cmdObj.firstElement().fieldNameStringData();
};
HostAndPort syncSource("localhost", 12345);
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(syncSource);
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
ASSERT_EQUALS(syncSource, request.target);
ASSERT_EQUALS(_options.localOplogNS.db(), request.dbname);
assertRemoteCommandNameEquals("find", request);
ASSERT_BSONOBJ_EQ(BSON("$natural" << -1), request.cmdObj.getObjectField("sort"));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughLastOplogEntryFetcherCallbackError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
assertRemoteCommandNameEquals(
"find",
net->scheduleErrorResponse(
Status(ErrorCodes::OperationFailed, "find command failed at sync source")));
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerCancelsLastOplogEntryFetcherOnShutdown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
ASSERT_TRUE(net->hasReadyRequests());
}
ASSERT_OK(initialSyncer->shutdown());
executor::NetworkInterfaceMock::InNetworkGuard(net)->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsNoMatchingDocumentIfLastOplogEntryFetcherReturnsEmptyBatchOfDocuments) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({});
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::NoMatchingDocument, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerResendsFindCommandIfLastOplogEntryFetcherReturnsRetriableError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry first attempt - retriable error.
assertRemoteCommandNameEquals("find",
net->scheduleErrorResponse(Status(ErrorCodes::HostNotFound, "")));
net->runReadyNetworkOperations();
// InitialSyncer stays active because it resends the find request for the last oplog entry.
ASSERT_TRUE(initialSyncer->isActive());
// Last oplog entry second attempt.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsNoSuchKeyIfLastOplogEntryFetcherReturnsEntryWithMissingHash) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({BSONObj()});
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::NoSuchKey, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsNoSuchKeyIfLastOplogEntryFetcherReturnsEntryWithMissingTimestamp) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({BSON("h" << 1LL)});
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::NoSuchKey, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerPassesThroughErrorFromDataReplicatorExternalStateGetCurrentConfig) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
getExternalState()->replSetConfigResult = Status(ErrorCodes::OperationFailed, "");
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughOplogFetcherScheduleError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Make the tailable oplog query fail. Allow all other requests to be scheduled.
executor::RemoteCommandRequest request;
_shouldFailRequest = [&request](const executor::RemoteCommandRequest& requestToSend) {
if ("find" == requestToSend.cmdObj.firstElement().fieldNameStringData() &&
requestToSend.cmdObj.getBoolField("tailable")) {
request = requestToSend;
return true;
}
return false;
};
HostAndPort syncSource("localhost", 12345);
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(syncSource);
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
ASSERT_EQUALS(syncSource, request.target);
ASSERT_EQUALS(_options.localOplogNS.db(), request.dbname);
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughOplogFetcherCallbackError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, _options.localOplogNS, {makeOplogEntry(1)}));
net->runReadyNetworkOperations();
// Oplog tailing query.
auto request = assertRemoteCommandNameEquals(
"find", net->scheduleErrorResponse(Status(ErrorCodes::OperationFailed, "dead cursor")));
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->runReadyNetworkOperations();
// OplogFetcher will shut down DatabasesCloner on error after setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _databasesClonerCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerSucceedsOnEarlyOplogFetcherCompletionIfThereAreNoOperationsToApply) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
auto request =
assertRemoteCommandNameEquals("find",
net->scheduleSuccessfulResponse(makeCursorResponse(
0LL, _options.localOplogNS, {makeOplogEntry(1)})));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
net->runReadyNetworkOperations();
// Oplog tailing query.
// Simulate cursor closing on sync source.
request =
assertRemoteCommandNameEquals("find",
net->scheduleSuccessfulResponse(makeCursorResponse(
0LL, _options.localOplogNS, {makeOplogEntry(1)})));
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->runReadyNetworkOperations();
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// Last rollback checker replSetGetRBID command.
assertRemoteCommandNameEquals(
"replSetGetRBID", net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1)));
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(OplogEntry(makeOplogEntry(1)).getOpTime(),
unittest::assertGet(_lastApplied).opTime);
}
TEST_F(
InitialSyncerTest,
InitialSyncerSucceedsOnEarlyOplogFetcherCompletionIfThereAreEnoughOperationsInTheOplogBufferToReachEndTimestamp) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// Oplog tailing query.
// Simulate cursor closing on sync source.
auto request = assertRemoteCommandNameEquals(
"find",
net->scheduleSuccessfulResponse(makeCursorResponse(
0LL,
_options.localOplogNS,
{makeOplogEntry(1), makeOplogEntry(2, "c"), makeOplogEntry(3, "c")})));
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->runReadyNetworkOperations();
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(3)});
// Last rollback checker replSetGetRBID command.
assertRemoteCommandNameEquals(
"replSetGetRBID", net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1)));
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(OplogEntry(makeOplogEntry(3)).getOpTime(),
unittest::assertGet(_lastApplied).opTime);
}
TEST_F(
InitialSyncerTest,
InitialSyncerReturnsRemoteResultsUnavailableOnEarlyOplogFetcherCompletionIfThereAreNotEnoughOperationsInTheOplogBufferToReachEndTimestamp) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// Oplog tailing query.
// Simulate cursor closing on sync source.
auto request = assertRemoteCommandNameEquals(
"find",
net->scheduleSuccessfulResponse(makeCursorResponse(
0LL,
_options.localOplogNS,
{makeOplogEntry(1), makeOplogEntry(2, "c"), makeOplogEntry(3, "c")})));
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->runReadyNetworkOperations();
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
// Return an oplog entry with an optime that is more recent than what the completed
// OplogFetcher has read from the sync source.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(4)});
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::RemoteResultsUnavailable, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerPassesThroughDatabasesClonerScheduleErrorAndCancelsOplogFetcher) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Make the listDatabases command fail. Allow all other requests to be scheduled.
executor::RemoteCommandRequest request;
_shouldFailRequest = [&request](const executor::RemoteCommandRequest& requestToSend) {
if ("listDatabases" == requestToSend.cmdObj.firstElement().fieldNameStringData()) {
request = requestToSend;
return true;
}
return false;
};
HostAndPort syncSource("localhost", 12345);
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(syncSource);
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// InitialSyncer shuts down OplogFetcher when it fails to schedule DatabasesCloner
// so we should not expect any network requests in the queue.
ASSERT_FALSE(net->hasReadyRequests());
// OplogFetcher is shutting down but we still need to call runReadyNetworkOperations()
// to deliver the cancellation status to the 'InitialSyncer::_oplogFetcherCallback'
// callback.
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
ASSERT_EQUALS(syncSource, request.target);
ASSERT_EQUALS("admin", request.dbname);
assertRemoteCommandNameEquals("listDatabases", request);
}
TEST_F(InitialSyncerTest,
InitialSyncerPassesThroughDatabasesClonerCallbackErrorAndCancelsOplogFetcher) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// Oplog tailing query.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// DatabasesCloner's first remote command - listDatabases
assertRemoteCommandNameEquals(
"listDatabases",
net->scheduleErrorResponse(Status(ErrorCodes::FailedToParse, "listDatabases failed")));
net->runReadyNetworkOperations();
// DatabasesCloner will shut down OplogFetcher on error after setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::FailedToParse, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerIgnoresLocalDatabasesWhenCloningDatabases) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// Oplog tailing query.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// DatabasesCloner's first remote command - listDatabases
assertRemoteCommandNameEquals(
"listDatabases",
net->scheduleSuccessfulResponse(makeListDatabasesResponse({"a", "local", "b"})));
net->runReadyNetworkOperations();
// DatabasesCloner should only send listCollections requests for databases 'a' and 'b'.
request = assertRemoteCommandNameEquals(
"listCollections",
net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, NamespaceString::makeListCollectionsNSS("a"), {})));
ASSERT_EQUALS("a", request.dbname);
request = assertRemoteCommandNameEquals(
"listCollections",
net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, NamespaceString::makeListCollectionsNSS("b"), {})));
ASSERT_EQUALS("b", request.dbname);
// After processing all the database names and returning empty lists of collections for each
// database, data cloning should run to completion and we should expect to see a last oplog
// entry fetcher request.
request = assertRemoteCommandNameEquals(
"find",
net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, NamespaceString::makeListCollectionsNSS("b"), {})));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
}
getExecutor().shutdown();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerIgnoresDatabaseInfoDocumentWithoutNameFieldWhenCloningDatabases) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// Oplog tailing query.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// DatabasesCloner's first remote command - listDatabases
assertRemoteCommandNameEquals(
"listDatabases",
net->scheduleSuccessfulResponse(BSON("databases" << BSON_ARRAY(BSON("name"
<< "a")
<< BSON("bad"
<< "dbinfo")
<< BSON("name"
<< "b"))
<< "ok"
<< 1)));
net->runReadyNetworkOperations();
// DatabasesCloner should only send listCollections requests for databases 'a' and 'b'.
request = assertRemoteCommandNameEquals(
"listCollections",
net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, NamespaceString::makeListCollectionsNSS("a"), {})));
ASSERT_EQUALS("a", request.dbname);
request = assertRemoteCommandNameEquals(
"listCollections",
net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, NamespaceString::makeListCollectionsNSS("b"), {})));
ASSERT_EQUALS("b", request.dbname);
// After processing all the database names and returning empty lists of collections for each
// database, data cloning should run to completion and we should expect to see a last oplog
// entry fetcher request.
request = assertRemoteCommandNameEquals(
"find",
net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, NamespaceString::makeListCollectionsNSS("b"), {})));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
}
getExecutor().shutdown();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerCancelsBothOplogFetcherAndDatabasesClonerOnShutdown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
}
ASSERT_OK(initialSyncer->shutdown());
executor::NetworkInterfaceMock::InNetworkGuard(net)->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerPassesThroughSecondLastOplogEntryFetcherScheduleErrorAndCancelsOplogFetcher) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Make the second last oplog entry fetcher command fail. Allow all other requests to be
// scheduled.
executor::RemoteCommandRequest request;
bool first = true;
_shouldFailRequest = [&first, &request](const executor::RemoteCommandRequest& requestToSend) {
if ("find" == requestToSend.cmdObj.firstElement().fieldNameStringData() &&
requestToSend.cmdObj.hasField("sort") &&
1 == requestToSend.cmdObj.getIntField("limit")) {
if (first) {
first = false;
return false;
}
request = requestToSend;
return true;
}
return false;
};
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// DatabasesCloner will shut down the OplogFetcher on failing to schedule the last entry
// oplog fetcher after setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerPassesThroughSecondLastOplogEntryFetcherCallbackErrorAndCancelsOplogFetcher) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
request = assertRemoteCommandNameEquals(
"find",
net->scheduleErrorResponse(
Status(ErrorCodes::OperationFailed, "second last oplog entry fetcher failed")));
ASSERT_TRUE(request.cmdObj.hasField("sort"));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
net->runReadyNetworkOperations();
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerCancelsBothSecondLastOplogEntryFetcherAndOplogFetcherOnShutdown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
request = assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
noi = net->getNextReadyRequest();
request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.hasField("sort"));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
net->blackHole(noi);
}
initialSyncer->shutdown();
executor::NetworkInterfaceMock::InNetworkGuard(net)->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerCancelsSecondLastOplogEntryFetcherOnOplogFetcherCallbackError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// Save request for OplogFetcher's oplog tailing query. This request will be canceled.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
auto oplogFetcherNetworkOperationIterator = noi;
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
// Blackhole this request which will be canceled when oplog fetcher fails.
noi = net->getNextReadyRequest();
request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.hasField("sort"));
ASSERT_EQUALS(1, request.cmdObj.getIntField("limit"));
net->blackHole(noi);
// Make oplog fetcher fail.
net->scheduleErrorResponse(oplogFetcherNetworkOperationIterator,
Status(ErrorCodes::OperationFailed, "oplog fetcher failed"));
net->runReadyNetworkOperations();
// _oplogFetcherCallback() will shut down the '_lastOplogEntryFetcher' after setting the
// completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _lastOplogEntryFetcherCallbackAfterCloningData().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(
InitialSyncerTest,
InitialSyncerReturnsTypeMismatchErrorWhenSecondLastOplogEntryFetcherReturnsMalformedDocument) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({BSON("h"
<< "not a hash")});
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::TypeMismatch, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsOplogOutOfOrderIfStopTimestampPrecedesBeginTimestamp) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OplogOutOfOrder, _lastApplied);
}
TEST_F(
InitialSyncerTest,
InitialSyncerPassesThroughInsertOplogSeedDocumentErrorAfterDataCloningFinishesWithNoOperationsToApply) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
NamespaceString insertDocumentNss;
BSONObj insertDocumentDoc;
_storageInterface->insertDocumentFn = [&insertDocumentDoc, &insertDocumentNss](
OperationContext*, const NamespaceString& nss, const BSONObj& doc) {
insertDocumentNss = nss;
insertDocumentDoc = doc;
return Status(ErrorCodes::OperationFailed, "failed to insert oplog entry");
};
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
ASSERT_EQUALS(_options.localOplogNS, insertDocumentNss);
ASSERT_BSONOBJ_EQ(oplogEntry, insertDocumentDoc);
}
TEST_F(
InitialSyncerTest,
InitialSyncerReturnsCallbackCanceledAndDoesNotScheduleRollbackCheckerIfShutdownAfterInsertingInsertOplogSeedDocument) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
NamespaceString insertDocumentNss;
BSONObj insertDocumentDoc;
_storageInterface->insertDocumentFn = [initialSyncer, &insertDocumentDoc, &insertDocumentNss](
OperationContext*, const NamespaceString& nss, const BSONObj& doc) {
insertDocumentNss = nss;
insertDocumentDoc = doc;
initialSyncer->shutdown();
return Status::OK();
};
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
ASSERT_EQUALS(_options.localOplogNS, insertDocumentNss);
ASSERT_BSONOBJ_EQ(oplogEntry, insertDocumentDoc);
}
TEST_F(
InitialSyncerTest,
InitialSyncerPassesThroughRollbackCheckerScheduleErrorAfterCloningFinishesWithNoOperationsToApply) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Make the second replSetGetRBID command fail. Allow all other requests to be scheduled.
executor::RemoteCommandRequest request;
bool first = true;
_shouldFailRequest = [&first, &request](const executor::RemoteCommandRequest& requestToSend) {
if ("replSetGetRBID" == requestToSend.cmdObj.firstElement().fieldNameStringData()) {
if (first) {
first = false;
return false;
}
request = requestToSend;
return true;
}
return false;
};
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(
InitialSyncerTest,
InitialSyncerPassesThroughRollbackCheckerCallbackErrorAfterCloningFinishesWithNoOperationsToApply) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// Last rollback checker replSetGetRBID command.
assertRemoteCommandNameEquals(
"replSetGetRBID",
net->scheduleErrorResponse(
Status(ErrorCodes::OperationFailed, "replSetGetRBID command failed")));
net->runReadyNetworkOperations();
// _rollbackCheckerCheckForRollbackCallback() will shut down the OplogFetcher after setting
// the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerCancelsLastRollbackCheckerOnShutdown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// Last rollback checker replSetGetRBID command.
noi = net->getNextReadyRequest();
assertRemoteCommandNameEquals("replSetGetRBID", noi->getRequest());
net->blackHole(noi);
// _rollbackCheckerCheckForRollbackCallback() will shut down the OplogFetcher after setting
// the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
ASSERT_OK(initialSyncer->shutdown());
executor::NetworkInterfaceMock::InNetworkGuard(net)->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerCancelsLastRollbackCheckerOnOplogFetcherCallbackError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(1));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// Save request for OplogFetcher's oplog tailing query. This request will be canceled.
auto noi = net->getNextReadyRequest();
auto request = assertRemoteCommandNameEquals("find", noi->getRequest());
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
auto oplogFetcherNetworkOperationIterator = noi;
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// Last rollback checker replSetGetRBID command.
noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("replSetGetRBID", request);
net->blackHole(noi);
// Make oplog fetcher fail.
net->scheduleErrorResponse(oplogFetcherNetworkOperationIterator,
Status(ErrorCodes::OperationFailed, "oplog fetcher failed"));
net->runReadyNetworkOperations();
// _oplogFetcherCallback() will shut down the last rollback checker after setting the
// completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _rollbackCheckerCheckForRollbackCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsUnrecoverableRollbackErrorIfSyncSourceRolledBackAfterCloningData) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// Last rollback checker replSetGetRBID command.
request = net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId + 1));
net->runReadyNetworkOperations();
assertRemoteCommandNameEquals("replSetGetRBID", request);
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, _lastApplied);
}
TEST_F(InitialSyncerTest, LastOpTimeShouldBeSetEvenIfNoOperationsAreAppliedAfterCloning) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
auto oplogEntry = makeOplogEntry(1);
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Instead of fast forwarding to DatabasesCloner completion by returning an empty list of
// database names, we'll simulate copying a single database with a single collection on the
// sync source.
NamespaceString nss("a.a");
request = net->scheduleSuccessfulResponse(makeListDatabasesResponse({nss.db().toString()}));
assertRemoteCommandNameEquals("listDatabases", request);
net->runReadyNetworkOperations();
// listCollections for "a"
request = net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, nss, {BSON("name" << nss.coll() << "options" << BSONObj())}));
assertRemoteCommandNameEquals("listCollections", request);
// count:a
request = assertRemoteCommandNameEquals(
"count", net->scheduleSuccessfulResponse(BSON("n" << 1 << "ok" << 1)));
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// listIndexes:a
request = assertRemoteCommandNameEquals(
"listIndexes",
net->scheduleSuccessfulResponse(makeCursorResponse(
0LL,
NamespaceString(nss.getCommandNS()),
{BSON("v" << OplogEntry::kOplogVersion << "key" << BSON("_id" << 1) << "name"
<< "_id_"
<< "ns"
<< nss.ns())})));
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// find:a
request = assertRemoteCommandNameEquals("find",
net->scheduleSuccessfulResponse(makeCursorResponse(
0LL, nss, {BSON("_id" << 1 << "a" << 1)})));
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({oplogEntry});
// Last rollback checker replSetGetRBID command.
request = assertRemoteCommandNameEquals(
"replSetGetRBID",
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId)));
net->runReadyNetworkOperations();
// Deliver cancellation to OplogFetcher.
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(OplogEntry(oplogEntry).getOpTime(), unittest::assertGet(_lastApplied).opTime);
ASSERT_EQUALS(oplogEntry["h"].Long(), unittest::assertGet(_lastApplied).value);
ASSERT_FALSE(_storageInterface->getInitialSyncFlag(opCtx.get()));
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetNextApplierBatchScheduleError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Before processing scheduled last oplog entry fetcher response, set flag in
// TaskExecutorMock so that InitialSyncer will fail to schedule
// _getNextApplierBatchCallback().
_executorProxy->shouldFailScheduleWork = true;
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughSecondGetNextApplierBatchScheduleError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Before processing scheduled last oplog entry fetcher response, set flag in
// TaskExecutorMock so that InitialSyncer will fail to schedule second
// _getNextApplierBatchCallback() at (now + options.getApplierBatchCallbackRetryWait).
_executorProxy->shouldFailScheduleWorkAt = true;
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// _lastOplogEntryFetcherCallbackAfterCloningData() will shut down the OplogFetcher after
// setting the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerCancelsGetNextApplierBatchOnShutdown) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// We do not have to respond to the OplogFetcher's oplog tailing query. Blackhole and move
// on to the DatabasesCloner's request.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("tailable"));
net->blackHole(noi);
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// Since we black holed OplogFetcher's find request, _getNextApplierBatch_inlock() will
// not return any operations for us to apply, leading to _getNextApplierBatchCallback()
// rescheduling itself at new->now() + _options.getApplierBatchCallbackRetryWait.
}
ASSERT_OK(initialSyncer->shutdown());
executor::NetworkInterfaceMock::InNetworkGuard(net)->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetNextApplierBatchInLockError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
// _getNextApplierBatch_inlock() returns BadValue when it gets an oplog entry with an unexpected
// version (not OplogEntry::kOplogVersion).
auto oplogEntry = makeOplogEntry(1);
auto oplogEntryWithInconsistentVersion =
makeOplogEntry(2, "i", OplogEntry::kOplogVersion + 100);
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Return bad oplog entry that will be added to the
// oplog buffer and processed by _getNextApplierBatch_inlock().
auto request = assertRemoteCommandNameEquals(
"find",
net->scheduleSuccessfulResponse(makeCursorResponse(
1LL, _options.localOplogNS, {oplogEntry, oplogEntryWithInconsistentVersion})));
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// OplogFetcher's getMore request. Black hole because we already got our bad oplog entry
// into the oplog buffer.
auto noi = net->getNextReadyRequest();
assertRemoteCommandNameEquals("getMore", noi->getRequest());
net->blackHole(noi);
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// _getNextApplierBatchCallback() will shut down the OplogFetcher after setting the
// completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::BadValue, _lastApplied);
}
TEST_F(
InitialSyncerTest,
InitialSyncerReturnsEmptyBatchFromGetNextApplierBatchInLockIfRsSyncApplyStopFailPointIsEnabled) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
// _getNextApplierBatch_inlock() returns BadValue when it gets an oplog entry with an unexpected
// version (not OplogEntry::kOplogVersion).
auto oplogEntry = makeOplogEntry(1);
auto oplogEntryWithInconsistentVersion =
makeOplogEntry(2, "i", OplogEntry::kOplogVersion + 100);
// Enable 'rsSyncApplyStop' so that _getNextApplierBatch_inlock() returns an empty batch of
// operations instead of a batch containing an oplog entry with a bad version.
auto failPoint = getGlobalFailPointRegistry()->getFailPoint("rsSyncApplyStop");
failPoint->setMode(FailPoint::alwaysOn);
ON_BLOCK_EXIT([failPoint]() { failPoint->setMode(FailPoint::off); });
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Return bad oplog entry that will be added to the
// oplog buffer and processed by _getNextApplierBatch_inlock().
auto request = net->scheduleSuccessfulResponse(makeCursorResponse(
1LL, _options.localOplogNS, {oplogEntry, oplogEntryWithInconsistentVersion}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// OplogFetcher's getMore request. Black hole because we already got our bad oplog entry
// into the oplog buffer.
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
net->blackHole(noi);
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// Since the 'rsSyncApplyStop' fail point is enabled, InitialSyncer will get an empty
// batch of operations from _getNextApplierBatch_inlock() even though the oplog buffer
// is not empty.
}
// If the fail point is not working, the initial sync status will be set to BadValue (due to the
// bad oplog entry in the oplog buffer) and shutdown() will not be able to overwrite this status
// with CallbackCanceled.
// Otherwise, shutdown() will cancel both the OplogFetcher and the scheduled
// _getNextApplierBatchCallback() task. The final initial sync status will be CallbackCanceled.
ASSERT_OK(initialSyncer->shutdown());
executor::NetworkInterfaceMock::InNetworkGuard(net)->runReadyNetworkOperations();
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::CallbackCanceled, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsNoSuchKeyIfApplierBatchContainsAnOplogEntryWithoutHash) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
// This oplog entry (without a required "h" field) will be read by OplogFetcher and inserted
// into OplogBuffer to be retrieved by _getNextApplierBatch_inlock().
auto oplogEntryWithoutHash = BSON("ts" << Timestamp(2, 2) << "v" << OplogEntry::kOplogVersion);
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Save for later.
auto request = net->scheduleSuccessfulResponse(makeCursorResponse(
1LL, _options.localOplogNS, {makeOplogEntry(1), oplogEntryWithoutHash}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Ignore OplogFetcher's getMore request.
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
net->blackHole(noi);
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// _getNextApplierBatchCallback() will shut down the OplogFetcher after setting the
// completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::NoSuchKey, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughMultiApplierScheduleError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Save for later.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
auto oplogFetcherNoi = noi;
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// _getNextApplierBatchCallback() should have rescheduled itself.
// We'll insert some operations in the oplog buffer so that we'll attempt to schedule
// MultiApplier next time _getNextApplierBatchCallback() runs.
net->scheduleSuccessfulResponse(
oplogFetcherNoi,
makeCursorResponse(1LL, _options.localOplogNS, {makeOplogEntry(1), makeOplogEntry(2)}));
net->runReadyNetworkOperations();
// Ignore OplogFetcher's getMore request.
noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
// Make MultiApplier::startup() fail.
_executorProxy->shouldFailScheduleWork = true;
// Advance clock until _getNextApplierBatchCallback() runs.
auto when = net->now() + _options.getApplierBatchCallbackRetryWait;
ASSERT_EQUALS(when, net->runUntil(when));
// _getNextApplierBatchCallback() will shut down the OplogFetcher after setting the
// completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerPassesThroughMultiApplierCallbackError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
getExternalState()->multiApplyFn =
[](OperationContext*, const MultiApplier::Operations&, MultiApplier::ApplyOperationFn) {
return Status(ErrorCodes::OperationFailed, "multiApply failed");
};
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Provide enough operations to trigger MultiApplier.
auto request = net->scheduleSuccessfulResponse(
makeCursorResponse(1LL, _options.localOplogNS, {makeOplogEntry(1), makeOplogEntry(2)}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Ignore OplogFetcher's getMore request.
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// _multiApplierCallback() will shut down the OplogFetcher after setting the completion
// status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest, InitialSyncerCancelsGetNextApplierBatchCallbackOnOplogFetcherError) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Save for later.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
auto oplogFetcherNoi = noi;
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// Send error to _oplogFetcherCallback().
net->scheduleErrorResponse(oplogFetcherNoi,
Status(ErrorCodes::OperationFailed, "oplog fetcher failed"));
// _oplogFetcherCallback() will cancel the _getNextApplierBatchCallback() task after setting
// the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsLastAppliedOnReachingStopTimestampAfterApplyingOneBatch) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto lastOp = makeOplogEntry(2);
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Response has enough operations to reach
// end timestamp.
auto request = net->scheduleSuccessfulResponse(
makeCursorResponse(1LL, _options.localOplogNS, {makeOplogEntry(1), lastOp}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Black hole OplogFetcher's getMore request.
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
net->blackHole(noi);
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({lastOp});
// Last rollback ID.
request = net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
assertRemoteCommandNameEquals("replSetGetRBID", request);
net->runReadyNetworkOperations();
// _multiApplierCallback() will cancel the _getNextApplierBatchCallback() task after setting
// the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(OplogEntry(lastOp).getOpTime(), unittest::assertGet(_lastApplied).opTime);
ASSERT_EQUALS(lastOp["h"].Long(), unittest::assertGet(_lastApplied).value);
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsLastAppliedOnReachingStopTimestampAfterApplyingMultipleBatches) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
// To make InitialSyncer apply multiple batches, we make the third and last operation a command
// so that it will go into a separate batch from the second operation. First operation is the
// last fetched entry before data cloning and is not applied.
auto lastOp = makeOplogEntry(3, "c");
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Response has enough operations to reach
// end timestamp.
auto request = net->scheduleSuccessfulResponse(makeCursorResponse(
1LL, _options.localOplogNS, {makeOplogEntry(1), makeOplogEntry(2), lastOp}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Instead of fast forwarding to DatabasesCloner completion by returning an empty list of
// database names, we'll simulate copying a single database with a single collection on the
// sync source.
NamespaceString nss("a.a");
request = net->scheduleSuccessfulResponse(makeListDatabasesResponse({nss.db().toString()}));
assertRemoteCommandNameEquals("listDatabases", request);
net->runReadyNetworkOperations();
// Black hole OplogFetcher's getMore request.
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
net->blackHole(noi);
// listCollections for "a"
request = net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, nss, {BSON("name" << nss.coll() << "options" << BSONObj())}));
assertRemoteCommandNameEquals("listCollections", request);
// count:a
request = net->scheduleSuccessfulResponse(BSON("n" << 1 << "ok" << 1));
assertRemoteCommandNameEquals("count", request);
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// listIndexes:a
request = net->scheduleSuccessfulResponse(makeCursorResponse(
0LL,
NamespaceString(nss.getCommandNS()),
{BSON("v" << OplogEntry::kOplogVersion << "key" << BSON("_id" << 1) << "name"
<< "_id_"
<< "ns"
<< nss.ns())}));
assertRemoteCommandNameEquals("listIndexes", request);
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// find:a
request = net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, nss, {BSON("_id" << 1 << "a" << 1)}));
assertRemoteCommandNameEquals("find", request);
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// Second last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({lastOp});
// Last rollback ID.
request = net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
assertRemoteCommandNameEquals("replSetGetRBID", request);
net->runReadyNetworkOperations();
// _multiApplierCallback() will cancel the _getNextApplierBatchCallback() task after setting
// the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(OplogEntry(lastOp).getOpTime(), unittest::assertGet(_lastApplied).opTime);
ASSERT_EQUALS(lastOp["h"].Long(), unittest::assertGet(_lastApplied).value);
}
TEST_F(
InitialSyncerTest,
InitialSyncerSchedulesLastOplogEntryFetcherToGetNewStopTimestampIfMissingDocumentsHaveBeenFetchedDuringMultiInitialSyncApply) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// Override DataReplicatorExternalState::_multiInitialSyncApply() so that it will also fetch a
// missing document.
// This forces InitialSyncer to evaluate its end timestamp for applying operations after each
// batch.
getExternalState()->multiApplyFn = [](OperationContext*,
const MultiApplier::Operations& ops,
MultiApplier::ApplyOperationFn applyOperation) {
// 'OperationPtr*' is ignored by our overridden _multiInitialSyncApply().
applyOperation(nullptr);
return ops.back().getOpTime();
};
bool fetchCountIncremented = false;
getExternalState()->multiInitialSyncApplyFn = [&fetchCountIncremented](
MultiApplier::OperationPtrs*, const HostAndPort&, AtomicUInt32* fetchCount) {
if (!fetchCountIncremented) {
fetchCount->addAndFetch(1);
fetchCountIncremented = true;
}
return Status::OK();
};
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
// Use command for third and last operation to ensure we have two batches to apply.
auto lastOp = makeOplogEntry(3, "c");
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query. Response has enough operations to reach
// end timestamp.
auto request = net->scheduleSuccessfulResponse(makeCursorResponse(
1LL, _options.localOplogNS, {makeOplogEntry(1), makeOplogEntry(2), lastOp}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Quickest path to a successful DatabasesCloner completion is to respond to the
// listDatabases with an empty list of database names.
assertRemoteCommandNameEquals(
"listDatabases", net->scheduleSuccessfulResponse(makeListDatabasesResponse({})));
net->runReadyNetworkOperations();
// Black hole OplogFetcher's getMore request.
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
net->blackHole(noi);
// Second last oplog entry fetcher.
// Send oplog entry with timestamp 2. InitialSyncer will update this end timestamp after
// applying the first batch.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(2)});
// Third last oplog entry fetcher.
processSuccessfulLastOplogEntryFetcherResponse({lastOp});
// Last rollback ID.
request = net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
assertRemoteCommandNameEquals("replSetGetRBID", request);
net->runReadyNetworkOperations();
// _multiApplierCallback() will cancel the _getNextApplierBatchCallback() task after setting
// the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(OplogEntry(lastOp).getOpTime(), unittest::assertGet(_lastApplied).opTime);
ASSERT_EQUALS(lastOp["h"].Long(), unittest::assertGet(_lastApplied).value);
ASSERT_TRUE(fetchCountIncremented);
auto progress = initialSyncer->getInitialSyncProgress();
log() << "Progress after failed initial sync attempt: " << progress;
ASSERT_EQUALS(1, progress.getIntField("fetchedMissingDocs")) << progress;
}
TEST_F(InitialSyncerTest,
InitialSyncerReturnsInvalidSyncSourceWhenFailInitialSyncWithBadHostFailpointIsEnabled) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
// This fail point makes chooseSyncSourceCallback fail with an InvalidSyncSource error.
auto failPoint = getGlobalFailPointRegistry()->getFailPoint("failInitialSyncWithBadHost");
failPoint->setMode(FailPoint::alwaysOn);
ON_BLOCK_EXIT([failPoint]() { failPoint->setMode(FailPoint::off); });
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::InvalidSyncSource, _lastApplied);
}
TEST_F(InitialSyncerTest, OplogOutOfOrderOnOplogFetchFinish) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
auto net = getNet();
int baseRollbackId = 1;
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
// OplogFetcher's oplog tailing query.
auto request = net->scheduleSuccessfulResponse(
makeCursorResponse(1LL, _options.localOplogNS, {makeOplogEntry(1)}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// Ignore listDatabases request.
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("listDatabases", request);
net->blackHole(noi);
// Ensure that OplogFetcher fails with an OplogOutOfOrder error by responding to the getMore
// request with oplog entries containing the following timestamps (most recently processed
// oplog entry has a timestamp of 1):
// (last=1), 5, 4
request = net->scheduleSuccessfulResponse(makeCursorResponse(
1LL, _options.localOplogNS, {makeOplogEntry(5), makeOplogEntry(4)}, false));
assertRemoteCommandNameEquals("getMore", request);
net->runReadyNetworkOperations();
// Deliver cancellation signal to DatabasesCloner.
net->runReadyNetworkOperations();
}
initialSyncer->join();
ASSERT_EQUALS(ErrorCodes::OplogOutOfOrder, _lastApplied);
}
TEST_F(InitialSyncerTest, GetInitialSyncProgressReturnsCorrectProgress) {
auto initialSyncer = &getInitialSyncer();
auto opCtx = makeOpCtx();
_syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 27017));
ASSERT_OK(initialSyncer->startup(opCtx.get(), 2U));
auto net = getNet();
int baseRollbackId = 1;
// Play first 2 responses to ensure initial syncer has started the oplog fetcher.
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Base rollback ID.
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
}
log() << "Done playing first failed response";
auto progress = initialSyncer->getInitialSyncProgress();
log() << "Progress after first failed response: " << progress;
ASSERT_EQUALS(progress.nFields(), 8) << progress;
ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), 0) << progress;
ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), 2) << progress;
ASSERT_EQUALS(progress["initialSyncStart"].type(), Date) << progress;
ASSERT_EQUALS(progress["initialSyncOplogStart"].timestamp(), Timestamp(1, 1)) << progress;
ASSERT_BSONOBJ_EQ(progress.getObjectField("initialSyncAttempts"), BSONObj());
ASSERT_EQUALS(progress.getIntField("fetchedMissingDocs"), 0) << progress;
ASSERT_EQUALS(progress.getIntField("appliedOps"), 0) << progress;
ASSERT_BSONOBJ_EQ(progress.getObjectField("databases"), BSON("databasesCloned" << 0));
// Play rest of the failed round of responses.
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Ignore oplog tailing query.
auto noi = net->getNextReadyRequest();
auto request = noi->getRequest();
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->blackHole(noi);
request = net->scheduleErrorResponse(
Status(ErrorCodes::FailedToParse, "fail on clone -- listDBs injected failure"));
assertRemoteCommandNameEquals("listDatabases", request);
net->runReadyNetworkOperations();
// Deliver cancellation to OplogFetcher
net->runReadyNetworkOperations();
}
log() << "Done playing failed responses";
// Play the first 2 responses of the successful round of responses to ensure that the
// initial syncer starts the oplog fetcher.
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
auto when = net->now() + _options.initialSyncRetryWait;
ASSERT_EQUALS(when, net->runUntil(when));
// Base rollback ID.
auto request = net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId));
assertRemoteCommandNameEquals("replSetGetRBID", request);
net->runReadyNetworkOperations();
// Last oplog entry.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(1)});
}
log() << "Done playing first successful response";
progress = initialSyncer->getInitialSyncProgress();
log() << "Progress after failure: " << progress;
ASSERT_EQUALS(progress.nFields(), 8) << progress;
ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), 1) << progress;
ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), 2) << progress;
ASSERT_EQUALS(progress["initialSyncStart"].type(), Date) << progress;
ASSERT_EQUALS(progress["initialSyncOplogStart"].timestamp(), Timestamp(1, 1)) << progress;
ASSERT_EQUALS(progress.getIntField("fetchedMissingDocs"), 0) << progress;
ASSERT_EQUALS(progress.getIntField("appliedOps"), 0) << progress;
ASSERT_BSONOBJ_EQ(progress.getObjectField("databases"), BSON("databasesCloned" << 0));
BSONObj attempts = progress["initialSyncAttempts"].Obj();
ASSERT_EQUALS(attempts.nFields(), 1) << attempts;
BSONObj attempt0 = attempts["0"].Obj();
ASSERT_EQUALS(attempt0.nFields(), 3) << attempt0;
ASSERT_EQUALS(
attempt0.getStringField("status"),
std::string(
"FailedToParse: error cloning databases: fail on clone -- listDBs injected failure"))
<< attempt0;
ASSERT_EQUALS(attempt0["durationMillis"].type(), NumberInt) << attempt0;
ASSERT_EQUALS(attempt0.getStringField("syncSource"), std::string("localhost:27017"))
<< attempt0;
// Play all but last of the successful round of responses.
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Ignore oplog tailing query.
auto request = net->scheduleSuccessfulResponse(makeCursorResponse(1LL,
_options.localOplogNS,
{makeOplogEntry(1),
makeOplogEntry(2),
makeOplogEntry(3),
makeOplogEntry(4),
makeOplogEntry(5),
makeOplogEntry(6),
makeOplogEntry(7)
}));
assertRemoteCommandNameEquals("find", request);
ASSERT_TRUE(request.cmdObj.getBoolField("oplogReplay"));
net->runReadyNetworkOperations();
// listDatabases
NamespaceString nss("a.a");
request = net->scheduleSuccessfulResponse(makeListDatabasesResponse({nss.db().toString()}));
assertRemoteCommandNameEquals("listDatabases", request);
net->runReadyNetworkOperations();
auto noi = net->getNextReadyRequest();
request = noi->getRequest();
assertRemoteCommandNameEquals("getMore", request);
net->blackHole(noi);
// listCollections for "a"
request = net->scheduleSuccessfulResponse(
makeCursorResponse(0LL, nss, {BSON("name" << nss.coll() << "options" << BSONObj())}));
assertRemoteCommandNameEquals("listCollections", request);
// count:a
request = net->scheduleSuccessfulResponse(BSON("n" << 5 << "ok" << 1));
assertRemoteCommandNameEquals("count", request);
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// listIndexes:a
request = net->scheduleSuccessfulResponse(makeCursorResponse(
0LL,
NamespaceString(nss.getCommandNS()),
{BSON("v" << OplogEntry::kOplogVersion << "key" << BSON("_id" << 1) << "name"
<< "_id_"
<< "ns"
<< nss.ns())}));
assertRemoteCommandNameEquals("listIndexes", request);
ASSERT_EQUALS(nss.coll(), request.cmdObj.firstElement().String());
ASSERT_EQUALS(nss.db(), request.dbname);
// find:a - 5 batches
for (int i = 1; i <= 5; ++i) {
request = net->scheduleSuccessfulResponse(
makeCursorResponse(i < 5 ? 2LL : 0LL, nss, {BSON("_id" << i << "a" << i)}, i == 1));
ASSERT_EQUALS(i == 1 ? "find" : "getMore",
request.cmdObj.firstElement().fieldNameStringData());
net->runReadyNetworkOperations();
}
// Second last oplog entry fetcher.
// Send oplog entry with timestamp 2. InitialSyncer will update this end timestamp after
// applying the first batch.
processSuccessfulLastOplogEntryFetcherResponse({makeOplogEntry(7)});
}
log() << "Done playing all but last successful response";
progress = initialSyncer->getInitialSyncProgress();
log() << "Progress after all but last successful response: " << progress;
ASSERT_EQUALS(progress.nFields(), 9) << progress;
ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), 1) << progress;
ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), 2) << progress;
ASSERT_EQUALS(progress["initialSyncOplogStart"].timestamp(), Timestamp(1, 1)) << progress;
ASSERT_EQUALS(progress["initialSyncOplogEnd"].timestamp(), Timestamp(7, 1)) << progress;
ASSERT_EQUALS(progress["initialSyncStart"].type(), Date) << progress;
ASSERT_EQUALS(progress.getIntField("fetchedMissingDocs"), 0) << progress;
// Expected applied ops to be a superset of this range: Timestamp(2,1) ... Timestamp(7,1).
ASSERT_GREATER_THAN_OR_EQUALS(progress.getIntField("appliedOps"), 6) << progress;
auto databasesProgress = progress.getObjectField("databases");
ASSERT_EQUALS(1, databasesProgress.getIntField("databasesCloned")) << databasesProgress;
auto dbProgress = databasesProgress.getObjectField("a");
ASSERT_EQUALS(1, dbProgress.getIntField("collections")) << dbProgress;
ASSERT_EQUALS(1, dbProgress.getIntField("clonedCollections")) << dbProgress;
auto collectionProgress = dbProgress.getObjectField("a.a");
ASSERT_EQUALS(
5, collectionProgress.getIntField(CollectionCloner::Stats::kDocumentsToCopyFieldName))
<< collectionProgress;
ASSERT_EQUALS(
5, collectionProgress.getIntField(CollectionCloner::Stats::kDocumentsCopiedFieldName))
<< collectionProgress;
ASSERT_EQUALS(1, collectionProgress.getIntField("indexes")) << collectionProgress;
ASSERT_EQUALS(5, collectionProgress.getIntField("fetchedBatches")) << collectionProgress;
attempts = progress["initialSyncAttempts"].Obj();
ASSERT_EQUALS(attempts.nFields(), 1) << progress;
attempt0 = attempts["0"].Obj();
ASSERT_EQUALS(attempt0.nFields(), 3) << attempt0;
ASSERT_EQUALS(
attempt0.getStringField("status"),
std::string(
"FailedToParse: error cloning databases: fail on clone -- listDBs injected failure"))
<< attempt0;
ASSERT_EQUALS(attempt0["durationMillis"].type(), NumberInt) << attempt0;
ASSERT_EQUALS(attempt0.getStringField("syncSource"), std::string("localhost:27017"))
<< attempt0;
// Play last successful response.
{
executor::NetworkInterfaceMock::InNetworkGuard guard(net);
// Last rollback ID.
assertRemoteCommandNameEquals(
"replSetGetRBID",
net->scheduleSuccessfulResponse(makeRollbackCheckerResponse(baseRollbackId)));
net->runReadyNetworkOperations();
// _multiApplierCallback() will cancel the _getNextApplierBatchCallback() task after setting
// the completion status.
// We call runReadyNetworkOperations() again to deliver the cancellation status to
// _oplogFetcherCallback().
net->runReadyNetworkOperations();
}
log() << "waiting for initial sync to verify it completed OK";
initialSyncer->join();
ASSERT_EQUALS(OplogEntry(makeOplogEntry(7)).getOpTime(),
unittest::assertGet(_lastApplied).opTime);
progress = initialSyncer->getInitialSyncProgress();
log() << "Progress at end: " << progress;
ASSERT_EQUALS(progress.nFields(), 11) << progress;
ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), 1) << progress;
ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), 2) << progress;
ASSERT_EQUALS(progress["initialSyncStart"].type(), Date) << progress;
ASSERT_EQUALS(progress["initialSyncEnd"].type(), Date) << progress;
ASSERT_EQUALS(progress["initialSyncOplogStart"].timestamp(), Timestamp(1, 1)) << progress;
ASSERT_EQUALS(progress["initialSyncOplogEnd"].timestamp(), Timestamp(7, 1)) << progress;
ASSERT_EQUALS(progress["initialSyncElapsedMillis"].type(), NumberInt) << progress;
ASSERT_EQUALS(progress.getIntField("fetchedMissingDocs"), 0) << progress;
// Expected applied ops to be a superset of this range: Timestamp(2,1) ... Timestamp(7,1).
ASSERT_GREATER_THAN_OR_EQUALS(progress.getIntField("appliedOps"), 6) << progress;
attempts = progress["initialSyncAttempts"].Obj();
ASSERT_EQUALS(attempts.nFields(), 2) << attempts;
attempt0 = attempts["0"].Obj();
ASSERT_EQUALS(attempt0.nFields(), 3) << attempt0;
ASSERT_EQUALS(
attempt0.getStringField("status"),
std::string(
"FailedToParse: error cloning databases: fail on clone -- listDBs injected failure"))
<< attempt0;
ASSERT_EQUALS(attempt0["durationMillis"].type(), NumberInt) << attempt0;
ASSERT_EQUALS(attempt0.getStringField("syncSource"), std::string("localhost:27017"))
<< attempt0;
BSONObj attempt1 = attempts["1"].Obj();
ASSERT_EQUALS(attempt1.nFields(), 3) << attempt1;
ASSERT_EQUALS(attempt1.getStringField("status"), std::string("OK")) << attempt1;
ASSERT_EQUALS(attempt1["durationMillis"].type(), NumberInt) << attempt1;
ASSERT_EQUALS(attempt1.getStringField("syncSource"), std::string("localhost:27017"))
<< attempt1;
}
} // namespace