<|start_filename|>jstests/aggregation/bugs/lookup_unwind_getmore.js<|end_filename|>
/**
 * Tests that the server correctly handles when the OperationContext of the DBDirectClient used by
 * the $lookup stage changes as it unwinds the results.
 *
 * This test was designed to reproduce SERVER-22537.
 */
(function() {
    'use strict';

    // We use a batch size of 2 to ensure that the mongo shell issues a getMore when unwinding the
    // results from the 'dest' collection for the same document in the 'source' collection under a
    // different OperationContext.
    const batchSize = 2;

    const conn =
        MongoRunner.runMongod({setParameter: "internalAggregationLookupBatchSize=" + batchSize});
    assert.neq(null, conn, "mongod failed to start up");
    const testDB = conn.getDB("test");

    testDB.source.drop();
    testDB.dest.drop();

    assert.writeOK(testDB.source.insert({local: 1}));

    // The cursor batching logic actually requests one more document than it needs to fill the
    // first batch, so if we want to leave the $lookup stage paused with a cursor open we'll
    // need two more matching documents than the batch size.
    const numMatches = batchSize + 2;
    for (var i = 0; i < numMatches; ++i) {
        assert.writeOK(testDB.dest.insert({foreign: 1}));
    }

    var res = testDB.runCommand({
        aggregate: 'source',
        pipeline: [
            {
              $lookup: {
                  from: 'dest',
                  localField: 'local',
                  foreignField: 'foreign',
                  as: 'matches',
              }
            },
            {
              $unwind: {
                  path: '$matches',
              },
            },
        ],
        cursor: {
            batchSize: batchSize,
        },
    });
    assert.commandWorked(res);

    var cursor = new DBCommandCursor(conn, res, batchSize);
    assert.eq(numMatches, cursor.itcount());

    assert.eq(0, MongoRunner.stopMongod(conn), "expected mongod to shutdown cleanly");
})();


<|start_filename|>src/mongo/s/s_only.cpp<|end_filename|>
// s_only.cpp

/*    Copyright 2009 10gen 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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding

#include "mongo/platform/basic.h"

#include <tuple>

#include "mongo/db/auth/authorization_manager.h"
#include "mongo/db/auth/authorization_manager_global.h"
#include "mongo/db/auth/authorization_session.h"
#include "mongo/db/client.h"
#include "mongo/db/commands.h"
#include "mongo/db/service_context.h"
#include "mongo/db/stats/counters.h"
#include "mongo/rpc/metadata.h"
#include "mongo/rpc/reply_builder_interface.h"
#include "mongo/rpc/request_interface.h"
#include "mongo/s/cluster_last_error_info.h"
#include "mongo/util/assert_util.h"
#include "mongo/util/concurrency/thread_name.h"
#include "mongo/util/log.h"

namespace mongo {

using std::string;
using std::stringstream;


bool isMongos() {
    return true;
}

/** When this callback is run, we record a shard that we've used for useful work
 *  in an operation to be read later by getLastError()
*/
void usingAShardConnection(const std::string& addr) {
    ClusterLastErrorInfo::get(cc()).addShardHost(addr);
}

// called into by the web server. For now we just translate the parameters
// to their old style equivalents.
void Command::execCommand(OperationContext* txn,
                          Command* command,
                          const rpc::RequestInterface& request,
                          rpc::ReplyBuilderInterface* replyBuilder) {
    int queryFlags = 0;
    BSONObj cmdObj;

    std::tie(cmdObj, queryFlags) = uassertStatusOK(
        rpc::downconvertRequestMetadata(request.getCommandArgs(), request.getMetadata()));

    std::string db = request.getDatabase().rawData();
    BSONObjBuilder result;

    execCommandClientBasic(txn,
                           command,
                           *txn->getClient(),
                           queryFlags,
                           request.getDatabase().rawData(),
                           cmdObj,
                           result);

    replyBuilder->setCommandReply(result.done()).setMetadata(rpc::makeEmptyMetadata());
}

void Command::execCommandClientBasic(OperationContext* txn,
                                     Command* c,
                                     ClientBasic& client,
                                     int queryOptions,
                                     const char* ns,
                                     BSONObj& cmdObj,
                                     BSONObjBuilder& result) {
    std::string dbname = nsToDatabase(ns);

    if (cmdObj.getBoolField("help")) {
        stringstream help;
        help << "help for: " << c->name << " ";
        c->help(help);
        result.append("help", help.str());
        result.append("lockType", c->isWriteCommandForConfigServer() ? 1 : 0);
        appendCommandStatus(result, true, "");
        return;
    }

    Status status = checkAuthorization(c, txn, dbname, cmdObj);
    if (!status.isOK()) {
        appendCommandStatus(result, status);
        return;
    }

    c->_commandsExecuted.increment();

    if (c->shouldAffectCommandCounter()) {
        globalOpCounters.gotCommand();
    }

    std::string errmsg;
    bool ok = false;
    try {
        ok = c->run(txn, dbname, cmdObj, queryOptions, errmsg, result);
    } catch (const DBException& e) {
        result.resetToEmpty();
        const int code = e.getCode();

        // Codes for StaleConfigException
        if (code == ErrorCodes::RecvStaleConfig || code == ErrorCodes::SendStaleConfig) {
            throw;
        }

        errmsg = e.what();
        result.append("code", code);
    }

    if (!ok) {
        c->_commandsFailed.increment();
    }

    appendCommandStatus(result, ok, errmsg);
}

void Command::runAgainstRegistered(OperationContext* txn,
                                   const char* ns,
                                   BSONObj& jsobj,
                                   BSONObjBuilder& anObjBuilder,
                                   int queryOptions) {
    // It should be impossible for this uassert to fail since there should be no way to get
    // into this function with any other collection name.
    uassert(16618,
            "Illegal attempt to run a command against a namespace other than $cmd.",
            nsToCollectionSubstring(ns) == "$cmd");

    BSONElement e = jsobj.firstElement();
    std::string commandName = e.fieldName();
    Command* c = e.type() ? Command::findCommand(commandName) : NULL;
    if (!c) {
        Command::appendCommandStatus(
            anObjBuilder, false, str::stream() << "no such cmd: " << commandName);
        anObjBuilder.append("code", ErrorCodes::CommandNotFound);
        Command::unknownCommands.increment();
        return;
    }

    execCommandClientBasic(txn, c, cc(), queryOptions, ns, jsobj, anObjBuilder);
}

void Command::registerError(OperationContext* txn, const DBException& exception) {}

}  // namespace mongo


<|start_filename|>src/mongo/executor/connection_pool_asio_integration_test.cpp<|end_filename|>
/**
 *    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 <http://www.gnu.org/licenses/>.
 *
 *    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 "mongo/executor/connection_pool_asio.h"

#include "mongo/client/connection_string.h"
#include "mongo/executor/async_stream_factory.h"
#include "mongo/executor/async_timer_asio.h"
#include "mongo/executor/connection_pool_stats.h"
#include "mongo/executor/network_connection_hook.h"
#include "mongo/executor/network_interface_asio_test_utils.h"
#include "mongo/rpc/get_status_from_command_result.h"
#include "mongo/stdx/future.h"
#include "mongo/stdx/memory.h"
#include "mongo/unittest/integration_test.h"
#include "mongo/unittest/unittest.h"
#include "mongo/util/fail_point_service.h"
#include "mongo/util/scopeguard.h"
#include "mongo/util/time_support.h"

namespace mongo {
namespace executor {
namespace {

class MyNetworkConnectionHook : public NetworkConnectionHook {
public:
    Status validateHost(const HostAndPort& remoteHost,
                        const RemoteCommandResponse& isMasterReply) override {
        stdx::unique_lock<stdx::mutex> lk(_mutex);

        _count++;

        return Status::OK();
    }

    StatusWith<boost::optional<RemoteCommandRequest>> makeRequest(
        const HostAndPort& remoteHost) override {
        return StatusWith<boost::optional<RemoteCommandRequest>>(boost::none);
    }

    Status handleReply(const HostAndPort& remoteHost, RemoteCommandResponse&& response) override {
        MONGO_UNREACHABLE;
    }

    static int count() {
        stdx::unique_lock<stdx::mutex> lk(_mutex);

        return _count;
    }

private:
    static stdx::mutex _mutex;
    static int _count;
};

stdx::mutex MyNetworkConnectionHook::_mutex;
int MyNetworkConnectionHook::_count = 0;

TEST(ConnectionPoolASIO, TestPing) {
    auto fixture = unittest::getFixtureConnectionString();

    NetworkInterfaceASIO::Options options;
    options.streamFactory = stdx::make_unique<AsyncStreamFactory>();
    options.networkConnectionHook = stdx::make_unique<MyNetworkConnectionHook>();
    options.connectionPoolOptions.maxConnections = 10;
    options.timerFactory = stdx::make_unique<AsyncTimerFactoryASIO>();
    NetworkInterfaceASIO net{std::move(options)};

    net.startup();
    auto guard = MakeGuard([&] { net.shutdown(); });

    const int N = 50;

    std::array<stdx::thread, N> threads;

    for (auto& thread : threads) {
        thread = stdx::thread([&net, &fixture]() {
            auto status = Status::OK();
            Deferred<StatusWith<RemoteCommandResponse>> deferred;

            net.startCommand(makeCallbackHandle(),
                             RemoteCommandRequest{
                                 fixture.getServers()[0], "admin", BSON("ping" << 1), BSONObj()},
                             [&deferred](StatusWith<RemoteCommandResponse> resp) {
                                 deferred.emplace(std::move(resp));
                             });

            ASSERT_OK(deferred.get().getStatus());
        });
    }

    for (auto&& thread : threads) {
        thread.join();
    }

    ASSERT_LTE(MyNetworkConnectionHook::count(), 10);
}

/**
 * This test verifies a fix for SERVER-22391, where we raced if a new request
 * came in at the same time a host timeout was triggering.
 */
TEST(ConnectionPoolASIO, TestHostTimeoutRace) {
    auto fixture = unittest::getFixtureConnectionString();

    NetworkInterfaceASIO::Options options;
    options.streamFactory = stdx::make_unique<AsyncStreamFactory>();
    options.connectionPoolOptions.hostTimeout = Milliseconds(1);
    options.timerFactory = stdx::make_unique<AsyncTimerFactoryASIO>();
    NetworkInterfaceASIO net{std::move(options)};

    net.startup();
    auto guard = MakeGuard([&] { net.shutdown(); });

    for (int i = 0; i < 1000; i++) {
        Deferred<StatusWith<RemoteCommandResponse>> deferred;
        net.startCommand(
            makeCallbackHandle(),
            RemoteCommandRequest{fixture.getServers()[0], "admin", BSON("ping" << 1), BSONObj()},
            [&](StatusWith<RemoteCommandResponse> resp) { deferred.emplace(std::move(resp)); });

        ASSERT_OK(deferred.get().getStatus());
        sleepmillis(1);
    }
}


/**
 * Verify that a connections that timeout immediately don't invariant.
 */
TEST(ConnectionPoolASIO, ConnSetupTimeout) {
    auto fixture = unittest::getFixtureConnectionString();

    NetworkInterfaceASIO::Options options;
    options.streamFactory = stdx::make_unique<AsyncStreamFactory>();
    options.timerFactory = stdx::make_unique<AsyncTimerFactoryASIO>();
    options.connectionPoolOptions.refreshTimeout = Milliseconds(-2);
    NetworkInterfaceASIO net{std::move(options)};

    net.startup();
    auto guard = MakeGuard([&] { net.shutdown(); });

    Deferred<StatusWith<RemoteCommandResponse>> deferred;
    net.startCommand(
        makeCallbackHandle(),
        RemoteCommandRequest{fixture.getServers()[0], "admin", BSON("ping" << 1), BSONObj()},
        [&](StatusWith<RemoteCommandResponse> resp) { deferred.emplace(std::move(resp)); });

    ASSERT_EQ(deferred.get().getStatus().code(), ErrorCodes::ExceededTimeLimit);
}

/**
 * Verify that connection refreshes actually occur, and that they drop down the totalAvailable
 * correctly.  Verifies SERVER-25006
 */
TEST(ConnectionPoolASIO, ConnRefreshHappens) {
    auto fixture = unittest::getFixtureConnectionString();

    NetworkInterfaceASIO::Options options;
    options.streamFactory = stdx::make_unique<AsyncStreamFactory>();
    options.timerFactory = stdx::make_unique<AsyncTimerFactoryASIO>();
    options.connectionPoolOptions.refreshRequirement = Milliseconds(10);
    NetworkInterfaceASIO net{std::move(options)};

    net.startup();
    auto guard = MakeGuard([&] { net.shutdown(); });

    std::array<Deferred<StatusWith<RemoteCommandResponse>>, 10> deferreds;

    for (auto& deferred : deferreds) {
        net.startCommand(
            makeCallbackHandle(),
            RemoteCommandRequest{fixture.getServers()[0],
                                 "admin",
                                 BSON("sleep" << 1 << "lock"
                                              << "none"
                                              << "secs" << 2),
                                 BSONObj()},
            [&](StatusWith<RemoteCommandResponse> resp) { deferred.emplace(std::move(resp)); });
    }

    for (auto& deferred : deferreds) {
        ASSERT_EQ(deferred.get().isOK(), true);
    }

    sleepmillis(1000);

    ConnectionPoolStats cps;
    net.appendConnectionStats(&cps);

    ASSERT_LTE(cps.totalAvailable + cps.totalInUse, 1u);
    ASSERT_EQ(cps.totalCreated, 10u);
}

/**
 * Verify that when a refresh fails, it doesn't trigger an invariant
 */
TEST(ConnectionPoolASIO, ConnRefreshSurvivesFailure) {
    auto fixture = unittest::getFixtureConnectionString();

    NetworkInterfaceASIO::Options options;
    options.streamFactory = stdx::make_unique<AsyncStreamFactory>();
    options.timerFactory = stdx::make_unique<AsyncTimerFactoryASIO>();
    options.connectionPoolOptions.refreshRequirement = Milliseconds(0);
    NetworkInterfaceASIO net{std::move(options)};

    net.startup();
    auto guard = MakeGuard([&] { net.shutdown(); });

    Deferred<StatusWith<RemoteCommandResponse>> deferred;

    net.startCommand(
        makeCallbackHandle(),
        RemoteCommandRequest{fixture.getServers()[0], "admin", BSON("ping" << 1), BSONObj()},
        [&](StatusWith<RemoteCommandResponse> resp) { deferred.emplace(std::move(resp)); });

    deferred.get();

    getGlobalFailPointRegistry()
        ->getFailPoint("NetworkInterfaceASIOasyncRunCommandFail")
        ->setMode(FailPoint::nTimes, 1);

    sleepmillis(1000);

    ConnectionPoolStats cps;
    net.appendConnectionStats(&cps);

    ASSERT_EQ(cps.totalAvailable, 0u);
    ASSERT_EQ(cps.totalCreated, 1u);
}

/**
 * Tests that thrashing the timer while calls to setup are occurring doesn't crash.  This could
 * occur if a setup connection goes through, but the timer thread is too over-worked to cancel the
 * timers before they're invoked
 */
TEST(ConnectionPoolASIO, ConnSetupSurvivesFailure) {
    auto fixture = unittest::getFixtureConnectionString();

    NetworkInterfaceASIO::Options options;
    options.streamFactory = stdx::make_unique<AsyncStreamFactory>();
    options.timerFactory = stdx::make_unique<AsyncTimerFactoryASIO>();
    options.connectionPoolOptions.refreshTimeout = Seconds(1);
    NetworkInterfaceASIO net{std::move(options)};

    net.startup();
    auto guard = MakeGuard([&] { net.shutdown(); });

    const int kNumThreads = 8;
    const int kNumOps = 1000;

    AtomicWord<size_t> unfinished(kNumThreads * kNumOps);
    AtomicWord<size_t> unstarted(kNumThreads * kNumOps);

    std::array<stdx::thread, kNumThreads> threads;
    stdx::mutex mutex;
    stdx::condition_variable condvar;

    for (auto& thread : threads) {
        thread = stdx::thread([&] {
            for (int i = 0; i < kNumOps; i++) {
                RemoteCommandRequest request{fixture.getServers()[0],
                                             "admin",
                                             BSON("sleep" << 1 << "lock"
                                                          << "none"
                                                          << "secs" << 3),
                                             BSONObj()};
                net.startCommand(makeCallbackHandle(),
                                 request,
                                 [&](StatusWith<RemoteCommandResponse> resp) {
                                     if (!unfinished.subtractAndFetch(1)) {
                                         condvar.notify_one();
                                     }
                                 });
                unstarted.subtractAndFetch(1);
            }
        });
    }

    stdx::thread timerThrasher([&] {
        while (unstarted.load()) {
            net.setAlarm(Date_t::now() + Seconds(1), [] {});
        }
    });


    for (auto& thread : threads) {
        thread.join();
    }

    stdx::unique_lock<stdx::mutex> lk(mutex);
    condvar.wait(lk, [&] { return !unfinished.load(); });

    timerThrasher.join();
}

}  // namespace
}  // namespace executor
}  // namespace mongo


<|start_filename|>jstests/noPassthrough/yield_group.js<|end_filename|>
/**
 * Test that the "group" command yields periodically (SERVER-1395).
 */
(function() {
    'use strict';

    const nDocsToInsert = 300;
    const worksPerYield = 50;

    // Start a mongod that will yield every 50 work cycles.
    // clang-format off
    var conn = MongoRunner.runMongod(
        {setParameter: `internalQueryExecYieldIterations=${worksPerYield}`});
    // clang-format on
    assert.neq(null, conn, 'mongod was unable to start up');

    var coll = conn.getDB('test').yield_group;

    var bulk = coll.initializeUnorderedBulkOp();
    for (var i = 0; i < nDocsToInsert; ++i) {
        bulk.insert({_id: i});
    }

    var res = bulk.execute();
    assert.writeOK(res);
    assert.eq(nDocsToInsert, res.nInserted, tojson(res));

    // A "group" command performing a collection scan should yield approximately
    // nDocsToInsert / worksPerYield times.
    var explain =
        coll.explain('executionStats').group({key: {_id: 1}, reduce: function() {}, initial: {}});
    var numYields = explain.executionStats.executionStages.saveState;
    assert.gt(numYields, (nDocsToInsert / worksPerYield) - 2, tojson(explain));
})();


<|start_filename|>src/mongo/s/catalog/replset/catalog_manager_replica_set.h<|end_filename|>
/**
 *    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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#pragma once

#include "mongo/client/connection_string.h"
#include "mongo/db/repl/optime.h"
#include "mongo/s/catalog/catalog_manager_common.h"
#include "mongo/s/client/shard_registry.h"
#include "mongo/stdx/mutex.h"

namespace mongo {

class NamespaceString;
struct ReadPreferenceSetting;
class VersionType;

/**
 * Implements the catalog manager for talking to replica set config servers.
 */
class CatalogManagerReplicaSet final : public CatalogManagerCommon {
public:
    explicit CatalogManagerReplicaSet(std::unique_ptr<DistLockManager> distLockManager);
    virtual ~CatalogManagerReplicaSet();

    ConfigServerMode getMode() override {
        return ConfigServerMode::CSRS;
    }

    /**
     * Safe to call multiple times as long as they
     */
    Status startup(OperationContext* txn, bool allowNetworking) override;

    void shutDown(OperationContext* txn, bool allowNetworking) override;

    Status shardCollection(OperationContext* txn,
                           const std::string& ns,
                           const ShardKeyPattern& fieldsAndOrder,
                           bool unique,
                           const std::vector<BSONObj>& initPoints,
                           const std::set<ShardId>& initShardsIds) override;

    StatusWith<ShardDrainingStatus> removeShard(OperationContext* txn,
                                                const std::string& name) override;

    StatusWith<OpTimePair<DatabaseType>> getDatabase(OperationContext* txn,
                                                     const std::string& dbName) override;

    StatusWith<OpTimePair<CollectionType>> getCollection(OperationContext* txn,
                                                         const std::string& collNs) override;

    Status getCollections(OperationContext* txn,
                          const std::string* dbName,
                          std::vector<CollectionType>* collections,
                          repl::OpTime* optime) override;

    Status dropCollection(OperationContext* txn, const NamespaceString& ns) override;

    Status getDatabasesForShard(OperationContext* txn,
                                const std::string& shardName,
                                std::vector<std::string>* dbs) override;

    Status getChunks(OperationContext* txn,
                     const BSONObj& query,
                     const BSONObj& sort,
                     boost::optional<int> limit,
                     std::vector<ChunkType>* chunks,
                     repl::OpTime* opTime) override;

    Status getTagsForCollection(OperationContext* txn,
                                const std::string& collectionNs,
                                std::vector<TagsType>* tags) override;

    StatusWith<std::string> getTagForChunk(OperationContext* txn,
                                           const std::string& collectionNs,
                                           const ChunkType& chunk) override;

    StatusWith<OpTimePair<std::vector<ShardType>>> getAllShards(OperationContext* txn) override;

    bool runUserManagementWriteCommand(OperationContext* txn,
                                       const std::string& commandName,
                                       const std::string& dbname,
                                       const BSONObj& cmdObj,
                                       BSONObjBuilder* result) override;

    bool runUserManagementReadCommand(OperationContext* txn,
                                      const std::string& dbname,
                                      const BSONObj& cmdObj,
                                      BSONObjBuilder* result) override;

    Status applyChunkOpsDeprecated(OperationContext* txn,
                                   const BSONArray& updateOps,
                                   const BSONArray& preCondition,
                                   const std::string& nss,
                                   const ChunkVersion& lastChunkVersion) override;

    StatusWith<SettingsType> getGlobalSettings(OperationContext* txn,
                                               const std::string& key) override;

    void writeConfigServerDirect(OperationContext* txn,
                                 const BatchedCommandRequest& request,
                                 BatchedCommandResponse* response) override;

    Status insertConfigDocument(OperationContext* txn,
                                const std::string& ns,
                                const BSONObj& doc) override;

    StatusWith<bool> updateConfigDocument(OperationContext* txn,
                                          const std::string& ns,
                                          const BSONObj& query,
                                          const BSONObj& update,
                                          bool upsert) override;

    Status removeConfigDocuments(OperationContext* txn,
                                 const std::string& ns,
                                 const BSONObj& query) override;

    DistLockManager* getDistLockManager() override;

    Status initConfigVersion(OperationContext* txn) override;

    Status appendInfoForConfigServerDatabases(OperationContext* txn,
                                              BSONArrayBuilder* builder) override;

    /**
     * Runs a read command against the config server with majority read concern.
     */
    bool runReadCommandForTest(OperationContext* txn,
                               const std::string& dbname,
                               const BSONObj& cmdObj,
                               BSONObjBuilder* result);

    bool isMetadataConsistentFromLastCheck(OperationContext* txn) override;

private:
    Status _checkDbDoesNotExist(OperationContext* txn,
                                const std::string& dbName,
                                DatabaseType* db) override;

    StatusWith<std::string> _generateNewShardName(OperationContext* txn) override;

    Status _createCappedConfigCollection(OperationContext* txn,
                                         StringData collName,
                                         int cappedSize) override;

    /**
     * Executes the specified batch write command on the current config server's primary and retries
     * on the specified set of errors using the default retry policy.
     */
    void _runBatchWriteCommand(OperationContext* txn,
                               const BatchedCommandRequest& request,
                               BatchedCommandResponse* response,
                               const ShardRegistry::ErrorCodesSet& errorsToCheck);

    /**
     * Helper method for running a count command against the config server with appropriate
     * error handling.
     */
    StatusWith<long long> _runCountCommandOnConfig(OperationContext* txn,
                                                   const NamespaceString& ns,
                                                   BSONObj query);

    StatusWith<OpTimePair<std::vector<BSONObj>>> _exhaustiveFindOnConfig(
        OperationContext* txn,
        const ReadPreferenceSetting& readPref,
        const NamespaceString& nss,
        const BSONObj& query,
        const BSONObj& sort,
        boost::optional<long long> limit);

    /**
     * Appends a read committed read concern to the request object.
     */
    void _appendReadConcern(BSONObjBuilder* builder);

    /**
     * Returns the current cluster schema/protocol version.
     */
    StatusWith<VersionType> _getConfigVersion(OperationContext* txn);

    /**
     * Queries the config servers for the database metadata for the given database, using the
     * given read preference.  Returns NamespaceNotFound if no database metadata is found.
     */
    StatusWith<OpTimePair<DatabaseType>> _fetchDatabaseMetadata(
        OperationContext* txn, const std::string& dbName, const ReadPreferenceSetting& readPref);

    //
    // All member variables are labeled with one of the following codes indicating the
    // synchronization rules for accessing them.
    //
    // (M) Must hold _mutex for access.
    // (R) Read only, can only be written during initialization.
    //

    stdx::mutex _mutex;

    // Distribted lock manager singleton.
    std::unique_ptr<DistLockManager> _distLockManager;  // (R)

    // True if shutDown() has been called. False, otherwise.
    bool _inShutdown = false;  // (M)

    // Last known highest opTime from the config server.
    repl::OpTime _configOpTime;  // (M)
};

}  // namespace mongo


<|start_filename|>src/mongo/bson/util/bson_check.h<|end_filename|>
/**
 *    Copyright 2014 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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#pragma once

#include "mongo/base/status.h"
#include "mongo/db/jsobj.h"
#include "mongo/util/mongoutils/str.h"

namespace mongo {

/**
 * Confirms that "o" only contains fields whose names are in "begin".."end",
 * and that no field name occurs multiple times.
 *
 * On failure, returns BadValue and a message naming the unexpected field or DuplicateKey and a
 * message naming the repeated field.  "objectName" is included in the message, for reporting
 * purposes.
 */
template <typename Iter>
Status bsonCheckOnlyHasFields(StringData objectName,
                              const BSONObj& o,
                              const Iter& begin,
                              const Iter& end) {
    std::vector<int> occurrences(std::distance(begin, end), 0);
    for (BSONObj::iterator iter(o); iter.more();) {
        const BSONElement e = iter.next();
        const Iter found = std::find(begin, end, e.fieldNameStringData());
        if (found != end) {
            ++occurrences[std::distance(begin, found)];
        } else {
            return Status(ErrorCodes::BadValue,
                          str::stream() << "Unexpected field " << e.fieldName() << " in "
                                        << objectName);
        }
    }
    int i = 0;
    for (Iter curr = begin; curr != end; ++curr, ++i) {
        if (occurrences[i] > 1) {
            return Status(ErrorCodes::DuplicateKey,
                          str::stream() << "Field " << *curr << " appears " << occurrences[i]
                                        << " times in " << objectName);
        }
    }
    return Status::OK();
}

/**
 * Same as above, but operates over an array of string-ish items, "legals", instead
 * of "begin".."end".
 */
template <typename StringType, int N>
Status bsonCheckOnlyHasFields(StringData objectName,
                              const BSONObj& o,
                              const StringType(&legals)[N]) {
    return bsonCheckOnlyHasFields(objectName, o, &legals[0], legals + N);
}

/**
 * Throws a uassert if the type of the elem does not match that provided in expectedType
 */
inline void checkBSONType(BSONType expectedType, const BSONElement& elem) {
    uassert(elem.type() == BSONType::EOO ? ErrorCodes::NoSuchKey : ErrorCodes::TypeMismatch,
            str::stream() << "Wrong type for '" << elem.fieldNameStringData() << "'. Expected a "
                          << typeName(expectedType) << ", got a " << typeName(elem.type()) << '.',
            elem.type() == expectedType);
}


}  // namespace mongo


<|start_filename|>src/mongo/db/repl/minvalid.h<|end_filename|>
/**
 *    Copyright (C) 2014 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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#pragma once

#include "mongo/db/repl/optime.h"

namespace mongo {
class BSONObj;
class OperationContext;

namespace repl {

/**
 * Helper functions for maintaining a single document in the local.replset.minvalid collection.
 *
 * When a member reaches its minValid optime it is in a consistent state.  Thus, minValid is
 * set as the last step in initial sync.  At the beginning of initial sync, doingInitialSync
 * is appended onto minValid to indicate that initial sync was started but has not yet
 * completed.
 *
 * The document is also updated during "normal" sync. The optime of the last op in each batch is
 * used to set minValid, along with a "begin" field to demark the start and the fact that a batch
 * is active. When the batch is done the "begin" field is removed to indicate that we are in a
 * consistent state when the batch has been fully applied.
 *
 * Example of all fields:
 * { _id:...,
 *      doingInitialSync: true // initial sync is active
 *      ts:..., t:...   // end-OpTime
 *      begin: {ts:..., t:...} // a batch is currently being applied, and not consistent
 * }
 */

/**
 * The initial sync flag is used to durably record that initial sync has not completed.
 *
 * These operations wait for durable writes (which will block on journaling/checkpointing).
 */
void clearInitialSyncFlag(OperationContext* txn);
void setInitialSyncFlag(OperationContext* txn);

/**
 * Returns true if the initial sync flag is set (and presumed active).
 */
bool getInitialSyncFlag(OperationContext* txn);
bool getInitialSyncFlag();

/**
 * The minValid value is the earliest (minimum) Timestamp that must be applied in order to
 * consider the dataset consistent.
 */
void setMinValid(OperationContext* txn, const OpTime& minValid);
OpTime getMinValid(OperationContext* txn);

/**
 * Sets minValid only if it is not already higher than endOpTime.
 * Warning, this compares the term and timestamp independently. Do not use if the current
 * minValid could be from the other fork of a rollback.
 */
void setMinValidToAtLeast(OperationContext* txn, const OpTime& endOpTime);

/**
 * On startup all oplog entries with a value >= the oplog delete from point should be deleted.
 * If null, no documents should be deleted.
 */
void setOplogDeleteFromPoint(OperationContext* txn, const Timestamp& timestamp);
Timestamp getOplogDeleteFromPoint(OperationContext* txn);

/**
 * The applied through point is a persistent record of where we've applied through. If null, the
 * applied through point is the top of the oplog.
 */
void setAppliedThrough(OperationContext* txn, const OpTime& optime);

/**
 * You should probably be calling ReplicationCoordinator::getLastAppliedOpTime() instead.
 *
 * This reads the value from storage which isn't always updated when the ReplicationCoordinator
 * is.
 */
OpTime getAppliedThrough(OperationContext* txn);
}
}


<|start_filename|>jstests/hooks/check_repl_dbhash.js<|end_filename|>
// Check that the dbhashes of all the nodes in a ReplSetTest are consistent.
'use strict';

function checkDBHashes(rst, dbBlacklist = [], phase = 'after test hook') {
    function generateUniqueDbName(dbNameSet, prefix) {
        var uniqueDbName;
        Random.setRandomSeed();
        do {
            uniqueDbName = prefix + Random.randInt(100000);
        } while (dbNameSet.has(uniqueDbName));
        return uniqueDbName;
    }

    // Return items that are in either Array `a` or `b` but not both. Note that this will not work
    // with arrays containing NaN. Array.indexOf(NaN) will always return -1.
    function arraySymmetricDifference(a, b) {
        var inAOnly = a.filter(function(elem) {
            return b.indexOf(elem) < 0;
        });

        var inBOnly = b.filter(function(elem) {
            return a.indexOf(elem) < 0;
        });

        return inAOnly.concat(inBOnly);
    }

    function dumpCollectionDiff(primary, secondary, dbName, collName) {
        print('Dumping collection: ' + dbName + '.' + collName);

        var primaryColl = primary.getDB(dbName).getCollection(collName);
        var secondaryColl = secondary.getDB(dbName).getCollection(collName);

        var primaryDocs = primaryColl.find().sort({_id: 1}).toArray();
        var secondaryDocs = secondaryColl.find().sort({_id: 1}).toArray();

        var primaryIndex = primaryDocs.length - 1;
        var secondaryIndex = secondaryDocs.length - 1;

        var missingOnPrimary = [];
        var missingOnSecondary = [];

        while (primaryIndex >= 0 || secondaryIndex >= 0) {
            var primaryDoc = primaryDocs[primaryIndex];
            var secondaryDoc = secondaryDocs[secondaryIndex];

            if (primaryIndex < 0) {
                missingOnPrimary.push(tojsononeline(secondaryDoc));
                secondaryIndex--;
            } else if (secondaryIndex < 0) {
                missingOnSecondary.push(tojsononeline(primaryDoc));
                primaryIndex--;
            } else {
                if (!bsonBinaryEqual(primaryDoc, secondaryDoc)) {
                    print('Mismatching documents:');
                    print('    primary: ' + tojsononeline(primaryDoc));
                    print('    secondary: ' + tojsononeline(secondaryDoc));
                    var ordering =
                        bsonWoCompare({wrapper: primaryDoc._id}, {wrapper: secondaryDoc._id});
                    if (ordering === 0) {
                        primaryIndex--;
                        secondaryIndex--;
                    } else if (ordering < 0) {
                        missingOnPrimary.push(tojsononeline(secondaryDoc));
                        secondaryIndex--;
                    } else if (ordering > 0) {
                        missingOnSecondary.push(tojsononeline(primaryDoc));
                        primaryIndex--;
                    }
                } else {
                    // Latest document matched.
                    primaryIndex--;
                    secondaryIndex--;
                }
            }
        }

        if (missingOnPrimary.length) {
            print('The following documents are missing on the primary:');
            print(missingOnPrimary.join('\n'));
        }
        if (missingOnSecondary.length) {
            print('The following documents are missing on the secondary:');
            print(missingOnSecondary.join('\n'));
        }
    }

    function checkDBHashesForReplSet(rst, dbBlacklist, phase) {
        // We don't expect the local database to match because some of its collections are not
        // replicated.
        dbBlacklist.push('local');

        var success = true;
        var hasDumpedOplog = false;

        // Use liveNodes.master instead of getPrimary() to avoid the detection of a new primary.
        // liveNodes must have been populated.
        var primary = rst.liveNodes.master;
        var combinedDBs = new Set(primary.getDBNames());

        rst.getSecondaries().forEach(secondary => {
            secondary.getDBNames().forEach(dbName => combinedDBs.add(dbName));
        });

        for (var dbName of combinedDBs) {
            if (Array.contains(dbBlacklist, dbName)) {
                continue;
            }

            var dbHashes = rst.getHashes(dbName);
            var primaryDBHash = dbHashes.master;
            assert.commandWorked(primaryDBHash);

            var primaryCollInfo = primary.getDB(dbName).getCollectionInfos();

            dbHashes.slaves.forEach(secondaryDBHash => {
                assert.commandWorked(secondaryDBHash);

                var secondary =
                    rst.liveNodes.slaves.find(node => node.host === secondaryDBHash.host);
                assert(secondary,
                       'could not find the replica set secondary listed in the dbhash response ' +
                           tojson(secondaryDBHash));

                var primaryCollections = Object.keys(primaryDBHash.collections);
                var secondaryCollections = Object.keys(secondaryDBHash.collections);

                if (primaryCollections.length !== secondaryCollections.length) {
                    print(phase +
                          ', the primary and secondary have a different number of collections: ' +
                          tojson(dbHashes));
                    for (var diffColl of
                             arraySymmetricDifference(primaryCollections, secondaryCollections)) {
                        dumpCollectionDiff(primary, secondary, dbName, diffColl);
                    }
                    success = false;
                }

                var nonCappedCollNames = primaryCollections.filter(
                    collName => !primary.getDB(dbName).getCollection(collName).isCapped());
                // Only compare the dbhashes of non-capped collections because capped collections
                // are not necessarily truncated at the same points across replica set members.
                nonCappedCollNames.forEach(collName => {
                    if (primaryDBHash.collections[collName] !==
                        secondaryDBHash.collections[collName]) {
                        print(phase + ', the primary and secondary have a different hash for the' +
                              ' collection ' + dbName + '.' + collName + ': ' + tojson(dbHashes));
                        dumpCollectionDiff(primary, secondary, dbName, collName);
                        success = false;
                    }

                });

                // Check that collection information is consistent on the primary and secondaries.
                var secondaryCollInfo = secondary.getDB(dbName).getCollectionInfos();
                secondaryCollInfo.forEach(secondaryInfo => {
                    primaryCollInfo.forEach(primaryInfo => {
                        if (secondaryInfo.name === primaryInfo.name) {
                            if (!bsonBinaryEqual(secondaryInfo, primaryInfo)) {
                                print(phase +
                                      ', the primary and secondary have different attributes for ' +
                                      'the collection ' + dbName + '.' + secondaryInfo.name);
                                print('Collection info on the primary: ' + tojson(primaryInfo));
                                print('Collection info on the secondary: ' + tojson(secondaryInfo));
                                success = false;
                            }
                        }
                    });
                });

                // Check that the following collection stats are the same across replica set
                // members:
                //  capped
                //  nindexes
                //  ns
                primaryCollections.forEach(collName => {
                    var primaryCollStats = primary.getDB(dbName).runCommand({collStats: collName});
                    assert.commandWorked(primaryCollStats);
                    var secondaryCollStats =
                        secondary.getDB(dbName).runCommand({collStats: collName});
                    assert.commandWorked(secondaryCollStats);

                    if (primaryCollStats.capped !== secondaryCollStats.capped ||
                        primaryCollStats.nindexes !== secondaryCollStats.nindexes ||
                        primaryCollStats.ns !== secondaryCollStats.ns) {
                        print(phase + ', the primary and secondary have different stats for the ' +
                              'collection ' + dbName + '.' + collName);
                        print('Collection stats on the primary: ' + tojson(primaryCollStats));
                        print('Collection stats on the secondary: ' + tojson(secondaryCollStats));
                        success = false;
                    }
                });

                if (nonCappedCollNames.length === primaryCollections.length) {
                    // If the primary and secondary have the same hashes for all the collections in
                    // the database and there aren't any capped collections, then the hashes for
                    // the whole database should match.
                    if (primaryDBHash.md5 !== secondaryDBHash.md5) {
                        print(phase + ', the primary and secondary have a different hash for ' +
                              'the ' + dbName + ' database: ' + tojson(dbHashes));
                        success = false;
                    }
                }

                if (!success) {
                    var dumpOplog = function(conn, limit) {
                        print('Dumping the latest ' + limit + ' documents from the oplog of ' +
                              conn.host);
                        var cursor = conn.getDB('local')
                                         .getCollection('oplog.rs')
                                         .find()
                                         .sort({$natural: -1})
                                         .limit(limit);
                        cursor.forEach(printjsononeline);
                    };

                    if (!hasDumpedOplog) {
                        dumpOplog(primary, 100);
                        rst.getSecondaries().forEach(secondary => dumpOplog(secondary, 100));
                        hasDumpedOplog = true;
                    }
                }
            });
        }

        assert(success, 'dbhash mismatch between primary and secondary');
    }

    // Call getPrimary to populate rst with information about the nodes.
    var primary = rst.getPrimary();
    assert(primary, 'calling getPrimary() failed');

    // Since we cannot determine if there is a background index in progress (SERVER-25176),
    // we flush indexing as follows:
    //  1. Create a foreground index on a dummy collection/database
    //  2. Insert a document into the dummy collection with a writeConcern for all nodes
    //  3. Drop the dummy database
    var dbNames = new Set(primary.getDBNames());
    var uniqueDbName = generateUniqueDbName(dbNames, "flush_all_background_indexes_");

    var dummyDB = primary.getDB(uniqueDbName);
    var dummyColl = dummyDB.dummy;
    dummyColl.drop();
    assert.commandWorked(dummyColl.createIndex({x: 1}));
    assert.writeOK(dummyColl.insert(
        {x: 1}, {writeConcern: {w: rst.nodeList().length, wtimeout: 5 * 60 * 1000}}));
    assert.commandWorked(dummyDB.dropDatabase());

    var activeException = false;

    try {
        // dbHash values for collections in the "config" database are cached and may be stale due to
        // SERVER-22156. We insert a document into each of these collections to invalidate
        // the potentially stale cache entries.
        var primaryConfigDb = rst.liveNodes.master.getDB('config');
        for (var collName of primaryConfigDb.getCollectionNames()) {
            // Invalidate the dbhash cache for non-system collections and user-writable system
            // collections.
            if (!collName.startsWith('system.') || collName === 'system.js' ||
                collName === 'system.users')
                // This insert can fail if there's a unique index on the collection, so we can't
                // check the return value.
                primaryConfigDb.getCollection(collName).insert({invalidate: 'cache'});
        }

        // Lock the primary to prevent the TTL monitor from deleting expired documents in
        // the background while we are getting the dbhashes of the replica set members.
        assert.commandWorked(primary.adminCommand({fsync: 1, lock: 1}),
                             'failed to lock the primary');
        rst.awaitReplication(60 * 1000 * 5);
        checkDBHashesForReplSet(rst, dbBlacklist, phase);
    } catch (e) {
        activeException = true;
        throw e;
    } finally {
        // Allow writes on the primary.
        var res = primary.adminCommand({fsyncUnlock: 1});

        if (!res.ok) {
            var msg = 'failed to unlock the primary, which may cause this' +
                ' test to hang: ' + tojson(res);
            if (activeException) {
                print(msg);
            } else {
                throw new Error(msg);
            }
        }
    }
}


<|start_filename|>jstests/sharding/conf_server_write_concern.js<|end_filename|>
/**
 * Test write concern with w parameter when writing directly to the config servers works as expected
 */
function writeToConfigTest() {
    jsTestLog("Testing data writes to config server with write concern");
    var st = new ShardingTest({shards: 2});
    var confDB = st.s.getDB('config');

    assert.writeOK(confDB.settings.update(
        {_id: 'balancer'}, {$set: {stopped: true}}, {writeConcern: {w: 'majority'}}));

    // w:1 should still work - it gets automatically upconverted to w:majority
    assert.writeOK(confDB.settings.update(
        {_id: 'balancer'}, {$set: {stopped: true}}, {writeConcern: {w: 1}}));

    // Write concerns other than w:1 and w:majority should fail.
    assert.writeError(confDB.settings.update(
        {_id: 'balancer'}, {$set: {stopped: true}}, {writeConcern: {w: 2}}));

    st.stop();
}

/**
 * Test write concern with w parameter will not cause an error when writes to mongos
 * would trigger writes to config servers (in this test, autosplit is used).
 */
function configTest() {
    jsTestLog("Testing metadata writes to config server with write concern");
    var st = new ShardingTest({shards: 1, rs: true, other: {chunkSize: 1, enableAutoSplit: true}});

    var mongos = st.s;
    var testDB = mongos.getDB('test');
    var coll = testDB.user;

    testDB.adminCommand({enableSharding: testDB.getName()});
    testDB.adminCommand({shardCollection: coll.getFullName(), key: {x: 1}});

    var chunkCount = function() {
        return mongos.getDB('config').chunks.find().count();
    };

    var initChunks = chunkCount();
    var currChunks = initChunks;
    var gleObj = null;
    var x = 0;
    var largeStr = new Array(1024 * 128).toString();

    assert.soon(
        function() {
            var bulk = coll.initializeUnorderedBulkOp();
            for (var i = 0; i < 100; i++) {
                bulk.insert({x: x++, largeStr: largeStr});
            }
            assert.writeOK(bulk.execute({w: 'majority', wtimeout: 60 * 1000}));
            currChunks = chunkCount();
            return currChunks > initChunks;
        },
        function() {
            return "currChunks: " + currChunks + ", initChunks: " + initChunks;
        });

    st.stop();
}

writeToConfigTest();
configTest();


<|start_filename|>src/mongo/client/fetcher.cpp<|end_filename|>
/**
 *    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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */
#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kExecutor

#include "mongo/platform/basic.h"

#include "mongo/client/fetcher.h"

#include "mongo/base/data_range_cursor.h"
#include "mongo/base/data_type_validated.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/namespace_string.h"
#include "mongo/rpc/get_status_from_command_result.h"
#include "mongo/rpc/object_check.h"
#include "mongo/util/assert_util.h"
#include "mongo/util/log.h"
#include "mongo/util/mongoutils/str.h"

namespace mongo {

namespace {

using executor::RemoteCommandRequest;
using executor::RemoteCommandResponse;

using RemoteCommandCallbackArgs = executor::TaskExecutor::RemoteCommandCallbackArgs;
const char* kCursorFieldName = "cursor";
const char* kCursorIdFieldName = "id";
const char* kNamespaceFieldName = "ns";

const char* kFirstBatchFieldName = "firstBatch";
const char* kNextBatchFieldName = "nextBatch";

Status parseCursorResponseFromResponseObj(const BSONObj& responseObj,
                                          const std::string& batchFieldName,
                                          Fetcher::QueryResponse* batchData) {
    BSONElement cursorElement = responseObj.getField(kCursorFieldName);
    if (cursorElement.eoo()) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "cursor response must contain '" << kCursorFieldName
                                    << "' field: " << responseObj);
    }
    if (!cursorElement.isABSONObj()) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "'" << kCursorFieldName
                                    << "' field must be an object: " << responseObj);
    }
    BSONObj cursorObj = cursorElement.Obj();

    BSONElement cursorIdElement = cursorObj.getField(kCursorIdFieldName);
    if (cursorIdElement.eoo()) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "cursor response must contain '" << kCursorFieldName << "."
                                    << kCursorIdFieldName << "' field: " << responseObj);
    }
    if (cursorIdElement.type() != mongo::NumberLong) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "'" << kCursorFieldName << "." << kCursorIdFieldName
                                    << "' field must be a 'long' but was a '"
                                    << typeName(cursorIdElement.type()) << "': " << responseObj);
    }
    batchData->cursorId = cursorIdElement.numberLong();

    BSONElement namespaceElement = cursorObj.getField(kNamespaceFieldName);
    if (namespaceElement.eoo()) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "cursor response must contain "
                                    << "'" << kCursorFieldName << "." << kNamespaceFieldName
                                    << "' field: " << responseObj);
    }
    if (namespaceElement.type() != mongo::String) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "'" << kCursorFieldName << "." << kNamespaceFieldName
                                    << "' field must be a string: " << responseObj);
    }
    NamespaceString tempNss(namespaceElement.valuestrsafe());
    if (!tempNss.isValid()) {
        return Status(ErrorCodes::BadValue,
                      str::stream() << "'" << kCursorFieldName << "." << kNamespaceFieldName
                                    << "' contains an invalid namespace: " << responseObj);
    }
    batchData->nss = tempNss;

    BSONElement batchElement = cursorObj.getField(batchFieldName);
    if (batchElement.eoo()) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "cursor response must contain '" << kCursorFieldName << "."
                                    << batchFieldName << "' field: " << responseObj);
    }
    if (!batchElement.isABSONObj()) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << "'" << kCursorFieldName << "." << batchFieldName
                                    << "' field must be an array: " << responseObj);
    }
    BSONObj batchObj = batchElement.Obj();
    for (auto itemElement : batchObj) {
        if (!itemElement.isABSONObj()) {
            return Status(ErrorCodes::FailedToParse,
                          str::stream() << "found non-object " << itemElement << " in "
                                        << "'" << kCursorFieldName << "." << batchFieldName
                                        << "' field: " << responseObj);
        }
        batchData->documents.push_back(itemElement.Obj().getOwned());
    }

    return Status::OK();
}

/**
 * Extracts the CursorId and array of results from a Message representing an OP_REPLY. Returns a
 * non-OK status if Message does not represent a well-formed OP_REPLY.
 */
StatusWith<std::tuple<CursorId, std::vector<BSONObj>>> getBatchFromReply(const Message* response) {
    auto header = response->header();
    if (header.getNetworkOp() != mongo::opReply) {
        return {ErrorCodes::ProtocolError,
                str::stream() << "Expected to be decoding an OP_REPLY but got "
                              << mongo::networkOpToString(header.getNetworkOp())};
    }

    if ((header.dataLen() < 0) ||
        (static_cast<std::size_t>(header.dataLen()) > mongo::MaxMessageSizeBytes)) {
        return {ErrorCodes::InvalidLength,
                str::stream() << "Received message has invalid length field with value "
                              << header.dataLen()};
    }

    QueryResult::View qr = response->header().view2ptr();

    auto resultFlags = qr.getResultFlags();

    if (resultFlags & ResultFlag_CursorNotFound) {
        return {ErrorCodes::CursorNotFound,
                str::stream() << "Cursor with id '" << qr.getCursorId() << "' not found"};
    }

    // Use CDRC directly instead of DocumentRange as DocumentRange has a throwing API.
    ConstDataRangeCursor cdrc{qr.data(), qr.data() + header.dataLen()};

    if (resultFlags & ResultFlag_ErrSet) {
        if (qr.getNReturned() != 1) {
            return {ErrorCodes::BadValue,
                    str::stream() << "ResultFlag_ErrSet flag set on reply, but nReturned was '"
                                  << qr.getNReturned() << "' - expected 1"};
        }
        // Convert error document to a Status.
        // Will throw if first document is invalid BSON.
        auto first = cdrc.readAndAdvance<Validated<BSONObj>>();
        if (!first.isOK()) {
            return first.getStatus();
        }

        // Convert error document to a status.
        return getStatusFromCommandResult(first.getValue());
    }

    const int32_t nReturned = qr.getNReturned();
    std::vector<BSONObj> batch;
    batch.reserve(qr.getNReturned());

    int32_t nParsed = 0;
    Validated<BSONObj> nextObj;
    while (!cdrc.empty() && nParsed < nReturned) {
        auto readStatus = cdrc.readAndAdvance(&nextObj);
        if (!readStatus.isOK()) {
            return readStatus;
        }
        ++nParsed;
        batch.emplace_back(nextObj.val.getOwned());
    }
    if (nParsed != nReturned) {
        return {ErrorCodes::InvalidLength,
                str::stream() << "Count of documents in OP_REPLY message (" << nParsed
                              << ") did not match the value specified in the nReturned field ("
                              << nReturned << ")"};
    }

    return {std::make_tuple(qr.getCursorId(), std::move(batch))};
}

Status parseCursorResponseFromRawMessage(const Message* message,
                                         Fetcher::QueryResponse* batchData) {
    auto batchStatus = getBatchFromReply(message);
    if (!batchStatus.isOK()) {
        return batchStatus.getStatus();
    }

    std::tie(batchData->cursorId, batchData->documents) = batchStatus.getValue();
    return Status::OK();
}

/**
 * Parses cursor response in command result for cursor ID, namespace and documents.
 * 'batchFieldName' will be 'firstBatch' for the initial remote command invocation and 'nextBatch'
 * for getMore.
 */
Status parseCursorResponse(const RemoteCommandResponse& response,
                           const std::string& batchFieldName,
                           Fetcher::QueryResponse* batchData) {
    invariant(batchFieldName == kFirstBatchFieldName || batchFieldName == kNextBatchFieldName);
    invariant(batchData);

    // If we are talking to a 3.0 mongod, then the response will have come back as an OP_QUERY, and
    // we'll need to parse the raw message to populate 'batchData'. Otherwise, we ran a find or
    // getMore command, and need to parse the BSON that is returned from those commands.
    Status status = getStatusFromCommandResult(response.data);
    if (status.isOK()) {
        return parseCursorResponseFromResponseObj(response.data, batchFieldName, batchData);
    } else if (status.code() == ErrorCodes::ReceivedOpReplyMessage) {
        auto ns = response.data["ns"];
        if (!ns) {
            return {ErrorCodes::FailedToParse,
                    str::stream() << "expected 'ns' field to be present in response: "
                                  << response.data};
        }
        if (ns.type() != String) {
            return {ErrorCodes::FailedToParse,
                    str::stream() << "expected 'ns' field to be a string, was "
                                  << typeName(ns.type()) << ": " << response.data};
        }
        auto nss = NamespaceString(ns.String());
        if (!nss.isValid()) {
            return {ErrorCodes::FailedToParse,
                    str::stream() << "invalid 'ns' field in response: " << response.data};
        }
        batchData->nss = nss;
        return parseCursorResponseFromRawMessage(response.message.get(), batchData);
    } else {
        return status;
    }
}

}  // namespace

Fetcher::Fetcher(executor::TaskExecutor* executor,
                 const HostAndPort& source,
                 const std::string& dbname,
                 const BSONObj& findCmdObj,
                 const CallbackFn& work,
                 const BSONObj& metadata)
    : Fetcher(
          executor, source, dbname, findCmdObj, work, metadata, RemoteCommandRequest::kNoTimeout) {}

Fetcher::Fetcher(executor::TaskExecutor* executor,
                 const HostAndPort& source,
                 const std::string& dbname,
                 const BSONObj& findCmdObj,
                 const CallbackFn& work,
                 const BSONObj& metadata,
                 Milliseconds timeout)
    : _executor(executor),
      _source(source),
      _dbname(dbname),
      _cmdObj(findCmdObj.getOwned()),
      _metadata(metadata.getOwned()),
      _work(work),
      _active(false),
      _first(true),
      _remoteCommandCallbackHandle(),
      _timeout(timeout) {
    uassert(ErrorCodes::BadValue, "null replication executor", executor);
    uassert(ErrorCodes::BadValue, "database name cannot be empty", !dbname.empty());
    uassert(ErrorCodes::BadValue, "command object cannot be empty", !findCmdObj.isEmpty());
    uassert(ErrorCodes::BadValue, "callback function cannot be null", work);
}

Fetcher::~Fetcher() {
    DESTRUCTOR_GUARD(cancel(); wait(););
}

std::string Fetcher::getDiagnosticString() const {
    stdx::lock_guard<stdx::mutex> lk(_mutex);
    str::stream output;
    output << "Fetcher";
    output << " executor: " << _executor->getDiagnosticString();
    output << " source: " << _source.toString();
    output << " database: " << _dbname;
    output << " query: " << _cmdObj;
    output << " query metadata: " << _metadata;
    output << " active: " << _active;
    output << " timeout: " << _timeout;
    return output;
}

bool Fetcher::isActive() const {
    stdx::lock_guard<stdx::mutex> lk(_mutex);
    return _active;
}

Status Fetcher::schedule() {
    stdx::lock_guard<stdx::mutex> lk(_mutex);
    if (_active) {
        return Status(ErrorCodes::IllegalOperation, "fetcher already scheduled");
    }
    return _schedule_inlock(_cmdObj, kFirstBatchFieldName);
}

void Fetcher::cancel() {
    executor::TaskExecutor::CallbackHandle remoteCommandCallbackHandle;
    {
        stdx::lock_guard<stdx::mutex> lk(_mutex);

        if (!_active) {
            return;
        }

        remoteCommandCallbackHandle = _remoteCommandCallbackHandle;
    }

    invariant(remoteCommandCallbackHandle.isValid());
    _executor->cancel(remoteCommandCallbackHandle);
}

void Fetcher::wait() {
    stdx::unique_lock<stdx::mutex> lk(_mutex);
    _condition.wait(lk, [this]() { return !_active; });
}

Status Fetcher::_schedule_inlock(const BSONObj& cmdObj, const char* batchFieldName) {
    StatusWith<executor::TaskExecutor::CallbackHandle> scheduleResult =
        _executor->scheduleRemoteCommand(
            RemoteCommandRequest(_source, _dbname, cmdObj, _metadata, _timeout),
            stdx::bind(&Fetcher::_callback, this, stdx::placeholders::_1, batchFieldName));

    if (!scheduleResult.isOK()) {
        return scheduleResult.getStatus();
    }

    _active = true;
    _remoteCommandCallbackHandle = scheduleResult.getValue();
    return Status::OK();
}

void Fetcher::_callback(const RemoteCommandCallbackArgs& rcbd, const char* batchFieldName) {
    if (!rcbd.response.isOK()) {
        _work(StatusWith<Fetcher::QueryResponse>(rcbd.response.getStatus()), nullptr, nullptr);
        _finishCallback();
        return;
    }

    QueryResponse batchData;
    auto status = parseCursorResponse(rcbd.response.getValue(), batchFieldName, &batchData);
    if (!status.isOK()) {
        _work(StatusWith<Fetcher::QueryResponse>(status), nullptr, nullptr);
        _finishCallback();
        return;
    }

    batchData.otherFields.metadata = std::move(rcbd.response.getValue().metadata);
    batchData.elapsedMillis = rcbd.response.getValue().elapsedMillis;
    {
        stdx::lock_guard<stdx::mutex> lk(_mutex);
        batchData.first = _first;
        _first = false;
    }

    NextAction nextAction = NextAction::kNoAction;

    if (!batchData.cursorId) {
        _work(StatusWith<QueryResponse>(batchData), &nextAction, nullptr);
        _finishCallback();
        return;
    }

    nextAction = NextAction::kGetMore;

    BSONObjBuilder bob;
    _work(StatusWith<QueryResponse>(batchData), &nextAction, &bob);

    // Callback function _work may modify nextAction to request the fetcher
    // not to schedule a getMore command.
    if (nextAction != NextAction::kGetMore) {
        _sendKillCursors(batchData.cursorId, batchData.nss);
        _finishCallback();
        return;
    }

    // Callback function may also disable the fetching of additional data by not filling in the
    // BSONObjBuilder for the getMore command.
    auto cmdObj = bob.obj();
    if (cmdObj.isEmpty()) {
        _sendKillCursors(batchData.cursorId, batchData.nss);
        _finishCallback();
        return;
    }

    {
        stdx::lock_guard<stdx::mutex> lk(_mutex);
        status = _schedule_inlock(cmdObj, kNextBatchFieldName);
    }
    if (!status.isOK()) {
        nextAction = NextAction::kNoAction;
        _work(StatusWith<Fetcher::QueryResponse>(status), nullptr, nullptr);
        _sendKillCursors(batchData.cursorId, batchData.nss);
        _finishCallback();
        return;
    }
}

void Fetcher::_sendKillCursors(const CursorId id, const NamespaceString& nss) {
    if (id) {
        auto logKillCursorsResult = [](const RemoteCommandCallbackArgs& args) {
            if (!args.response.isOK()) {
                warning() << "killCursors command task failed: " << args.response.getStatus();
                return;
            }
            auto status = getStatusFromCommandResult(args.response.getValue().data);
            if (!status.isOK()) {
                warning() << "killCursors command failed: " << status;
            }
        };
        auto cmdObj = BSON("killCursors" << nss.coll() << "cursors" << BSON_ARRAY(id));
        auto scheduleResult = _executor->scheduleRemoteCommand(
            RemoteCommandRequest(_source, _dbname, cmdObj), logKillCursorsResult);
        if (!scheduleResult.isOK()) {
            warning() << "failed to schedule killCursors command: " << scheduleResult.getStatus();
        }
    }
}
void Fetcher::_finishCallback() {
    stdx::lock_guard<stdx::mutex> lk(_mutex);
    _active = false;
    _first = false;
    _condition.notify_all();
}

}  // namespace mongo


<|start_filename|>src/mongo/db/concurrency/locker_noop.h<|end_filename|>
/**
 *    Copyright (C) 2014 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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#pragma once

#include "mongo/db/concurrency/locker.h"

namespace mongo {

/**
 * Locker, which cannot be used to lock/unlock resources and just returns true for checks for
 * whether a particular resource is locked. Do not use it for cases where actual locking
 * behaviour is expected or locking is performed.
 */
class LockerNoop : public Locker {
public:
    LockerNoop() {}

    virtual bool isNoop() const {
        return true;
    }

    virtual ClientState getClientState() const {
        invariant(false);
    }

    virtual LockerId getId() const {
        invariant(false);
    }

    virtual LockResult lockGlobal(LockMode mode, unsigned timeoutMs) {
        invariant(false);
    }

    virtual LockResult lockGlobalBegin(LockMode mode) {
        invariant(false);
    }

    virtual LockResult lockGlobalComplete(unsigned timeoutMs) {
        invariant(false);
    }

    virtual void lockMMAPV1Flush() {
        invariant(false);
    }

    virtual bool unlockAll() {
        invariant(false);
    }

    virtual void downgradeGlobalXtoSForMMAPV1() {
        invariant(false);
    }

    virtual void beginWriteUnitOfWork() {}

    virtual void endWriteUnitOfWork() {}

    virtual bool inAWriteUnitOfWork() const {
        invariant(false);
    }

    virtual LockResult lock(ResourceId resId,
                            LockMode mode,
                            unsigned timeoutMs,
                            bool checkDeadlock) {
        invariant(false);
    }

    virtual void downgrade(ResourceId resId, LockMode newMode) {
        invariant(false);
    }

    virtual bool unlock(ResourceId resId) {
        invariant(false);
    }

    virtual LockMode getLockMode(ResourceId resId) const {
        invariant(false);
    }

    virtual bool isLockHeldForMode(ResourceId resId, LockMode mode) const {
        return true;
    }

    virtual bool isDbLockedForMode(StringData dbName, LockMode mode) const {
        return true;
    }

    virtual bool isCollectionLockedForMode(StringData ns, LockMode mode) const {
        return true;
    }

    virtual ResourceId getWaitingResource() const {
        invariant(false);
    }

    virtual void getLockerInfo(LockerInfo* lockerInfo) const {
        invariant(false);
    }

    virtual bool saveLockStateAndUnlock(LockSnapshot* stateOut) {
        invariant(false);
    }

    virtual void restoreLockState(const LockSnapshot& stateToRestore) {
        invariant(false);
    }

    virtual void dump() const {
        invariant(false);
    }

    virtual bool isW() const {
        invariant(false);
    }

    virtual bool isR() const {
        invariant(false);
    }

    virtual bool isLocked() const {
        return false;
    }

    virtual bool isWriteLocked() const {
        return false;
    }

    virtual bool isReadLocked() const {
        invariant(false);
    }

    virtual void assertEmptyAndReset() {
        invariant(false);
    }

    virtual bool hasLockPending() const {
        invariant(false);
    }

    virtual void setIsBatchWriter(bool newValue) {
        invariant(false);
    }

    virtual bool isBatchWriter() const {
        invariant(false);
    }

    virtual bool hasStrongLocks() const {
        return false;
    }
};

}  // namespace mongo


<|start_filename|>src/mongo/executor/downconvert_find_and_getmore_commands.cpp<|end_filename|>
/**
 *    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 <http://www.gnu.org/licenses/>.
 *
 *    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 "mongo/executor/downconvert_find_and_getmore_commands.h"

#include <memory>
#include <string>
#include <tuple>

#include "mongo/base/status_with.h"
#include "mongo/client/constants.h"
#include "mongo/client/dbclientinterface.h"
#include "mongo/db/cursor_id.h"
#include "mongo/db/namespace_string.h"
#include "mongo/db/query/cursor_response.h"
#include "mongo/db/query/getmore_request.h"
#include "mongo/executor/remote_command_request.h"
#include "mongo/executor/remote_command_response.h"
#include "mongo/rpc/get_status_from_command_result.h"
#include "mongo/rpc/metadata/server_selection_metadata.h"
#include "mongo/util/assert_util.h"
#include "mongo/util/net/message.h"

namespace mongo {
namespace executor {

namespace {

/**
 * Returns a non-OK status if 'response' does not report it is a response to the request with id
 * 'requestId'.
 */
Status checkMessageResponseTo(std::uint32_t requestId, const Message* response) {
    return response->header().getResponseTo() == requestId
        ? Status::OK()
        : Status{ErrorCodes::ProtocolError,
                 str::stream() << "responseTo field of OP_REPLY header with value "
                               << response->header().getResponseTo() << " does not match requestId "
                               << requestId};
}

}  // namespace

StatusWith<Message> downconvertFindCommandRequest(const RemoteCommandRequest& request) {
    const auto& cmdObj = request.cmdObj;
    const NamespaceString nss(request.dbname, cmdObj.firstElement().String());
    if (!nss.isValid()) {
        return {ErrorCodes::InvalidNamespace,
                str::stream() << "Invalid collection name: " << nss.ns()};
    }

    const std::string& ns = nss.ns();

    // It is a little heavy handed to use LiteParsedQuery to convert the command object to
    // query() arguments but we get validation and consistent behavior with the find
    // command implementation on the remote server.
    auto lpqStatus = LiteParsedQuery::makeFromFindCommand(nss, cmdObj, false);
    if (!lpqStatus.isOK()) {
        return lpqStatus.getStatus();
    }

    auto lpq = std::move(lpqStatus.getValue());

    // We are downconverting a find command, and find command can only have ntoreturn
    // if it was generated by mongos.
    invariant(!lpq->getNToReturn());
    Query query(lpq->getFilter());
    if (!lpq->getSort().isEmpty()) {
        query.sort(lpq->getSort());
    }
    if (!lpq->getHint().isEmpty()) {
        query.hint(lpq->getHint());
    }
    if (!lpq->getMin().isEmpty()) {
        query.minKey(lpq->getMin());
    }
    if (!lpq->getMax().isEmpty()) {
        query.minKey(lpq->getMax());
    }
    if (lpq->isExplain()) {
        query.explain();
    }
    if (lpq->isSnapshot()) {
        query.snapshot();
    }

    const int nToReturn = lpq->getLimit().value_or(0) * -1;
    const int nToSkip = lpq->getSkip().value_or(0);
    const BSONObj* fieldsToReturn = &lpq->getProj();
    int queryOptions = lpq->getOptions();  // non-const so we can set slaveOk if we need to
    const int batchSize = lpq->getBatchSize().value_or(0);

    const int nextBatchSize = [batchSize, nToReturn]() {
        if (nToReturn == 0)
            return batchSize;
        if (batchSize == 0)
            return nToReturn;
        return batchSize < nToReturn ? batchSize : nToReturn;
    }();

    // We can't downconvert all metadata, since we aren't sending a command, but we do need to
    // downconvert $secondaryOk to the slaveOK bit.
    auto ssm = rpc::ServerSelectionMetadata::readFromMetadata(
        request.metadata.getField(rpc::ServerSelectionMetadata::fieldName()));
    if (!ssm.isOK()) {
        return ssm.getStatus();
    }
    if (ssm.getValue().isSecondaryOk()) {
        queryOptions |= mongo::QueryOption_SlaveOk;
    }

    Message message;
    assembleQueryRequest(
        ns, query.obj, nextBatchSize, nToSkip, fieldsToReturn, queryOptions, message);

    return {std::move(message)};
}

StatusWith<RemoteCommandResponse> prepareOpReplyErrorResponse(std::uint32_t requestId,
                                                              StringData cursorNamespace,
                                                              Message* response) {
    auto status = checkMessageResponseTo(requestId, response);
    if (!status.isOK()) {
        return status;
    }

    BSONObjBuilder result;
    result.append("ok", 0.0);
    result.append("code", ErrorCodes::ReceivedOpReplyMessage);
    result.append("errmsg", "Received an OP_REPLY, unable to parse into a single BSONObj");
    result.append("ns", cursorNamespace);

    return {RemoteCommandResponse{std::move(*response), result.obj(), {}, {}}};
}

StatusWith<Message> downconvertGetMoreCommandRequest(const RemoteCommandRequest& request) {
    auto swGetMoreRequest = GetMoreRequest::parseFromBSON(request.dbname, request.cmdObj);
    if (!swGetMoreRequest.isOK()) {
        return swGetMoreRequest.getStatus();
    }

    auto getMoreRequest = std::move(swGetMoreRequest.getValue());

    BufBuilder b;
    b.appendNum(std::int32_t{0});  // reserved bits
    b.appendStr(getMoreRequest.nss.ns());
    // Without this static cast, we will append batchSize as an int64 and get an invalid message.
    b.appendNum(static_cast<std::int32_t>(getMoreRequest.batchSize.value_or(0)));
    b.appendNum(getMoreRequest.cursorid);
    Message m;
    m.setData(dbGetMore, b.buf(), b.len());

    return {std::move(m)};
}

}  // namespace mongo
}  // namespace executor


<|start_filename|>jstests/concurrency/fsm_workloads/kill_rooted_or.js<|end_filename|>
'use strict';

/**
 * kill_rooted_or.js
 *
 * Queries using a rooted $or predicate to cause plan selection to use the subplanner. Tests that
 * the subplanner correctly halts plan execution when the collection is dropped or a candidate index
 * is dropped.
 *
 * This workload was designed to reproduce SERVER-24761.
 */
var $config = (function() {

    // Use the workload name as the collection name, since the workload name is assumed to be
    // unique.
    var uniqueCollectionName = 'kill_rooted_or';

    var data = {
        collName: uniqueCollectionName,
        indexSpecs: [{a: 1}, {a: 1, c: 1}, {b: 1}, {b: 1, c: 1}]
    };

    var states = {
        query: function query(db, collName) {
            var cursor = db[this.collName].find({$or: [{a: 0}, {b: 0}]});
            try {
                // No documents are ever inserted into the collection.
                assertAlways.eq(0, cursor.itcount());
            } catch (e) {
                // Ignore errors due to the plan executor being killed.
            }
        },

        dropCollection: function dropCollection(db, collName) {
            db[this.collName].drop();

            // Recreate all of the indexes on the collection.
            this.indexSpecs.forEach(indexSpec => {
                assertAlways.commandWorked(db[this.collName].createIndex(indexSpec));
            });
        },

        dropIndex: function dropIndex(db, collName) {
            var indexSpec = this.indexSpecs[Random.randInt(this.indexSpecs.length)];

            // We don't assert that the command succeeded when dropping an index because it's
            // possible another thread has already dropped this index.
            db[this.collName].dropIndex(indexSpec);

            // Recreate the index that was dropped.
            assertAlways.commandWorked(db[this.collName].createIndex(indexSpec));
        }
    };

    var transitions = {
        query: {query: 0.8, dropCollection: 0.1, dropIndex: 0.1},
        dropCollection: {query: 1},
        dropIndex: {query: 1}
    };

    function setup(db, collName, cluster) {
        this.indexSpecs.forEach(indexSpec => {
            assertAlways.commandWorked(db[this.collName].createIndex(indexSpec));
        });
    }

    return {
        threadCount: 10,
        iterations: 50,
        data: data,
        states: states,
        startState: 'query',
        transitions: transitions,
        setup: setup
    };

})();


<|start_filename|>src/mongo/shell/dbshell.cpp<|end_filename|>
// dbshell.cpp
/*
 *    Copyright 2010 10gen 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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kDefault

#include "mongo/platform/basic.h"

#include <boost/filesystem/operations.hpp>
#include <fstream>
#include <iostream>
#include <pcrecpp.h>
#include <signal.h>
#include <stdio.h>
#include <string.h>

#include "mongo/base/initializer.h"
#include "mongo/base/status.h"
#include "mongo/client/dbclientinterface.h"
#include "mongo/client/sasl_client_authenticate.h"
#include "mongo/db/client.h"
#include "mongo/db/log_process_details.h"
#include "mongo/db/server_options.h"
#include "mongo/logger/console_appender.h"
#include "mongo/logger/logger.h"
#include "mongo/logger/message_event_utf8_encoder.h"
#include "mongo/scripting/engine.h"
#include "mongo/shell/linenoise.h"
#include "mongo/shell/shell_options.h"
#include "mongo/shell/shell_utils.h"
#include "mongo/shell/shell_utils_launcher.h"
#include "mongo/util/exit_code.h"
#include "mongo/util/file.h"
#include "mongo/util/log.h"
#include "mongo/util/net/ssl_options.h"
#include "mongo/util/password.h"
#include "mongo/util/quick_exit.h"
#include "mongo/util/scopeguard.h"
#include "mongo/util/signal_handlers.h"
#include "mongo/util/stacktrace.h"
#include "mongo/util/startup_test.h"
#include "mongo/util/static_observer.h"
#include "mongo/util/text.h"
#include "mongo/util/version.h"

#ifdef _WIN32
#include <io.h>
#include <shlobj.h>
#define isatty _isatty
#define fileno _fileno
#else
#include <unistd.h>
#endif

using namespace std;
using namespace mongo;

string historyFile;
bool gotInterrupted = false;
bool inMultiLine = false;
static volatile bool atPrompt = false;  // can eval before getting to prompt

namespace mongo {

Scope* shellMainScope;

extern bool dbexitCalled;
}

void generateCompletions(const string& prefix, vector<string>& all) {
    if (prefix.find('"') != string::npos)
        return;

    try {
        BSONObj args = BSON("0" << prefix);
        shellMainScope->invokeSafe(
            "function callShellAutocomplete(x) {shellAutocomplete(x)}", &args, NULL);
        BSONObjBuilder b;
        shellMainScope->append(b, "", "__autocomplete__");
        BSONObj res = b.obj();
        BSONObj arr = res.firstElement().Obj();

        BSONObjIterator i(arr);
        while (i.more()) {
            BSONElement e = i.next();
            all.push_back(e.String());
        }
    } catch (...) {
    }
}

void completionHook(const char* text, linenoiseCompletions* lc) {
    vector<string> all;
    generateCompletions(text, all);

    for (unsigned i = 0; i < all.size(); ++i)
        linenoiseAddCompletion(lc, (char*)all[i].c_str());
}

void shellHistoryInit() {
    stringstream ss;
    const char* h = shell_utils::getUserDir();
    if (h)
        ss << h << "/";
    ss << ".dbshell";
    historyFile = ss.str();

    linenoiseHistoryLoad(historyFile.c_str());
    linenoiseSetCompletionCallback(completionHook);
}

void shellHistoryDone() {
    linenoiseHistorySave(historyFile.c_str());
    linenoiseHistoryFree();
}
void shellHistoryAdd(const char* line) {
    if (line[0] == '\0')
        return;

    // dont record duplicate lines
    static string lastLine;
    if (lastLine == line)
        return;
    lastLine = line;

    // We don't want any .auth() or .createUser() shell helpers added, but we want to
    // be able to add things like `.author`, so be smart about how this is
    // detected by using regular expresions. This is so we can avoid storing passwords
    // in the history file in plaintext.
    static pcrecpp::RE hiddenHelpers(
        "\\.\\s*(auth|createUser|updateUser|changeUserPassword)\\s*\\(");
    // Also don't want the raw user management commands to show in the shell when run directly
    // via runCommand.
    static pcrecpp::RE hiddenCommands(
        "(run|admin)Command\\s*\\(\\s*{\\s*(createUser|updateUser)\\s*:");
    if (!hiddenHelpers.PartialMatch(line) && !hiddenCommands.PartialMatch(line)) {
        linenoiseHistoryAdd(line);
    }
}

void killOps() {
    if (mongo::shell_utils::_nokillop)
        return;

    if (atPrompt)
        return;

    sleepmillis(10);  // give current op a chance to finish

    mongo::shell_utils::connectionRegistry.killOperationsOnAllConnections(
        !shellGlobalParams.autoKillOp);
}

// Stubs for signal_handlers.cpp
namespace mongo {
void logProcessDetailsForLogRotate() {}

void exitCleanly(ExitCode code) {
    {
        stdx::lock_guard<stdx::mutex> lk(mongo::shell_utils::mongoProgramOutputMutex);
        mongo::dbexitCalled = true;
    }

    ::killOps();
    ::shellHistoryDone();
    quickExit(0);
}
}

void quitNicely(int sig) {
    exitCleanly(EXIT_CLEAN);
}

// the returned string is allocated with strdup() or malloc() and must be freed by calling free()
char* shellReadline(const char* prompt, int handlesigint = 0) {
    atPrompt = true;

    char* ret = linenoise(prompt);
    if (!ret) {
        gotInterrupted = true;  // got ^C, break out of multiline
    }

    atPrompt = false;
    return ret;
}

void setupSignals() {
    signal(SIGINT, quitNicely);
}

string fixHost(const std::string& url, const std::string& host, const std::string& port) {
    if (host.size() == 0 && port.size() == 0) {
        if (url.find("/") == string::npos) {
            // check for ips
            if (url.find(".") != string::npos)
                return url + "/test";

            if (url.rfind(":") != string::npos && isdigit(url[url.rfind(":") + 1]))
                return url + "/test";
        }
        return url;
    }

    if (url.find("/") != string::npos) {
        cerr << "url can't have host or port if you specify them individually" << endl;
        quickExit(-1);
    }

    string newurl((host.size() == 0) ? "127.0.0.1" : host);
    if (port.size() > 0)
        newurl += ":" + port;
    else if (host.find(':') == string::npos) {
        // need to add port with IPv6 addresses
        newurl += ":27017";
    }

    newurl += "/" + url;

    return newurl;
}

static string OpSymbols = "~!%^&*-+=|:,<>/?.";

bool isOpSymbol(char c) {
    for (size_t i = 0; i < OpSymbols.size(); i++)
        if (OpSymbols[i] == c)
            return true;
    return false;
}

bool isUseCmd(const std::string& code) {
    string cmd = code;
    if (cmd.find(" ") > 0)
        cmd = cmd.substr(0, cmd.find(" "));
    return cmd == "use";
}

/**
 * Skip over a quoted string, including quotes escaped with backslash
 *
 * @param code      String
 * @param start     Starting position within string, always > 0
 * @param quote     Quote character (single or double quote)
 * @return          Position of ending quote, or code.size() if no quote found
 */
size_t skipOverString(const std::string& code, size_t start, char quote) {
    size_t pos = start;
    while (pos < code.size()) {
        pos = code.find(quote, pos);
        if (pos == std::string::npos) {
            return code.size();
        }
        // We want to break if the quote we found is not escaped, but we need to make sure
        // that the escaping backslash is not itself escaped.  Comparisons of start and pos
        // are to keep us from reading beyond the beginning of the quoted string.
        //
        if (start == pos || code[pos - 1] != '\\' ||  // previous char was backslash
            start == pos - 1 ||
            code[pos - 2] == '\\'  // char before backslash was not another
            ) {
            break;  // The quote we found was not preceded by an unescaped backslash; it is real
        }
        ++pos;  // The quote we found was escaped with backslash, so it doesn't count
    }
    return pos;
}

bool isBalanced(const std::string& code) {
    if (isUseCmd(code))
        return true;  // don't balance "use <dbname>" in case dbname contains special chars
    int curlyBrackets = 0;
    int squareBrackets = 0;
    int parens = 0;
    bool danglingOp = false;

    for (size_t i = 0; i < code.size(); i++) {
        switch (code[i]) {
            case '/':
                if (i + 1 < code.size() && code[i + 1] == '/') {
                    while (i < code.size() && code[i] != '\n')
                        i++;
                }
                continue;
            case '{':
                curlyBrackets++;
                break;
            case '}':
                if (curlyBrackets <= 0)
                    return true;
                curlyBrackets--;
                break;
            case '[':
                squareBrackets++;
                break;
            case ']':
                if (squareBrackets <= 0)
                    return true;
                squareBrackets--;
                break;
            case '(':
                parens++;
                break;
            case ')':
                if (parens <= 0)
                    return true;
                parens--;
                break;
            case '"':
            case '\'':
                i = skipOverString(code, i + 1, code[i]);
                if (i >= code.size()) {
                    return true;  // Do not let unterminated strings enter multi-line mode
                }
                break;
            case '\\':
                if (i + 1 < code.size() && code[i + 1] == '/')
                    i++;
                break;
            case '+':
            case '-':
                if (i + 1 < code.size() && code[i + 1] == code[i]) {
                    i++;
                    continue;  // postfix op (++/--) can't be a dangling op
                }
                break;
        }
        if (i >= code.size()) {
            danglingOp = false;
            break;
        }
        if (isOpSymbol(code[i]))
            danglingOp = true;
        else if (!std::isspace(static_cast<unsigned char>(code[i])))
            danglingOp = false;
    }

    return curlyBrackets == 0 && squareBrackets == 0 && parens == 0 && !danglingOp;
}

struct BalancedTest : public mongo::StartupTest {
public:
    void run() {
        verify(isBalanced("x = 5"));
        verify(isBalanced("function(){}"));
        verify(isBalanced("function(){\n}"));
        verify(!isBalanced("function(){"));
        verify(isBalanced("x = \"{\";"));
        verify(isBalanced("// {"));
        verify(!isBalanced("// \n {"));
        verify(!isBalanced("\"//\" {"));
        verify(isBalanced("{x:/x\\//}"));
        verify(!isBalanced("{ \\/// }"));
        verify(isBalanced("x = 5 + y "));
        verify(!isBalanced("x = "));
        verify(!isBalanced("x = // hello"));
        verify(!isBalanced("x = 5 +"));
        verify(isBalanced(" x ++"));
        verify(isBalanced("-- x"));
        verify(!isBalanced("a."));
        verify(!isBalanced("a. "));
        verify(isBalanced("a.b"));

        // SERVER-5809 and related cases --
        verify(isBalanced("a = {s:\"\\\"\"}"));            // a = {s:"\""}
        verify(isBalanced("db.test.save({s:\"\\\"\"})"));  // db.test.save({s:"\""})
        verify(isBalanced("printjson(\" \\\" \")"));       // printjson(" \" ") -- SERVER-8554
        verify(isBalanced("var a = \"\\\\\";"));           // var a = "\\";
        verify(isBalanced("var a = (\"\\\\\") //\""));     // var a = ("\\") //"
        verify(isBalanced("var a = (\"\\\\\") //\\\""));   // var a = ("\\") //\"
        verify(isBalanced("var a = (\"\\\\\") //"));       // var a = ("\\") //
        verify(isBalanced("var a = (\"\\\\\")"));          // var a = ("\\")
        verify(isBalanced("var a = (\"\\\\\\\"\")"));      // var a = ("\\\"")
        verify(!isBalanced("var a = (\"\\\\\" //\""));     // var a = ("\\" //"
        verify(!isBalanced("var a = (\"\\\\\" //"));       // var a = ("\\" //
        verify(!isBalanced("var a = (\"\\\\\""));          // var a = ("\\"
    }
} balanced_test;

string finishCode(string code) {
    while (!isBalanced(code)) {
        inMultiLine = true;
        code += "\n";
        // cancel multiline if two blank lines are entered
        if (code.find("\n\n\n") != string::npos)
            return ";";
        char* line = shellReadline("... ", 1);
        if (gotInterrupted) {
            if (line)
                free(line);
            return "";
        }
        if (!line)
            return "";

        char* linePtr = line;
        while (str::startsWith(linePtr, "... "))
            linePtr += 4;

        code += linePtr;
        free(line);
    }
    return code;
}

bool execPrompt(mongo::Scope& scope, const char* promptFunction, string& prompt) {
    string execStatement = string("__prompt__ = ") + promptFunction + "();";
    scope.exec("delete __prompt__;", "", false, false, false, 0);
    scope.exec(execStatement, "", false, false, false, 0);
    if (scope.type("__prompt__") == String) {
        prompt = scope.getString("__prompt__");
        return true;
    }
    return false;
}

/**
 * Edit a variable or input buffer text in an external editor -- EDITOR must be defined
 *
 * @param whatToEdit Name of JavaScript variable to be edited, or any text string
 */
static void edit(const string& whatToEdit) {
    // EDITOR may be defined in the JavaScript scope or in the environment
    string editor;
    if (shellMainScope->type("EDITOR") == String) {
        editor = shellMainScope->getString("EDITOR");
    } else {
        static const char* editorFromEnv = getenv("EDITOR");
        if (editorFromEnv) {
            editor = editorFromEnv;
        }
    }
    if (editor.empty()) {
        cout << "please define EDITOR as a JavaScript string or as an environment variable" << endl;
        return;
    }

    // "whatToEdit" might look like a variable/property name
    bool editingVariable = true;
    for (const char* p = whatToEdit.c_str(); *p; ++p) {
        if (!(isalnum(*p) || *p == '_' || *p == '.')) {
            editingVariable = false;
            break;
        }
    }

    string js;
    if (editingVariable) {
        // If "whatToEdit" is undeclared or uninitialized, declare
        int varType = shellMainScope->type(whatToEdit.c_str());
        if (varType == Undefined) {
            shellMainScope->exec("var " + whatToEdit, "(shell)", false, true, false);
        }

        // Convert "whatToEdit" to JavaScript (JSON) text
        if (!shellMainScope->exec(
                "__jsout__ = tojson(" + whatToEdit + ")", "tojs", false, false, false))
            return;  // Error already printed

        js = shellMainScope->getString("__jsout__");

        if (strstr(js.c_str(), "[native code]")) {
            cout << "can't edit native functions" << endl;
            return;
        }
    } else {
        js = whatToEdit;
    }

    // Pick a name to use for the temp file
    string filename;
    const int maxAttempts = 10;
    int i;
    for (i = 0; i < maxAttempts; ++i) {
        StringBuilder sb;
#ifdef _WIN32
        char tempFolder[MAX_PATH];
        GetTempPathA(sizeof tempFolder, tempFolder);
        sb << tempFolder << "mongo_edit" << time(0) + i << ".js";
#else
        sb << "/tmp/mongo_edit" << time(0) + i << ".js";
#endif
        filename = sb.str();
        if (!::mongo::shell_utils::fileExists(filename))
            break;
    }
    if (i == maxAttempts) {
        cout << "couldn't create unique temp file after " << maxAttempts << " attempts" << endl;
        return;
    }

    // Create the temp file
    FILE* tempFileStream;
    tempFileStream = fopen(filename.c_str(), "wt");
    if (!tempFileStream) {
        cout << "couldn't create temp file (" << filename << "): " << errnoWithDescription()
             << endl;
        return;
    }

    // Write JSON into the temp file
    size_t fileSize = js.size();
    if (fwrite(js.data(), sizeof(char), fileSize, tempFileStream) != fileSize) {
        int systemErrno = errno;
        cout << "failed to write to temp file: " << errnoWithDescription(systemErrno) << endl;
        fclose(tempFileStream);
        remove(filename.c_str());
        return;
    }
    fclose(tempFileStream);

    // Pass file to editor
    StringBuilder sb;
    sb << editor << " " << filename;
    int ret = ::system(sb.str().c_str());
    if (ret) {
        if (ret == -1) {
            int systemErrno = errno;
            cout << "failed to launch $EDITOR (" << editor
                 << "): " << errnoWithDescription(systemErrno) << endl;
        } else
            cout << "editor exited with error (" << ret << "), not applying changes" << endl;
        remove(filename.c_str());
        return;
    }

    // The editor gave return code zero, so read the file back in
    tempFileStream = fopen(filename.c_str(), "rt");
    if (!tempFileStream) {
        cout << "couldn't open temp file on return from editor: " << errnoWithDescription() << endl;
        remove(filename.c_str());
        return;
    }
    sb.reset();
    int bytes;
    do {
        char buf[1024];
        bytes = fread(buf, sizeof(char), sizeof buf, tempFileStream);
        if (ferror(tempFileStream)) {
            cout << "failed to read temp file: " << errnoWithDescription() << endl;
            fclose(tempFileStream);
            remove(filename.c_str());
            return;
        }
        sb.append(StringData(buf, bytes));
    } while (bytes);

    // Done with temp file, close and delete it
    fclose(tempFileStream);
    remove(filename.c_str());

    if (editingVariable) {
        // Try to execute assignment to copy edited value back into the variable
        const string code = whatToEdit + string(" = ") + sb.str();
        if (!shellMainScope->exec(code, "tojs", false, true, false)) {
            cout << "error executing assignment: " << code << endl;
        }
    } else {
        linenoisePreloadBuffer(sb.str().c_str());
    }
}

int _main(int argc, char* argv[], char** envp) {
    setupSignalHandlers();
    setupSignals();

    mongo::shell_utils::RecordMyLocation(argv[0]);

    shellGlobalParams.url = "test";

    mongo::runGlobalInitializersOrDie(argc, argv, envp);

    // hide password from ps output
    for (int i = 0; i < (argc - 1); ++i) {
        if (!strcmp(argv[i], "-p") || !strcmp(argv[i], "--password")) {
            char* arg = argv[i + 1];
            while (*arg) {
                *arg++ = 'x';
            }
        }
    }

    if (!mongo::serverGlobalParams.quiet)
        cout << "MongoDB shell version: " << mongo::versionString << endl;

    mongo::StartupTest::runTests();

    logger::globalLogManager()
        ->getNamedDomain("javascriptOutput")
        ->attachAppender(logger::MessageLogDomain::AppenderAutoPtr(
            new logger::ConsoleAppender<logger::MessageEventEphemeral>(
                new logger::MessageEventUnadornedEncoder)));

    if (!shellGlobalParams.nodb) {  // connect to db
        stringstream ss;
        if (mongo::serverGlobalParams.quiet)
            ss << "__quiet = true;";
        ss << "db = connect( \""
           << fixHost(shellGlobalParams.url, shellGlobalParams.dbhost, shellGlobalParams.port)
           << "\")";

        mongo::shell_utils::_dbConnect = ss.str();

        if (shellGlobalParams.usingPassword && shellGlobalParams.password.empty()) {
            shellGlobalParams.password = mongo::askPassword();
        }
    }

    // Construct the authentication-related code to execute on shell startup.
    //
    // This constructs and immediately executes an anonymous function, to avoid
    // the shell's default behavior of printing statement results to the console.
    //
    // It constructs a statement of the following form:
    //
    // (function() {
    //    // Set default authentication mechanism and, maybe, authenticate.
    //  }())
    stringstream authStringStream;
    authStringStream << "(function() { " << endl;
    if (!shellGlobalParams.authenticationMechanism.empty()) {
        authStringStream << "DB.prototype._defaultAuthenticationMechanism = \""
                         << escape(shellGlobalParams.authenticationMechanism) << "\";" << endl;
    }

    if (!shellGlobalParams.gssapiServiceName.empty()) {
        authStringStream << "DB.prototype._defaultGssapiServiceName = \""
                         << escape(shellGlobalParams.gssapiServiceName) << "\";" << endl;
    }

    if (!shellGlobalParams.nodb && shellGlobalParams.username.size()) {
        authStringStream << "var username = \"" << escape(shellGlobalParams.username) << "\";"
                         << endl;
        if (shellGlobalParams.usingPassword) {
            authStringStream << "var password = \"" << escape(shellGlobalParams.password) << "\";"
                             << endl;
        }
        if (shellGlobalParams.authenticationDatabase.empty()) {
            authStringStream << "var authDb = db;" << endl;
        } else {
            authStringStream << "var authDb = db.getSiblingDB(\""
                             << escape(shellGlobalParams.authenticationDatabase) << "\");" << endl;
        }
        authStringStream << "authDb._authOrThrow({ " << saslCommandUserFieldName << ": username ";
        if (shellGlobalParams.usingPassword) {
            authStringStream << ", " << saslCommandPasswordFieldName << ": password ";
        }

        if (!shellGlobalParams.gssapiHostName.empty()) {
            authStringStream << ", " << saslCommandServiceHostnameFieldName << ": \""
                             << escape(shellGlobalParams.gssapiHostName) << '"' << endl;
        }
        authStringStream << "});" << endl;
    }
    authStringStream << "}())";
    mongo::shell_utils::_dbAuth = authStringStream.str();

    mongo::ScriptEngine::setConnectCallback(mongo::shell_utils::onConnect);
    mongo::ScriptEngine::setup();
    mongo::globalScriptEngine->setScopeInitCallback(mongo::shell_utils::initScope);
    mongo::globalScriptEngine->enableJIT(!shellGlobalParams.nojit);
    mongo::globalScriptEngine->enableJavaScriptProtection(shellGlobalParams.javascriptProtection);

    auto poolGuard = MakeGuard([] { ScriptEngine::dropScopeCache(); });

    unique_ptr<mongo::Scope> scope(mongo::globalScriptEngine->newScope());
    shellMainScope = scope.get();

    if (shellGlobalParams.runShell)
        cout << "type \"help\" for help" << endl;

    // Load and execute /etc/mongorc.js before starting shell
    std::string rcGlobalLocation;
#ifndef _WIN32
    rcGlobalLocation = "/etc/mongorc.js";
#else
    wchar_t programDataPath[MAX_PATH];
    if (S_OK == SHGetFolderPathW(NULL, CSIDL_COMMON_APPDATA, NULL, 0, programDataPath)) {
        rcGlobalLocation = str::stream() << toUtf8String(programDataPath)
                                         << "\\MongoDB\\mongorc.js";
    }
#endif
    if (!rcGlobalLocation.empty() && ::mongo::shell_utils::fileExists(rcGlobalLocation)) {
        if (!scope->execFile(rcGlobalLocation, false, true)) {
            cout << "The \"" << rcGlobalLocation << "\" file could not be executed" << endl;
        }
    }

    if (!shellGlobalParams.script.empty()) {
        mongo::shell_utils::MongoProgramScope s;
        if (!scope->exec(shellGlobalParams.script, "(shell eval)", false, true, false))
            return -4;
        scope->exec("shellPrintHelper( __lastres__ );", "(shell2 eval)", true, true, false);
    }

    for (size_t i = 0; i < shellGlobalParams.files.size(); ++i) {
        mongo::shell_utils::MongoProgramScope s;

        if (shellGlobalParams.files.size() > 1)
            cout << "loading file: " << shellGlobalParams.files[i] << endl;

        if (!scope->execFile(shellGlobalParams.files[i], false, true)) {
            cout << "failed to load: " << shellGlobalParams.files[i] << endl;
            return -3;
        }
        if (mongo::shell_utils::KillMongoProgramInstances() != EXIT_SUCCESS) {
            return -3;
        }
    }

    if (shellGlobalParams.files.size() == 0 && shellGlobalParams.script.empty())
        shellGlobalParams.runShell = true;

    bool lastLineSuccessful = true;
    if (shellGlobalParams.runShell) {
        mongo::shell_utils::MongoProgramScope s;
        // If they specify norc, assume it's not their first time
        bool hasMongoRC = shellGlobalParams.norc;
        string rcLocation;
        if (!shellGlobalParams.norc) {
#ifndef _WIN32
            if (getenv("HOME") != NULL)
                rcLocation = str::stream() << getenv("HOME") << "/.mongorc.js";
#else
            if (getenv("HOMEDRIVE") != NULL && getenv("HOMEPATH") != NULL)
                rcLocation = str::stream() << toUtf8String(_wgetenv(L"HOMEDRIVE"))
                                           << toUtf8String(_wgetenv(L"HOMEPATH"))
                                           << "\\.mongorc.js";
#endif
            if (!rcLocation.empty() && ::mongo::shell_utils::fileExists(rcLocation)) {
                hasMongoRC = true;
                if (!scope->execFile(rcLocation, false, true)) {
                    cout << "The \".mongorc.js\" file located in your home folder could not be "
                            "executed" << endl;
                    return -5;
                }
            }
        }

        if (!hasMongoRC && isatty(fileno(stdin))) {
            cout
                << "Welcome to the MongoDB shell.\n"
                   "For interactive help, type \"help\".\n"
                   "For more comprehensive documentation, see\n\thttp://docs.mongodb.org/\n"
                   "Questions? Try the support group\n\thttp://groups.google.com/group/mongodb-user"
                << endl;
            File f;
            f.open(rcLocation.c_str(), false);  // Create empty .mongorc.js file
        }

        if (!shellGlobalParams.nodb && !mongo::serverGlobalParams.quiet && isatty(fileno(stdin))) {
            scope->exec(
                "shellHelper( 'show', 'startupWarnings' )", "(shellwarnings", false, true, false);
        }

        shellHistoryInit();

        string prompt;
        int promptType;

        while (1) {
            inMultiLine = false;
            gotInterrupted = false;

            promptType = scope->type("prompt");
            if (promptType == String) {
                prompt = scope->getString("prompt");
            } else if ((promptType == Code) && execPrompt(*scope, "prompt", prompt)) {
            } else if (execPrompt(*scope, "defaultPrompt", prompt)) {
            } else {
                prompt = "> ";
            }

            char* line = shellReadline(prompt.c_str());

            char* linePtr = line;  // can't clobber 'line', we need to free() it later
            if (linePtr) {
                while (linePtr[0] == ' ')
                    ++linePtr;
                int lineLen = strlen(linePtr);
                while (lineLen > 0 && linePtr[lineLen - 1] == ' ')
                    linePtr[--lineLen] = 0;
            }

            if (!linePtr || (strlen(linePtr) == 4 && strstr(linePtr, "exit"))) {
                if (!mongo::serverGlobalParams.quiet)
                    cout << "bye" << endl;
                if (line)
                    free(line);
                break;
            }

            string code = linePtr;
            if (code == "exit" || code == "exit;") {
                free(line);
                break;
            }
            if (code == "cls") {
                free(line);
                linenoiseClearScreen();
                continue;
            }

            if (code.size() == 0) {
                free(line);
                continue;
            }

            if (str::startsWith(linePtr, "edit ")) {
                shellHistoryAdd(linePtr);

                const char* s = linePtr + 5;  // skip "edit "
                while (*s && isspace(*s))
                    s++;

                edit(s);
                free(line);
                continue;
            }

            gotInterrupted = false;
            code = finishCode(code);
            if (gotInterrupted) {
                cout << endl;
                free(line);
                continue;
            }

            if (code.size() == 0) {
                free(line);
                break;
            }

            bool wascmd = false;
            {
                string cmd = linePtr;
                string::size_type firstSpace;
                if ((firstSpace = cmd.find(" ")) != string::npos)
                    cmd = cmd.substr(0, firstSpace);

                if (cmd.find("\"") == string::npos) {
                    try {
                        lastLineSuccessful =
                            scope->exec((string) "__iscmd__ = shellHelper[\"" + cmd + "\"];",
                                        "(shellhelp1)",
                                        false,
                                        true,
                                        true);
                        if (scope->getBoolean("__iscmd__")) {
                            lastLineSuccessful =
                                scope->exec((string) "shellHelper( \"" + cmd + "\" , \"" +
                                                code.substr(cmd.size()) + "\");",
                                            "(shellhelp2)",
                                            false,
                                            true,
                                            false);
                            wascmd = true;
                        }
                    } catch (std::exception& e) {
                        cout << "error2:" << e.what() << endl;
                        wascmd = true;
                        lastLineSuccessful = false;
                    }
                }
            }

            if (!wascmd) {
                try {
                    lastLineSuccessful = scope->exec(code.c_str(), "(shell)", false, true, false);
                    if (lastLineSuccessful) {
                        scope->exec(
                            "shellPrintHelper( __lastres__ );", "(shell2)", true, true, false);
                    }
                } catch (std::exception& e) {
                    cout << "error:" << e.what() << endl;
                    lastLineSuccessful = false;
                }
            }

            shellHistoryAdd(code.c_str());
            free(line);
        }

        shellHistoryDone();
    }

    {
        stdx::lock_guard<stdx::mutex> lk(mongo::shell_utils::mongoProgramOutputMutex);
        mongo::dbexitCalled = true;
    }
    return (lastLineSuccessful ? 0 : 1);
}

#ifdef _WIN32
int wmain(int argc, wchar_t* argvW[], wchar_t* envpW[]) {
    static mongo::StaticObserver staticObserver;
    int returnCode;
    try {
        WindowsCommandLine wcl(argc, argvW, envpW);
        returnCode = _main(argc, wcl.argv(), wcl.envp());
    } catch (mongo::DBException& e) {
        cerr << "exception: " << e.what() << endl;
        returnCode = 1;
    }
    quickExit(returnCode);
}
#else   // #ifdef _WIN32
int main(int argc, char* argv[], char** envp) {
    static mongo::StaticObserver staticObserver;
    int returnCode;
    try {
        returnCode = _main(argc, argv, envp);
    } catch (mongo::DBException& e) {
        cerr << "exception: " << e.what() << endl;
        returnCode = 1;
    }
    quickExit(returnCode);
}
#endif  // #ifdef _WIN32


<|start_filename|>src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h<|end_filename|>
// ephemeral_for_test_record_store.h

/**
*    Copyright (C) 2014 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 <http://www.gnu.org/licenses/>.
*
*    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.
*/

#pragma once

#include <boost/shared_array.hpp>
#include <map>

#include "mongo/db/storage/capped_callback.h"
#include "mongo/db/storage/record_store.h"

namespace mongo {

/**
 * A RecordStore that stores all data in-memory.
 *
 * @param cappedMaxSize - required if isCapped. limit uses dataSize() in this impl.
 */
class EphemeralForTestRecordStore : public RecordStore {
public:
    explicit EphemeralForTestRecordStore(StringData ns,
                                         std::shared_ptr<void>* dataInOut,
                                         bool isCapped = false,
                                         int64_t cappedMaxSize = -1,
                                         int64_t cappedMaxDocs = -1,
                                         CappedCallback* cappedCallback = nullptr);

    virtual const char* name() const;

    virtual RecordData dataFor(OperationContext* txn, const RecordId& loc) const;

    virtual bool findRecord(OperationContext* txn, const RecordId& loc, RecordData* rd) const;

    virtual void deleteRecord(OperationContext* txn, const RecordId& dl);

    virtual StatusWith<RecordId> insertRecord(OperationContext* txn,
                                              const char* data,
                                              int len,
                                              bool enforceQuota);

    virtual StatusWith<RecordId> insertRecord(OperationContext* txn,
                                              const DocWriter* doc,
                                              bool enforceQuota);

    virtual StatusWith<RecordId> updateRecord(OperationContext* txn,
                                              const RecordId& oldLocation,
                                              const char* data,
                                              int len,
                                              bool enforceQuota,
                                              UpdateNotifier* notifier);

    virtual bool updateWithDamagesSupported() const;

    virtual StatusWith<RecordData> updateWithDamages(OperationContext* txn,
                                                     const RecordId& loc,
                                                     const RecordData& oldRec,
                                                     const char* damageSource,
                                                     const mutablebson::DamageVector& damages);

    std::unique_ptr<SeekableRecordCursor> getCursor(OperationContext* txn,
                                                    bool forward) const final;

    virtual Status truncate(OperationContext* txn);

    virtual void temp_cappedTruncateAfter(OperationContext* txn, RecordId end, bool inclusive);

    virtual Status validate(OperationContext* txn,
                            bool full,
                            bool scanData,
                            ValidateAdaptor* adaptor,
                            ValidateResults* results,
                            BSONObjBuilder* output);

    virtual void appendCustomStats(OperationContext* txn,
                                   BSONObjBuilder* result,
                                   double scale) const;

    virtual Status touch(OperationContext* txn, BSONObjBuilder* output) const;

    virtual void increaseStorageSize(OperationContext* txn, int size, bool enforceQuota);

    virtual int64_t storageSize(OperationContext* txn,
                                BSONObjBuilder* extraInfo = NULL,
                                int infoLevel = 0) const;

    virtual long long dataSize(OperationContext* txn) const {
        return _data->dataSize;
    }

    virtual long long numRecords(OperationContext* txn) const {
        return _data->records.size();
    }

    virtual boost::optional<RecordId> oplogStartHack(OperationContext* txn,
                                                     const RecordId& startingPosition) const;

    void waitForAllEarlierOplogWritesToBeVisible(OperationContext* txn) const override {}

    virtual void updateStatsAfterRepair(OperationContext* txn,
                                        long long numRecords,
                                        long long dataSize) {
        invariant(_data->records.size() == size_t(numRecords));
        _data->dataSize = dataSize;
    }

protected:
    struct EphemeralForTestRecord {
        EphemeralForTestRecord() : size(0) {}
        EphemeralForTestRecord(int size) : size(size), data(new char[size]) {}

        RecordData toRecordData() const {
            return RecordData(data.get(), size);
        }

        int size;
        boost::shared_array<char> data;
    };

    virtual const EphemeralForTestRecord* recordFor(const RecordId& loc) const;
    virtual EphemeralForTestRecord* recordFor(const RecordId& loc);

public:
    //
    // Not in RecordStore interface
    //

    typedef std::map<RecordId, EphemeralForTestRecord> Records;

    bool isCapped() const {
        return _isCapped;
    }
    void setCappedCallback(CappedCallback* cb) {
        _cappedCallback = cb;
    }
    bool cappedMaxDocs() const {
        invariant(_isCapped);
        return _cappedMaxDocs;
    }
    bool cappedMaxSize() const {
        invariant(_isCapped);
        return _cappedMaxSize;
    }

private:
    class InsertChange;
    class RemoveChange;
    class TruncateChange;

    class Cursor;
    class ReverseCursor;

    StatusWith<RecordId> extractAndCheckLocForOplog(const char* data, int len) const;

    RecordId allocateLoc();
    bool cappedAndNeedDelete(OperationContext* txn) const;
    void cappedDeleteAsNeeded(OperationContext* txn);

    // TODO figure out a proper solution to metadata
    const bool _isCapped;
    const int64_t _cappedMaxSize;
    const int64_t _cappedMaxDocs;
    CappedCallback* _cappedCallback;

    // This is the "persistent" data.
    struct Data {
        Data(bool isOplog) : dataSize(0), nextId(1), isOplog(isOplog) {}

        int64_t dataSize;
        Records records;
        int64_t nextId;
        const bool isOplog;
    };

    Data* const _data;
};

}  // namespace mongo


<|start_filename|>jstests/hooks/run_validate_collections.js<|end_filename|>
// Runner for validateCollections that runs full validation on all collections when loaded into
// the mongo shell.
'use strict';

(function() {
    assert.eq(typeof db, 'object', 'Invalid `db` object, is the shell connected to a mongod?');
    load('jstests/hooks/validate_collections.js');  // For validateCollections

    var dbNames = db.getMongo().getDBNames();

    for (var dbName of dbNames) {
        if (!validateCollections(db.getSiblingDB(dbName), {full: true})) {
            throw new Error('Collection validation failed');
        }
    }
})();

<|start_filename|>jstests/sharding/sharding_balance4.js<|end_filename|>
// Check that doing updates done during a migrate all go to the right place
(function() {

    var s = new ShardingTest({
        name: "slow_sharding_balance4",
        shards: 2,
        mongos: 1,
        other: {chunkSize: 1, enableAutoSplit: true}
    });

    s.adminCommand({enablesharding: "test"});
    s.ensurePrimaryShard('test', 'shard0001');
    s.adminCommand({shardcollection: "test.foo", key: {_id: 1}});
    assert.eq(1, s.config.chunks.count(), "setup1");

    s.config.settings.find().forEach(printjson);

    db = s.getDB("test");

    bigString = "";
    while (bigString.length < 10000)
        bigString += "asdasdasdasdadasdasdasdasdasdasdasdasda";

    N = 3000;

    num = 0;

    counts = {};

    //
    // TODO: Rewrite to make much clearer.
    //
    // The core behavior of this test is to add a bunch of documents to a sharded collection, then
    // incrementally update each document and make sure the counts in the document match our update
    // counts while balancing occurs (doUpdate()).  Every once in a while we also check (check())
    // our counts via a query.
    //
    // If during a chunk migration an update is missed, we trigger an assertion and fail.
    //

    function doUpdate(bulk, includeString, optionalId) {
        var up = {
            $inc: {x: 1}
        };
        if (includeString)
            up["$set"] = {
                s: bigString
            };
        var myid = optionalId == undefined ? Random.randInt(N) : optionalId;
        bulk.find({_id: myid}).upsert().update(up);

        counts[myid] = (counts[myid] ? counts[myid] : 0) + 1;
        return myid;
    }

    // Initially update all documents from 1 to N, otherwise later checks can fail because no
    // document
    // previously existed
    var bulk = db.foo.initializeUnorderedBulkOp();
    for (i = 0; i < N; i++) {
        doUpdate(bulk, true, i);
    }

    for (i = 0; i < N * 9; i++) {
        doUpdate(bulk, false);
    }
    assert.writeOK(bulk.execute());

    for (var i = 0; i < 50; i++) {
        s.printChunks("test.foo");
        if (check("initial:" + i, true))
            break;
        sleep(5000);
    }
    check("initial at end");

    assert.lt(20, s.config.chunks.count(), "setup2");

    function check(msg, dontAssert) {
        for (var x in counts) {
            var e = counts[x];
            var z = db.foo.findOne({_id: parseInt(x)});

            if (z && z.x == e)
                continue;

            if (dontAssert) {
                if (z)
                    delete z.s;
                print("not asserting for key failure: " + x + " want: " + e + " got: " + tojson(z));
                return false;
            }

            s.s.getDB("admin").runCommand({setParameter: 1, logLevel: 2});

            printjson(db.foo.findOne({_id: parseInt(x)}));

            var y = db.foo.findOne({_id: parseInt(x)});

            if (y) {
                delete y.s;
            }

            s.printChunks("test.foo");

            assert(z, "couldn't find : " + x + " y:" + tojson(y) + " e: " + e + " " + msg);
            assert.eq(e, z.x, "count for : " + x + " y:" + tojson(y) + " " + msg);
        }

        return true;
    }

    function diff1() {
        jsTest.log("Running diff1...");

        bulk = db.foo.initializeUnorderedBulkOp();
        var myid = doUpdate(bulk, false);
        var res = assert.writeOK(bulk.execute());

        assert.eq(1,
                  res.nModified,
                  "diff myid: " + myid + " 2: " + res.toString() + "\n" + " correct count is: " +
                      counts[myid] + " db says count is: " + tojson(db.foo.findOne({_id: myid})));

        var x = s.chunkCounts("foo");
        if (Math.random() > .999)
            printjson(x);
        return Math.max(x.shard0000, x.shard0001) - Math.min(x.shard0000, x.shard0001);
    }

    assert.lt(20, diff1(), "initial load");
    print(diff1());

    s.startBalancer();

    assert.soon(function() {
        var d = diff1();
        return d < 5;
    }, "balance didn't happen", 1000 * 60 * 20, 1);

    s.stop();

})();


<|start_filename|>src/mongo/db/repl/minvalid.cpp<|end_filename|>
/**
 *    Copyright (C) 2014 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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplication

#include "mongo/platform/basic.h"

#include "mongo/db/repl/minvalid.h"

#include "mongo/db/concurrency/d_concurrency.h"
#include "mongo/db/concurrency/write_conflict_exception.h"
#include "mongo/db/dbhelpers.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/operation_context.h"
#include "mongo/db/operation_context_impl.h"
#include "mongo/db/repl/oplog.h"
#include "mongo/db/repl/replication_coordinator_global.h"
#include "mongo/util/assert_util.h"
#include "mongo/util/log.h"

namespace mongo {
namespace repl {

namespace {
const char kInitialSyncFlagFieldName[] = "doingInitialSync";
const BSONObj kInitialSyncFlag(BSON(kInitialSyncFlagFieldName << true));
NamespaceString minValidNss("local.replset.minvalid");
const char kBeginFieldName[] = "begin";
const char kOplogDeleteFromPointFieldName[] = "oplogDeleteFromPoint";

BSONObj getMinValidDocument(OperationContext* txn) {
    MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
        ScopedTransaction transaction(txn, MODE_IS);
        Lock::DBLock dblk(txn->lockState(), minValidNss.db(), MODE_IS);
        Lock::CollectionLock lk(txn->lockState(), minValidNss.ns(), MODE_IS);
        BSONObj doc;
        bool found = Helpers::getSingleton(txn, minValidNss.ns().c_str(), doc);
        invariant(found || doc.isEmpty());
        return doc;
    }
    MONGO_WRITE_CONFLICT_RETRY_LOOP_END(txn, "getMinValidDocument", minValidNss.ns());

    MONGO_UNREACHABLE;
}

void updateMinValidDocument(OperationContext* txn, const BSONObj& updateSpec) {
    MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
        ScopedTransaction transaction(txn, MODE_IX);
        // For now this needs to be MODE_X because it sometimes creates the collection.
        Lock::DBLock dblk(txn->lockState(), minValidNss.db(), MODE_X);
        Helpers::putSingleton(txn, minValidNss.ns().c_str(), updateSpec);
    }
    MONGO_WRITE_CONFLICT_RETRY_LOOP_END(txn, "updateMinValidDocument", minValidNss.ns());
}
}  // namespace

// Writes
void clearInitialSyncFlag(OperationContext* txn) {
    auto replCoord = repl::ReplicationCoordinator::get(txn);
    OpTime time = replCoord->getMyLastAppliedOpTime();
    updateMinValidDocument(txn,
                           BSON("$unset"
                                << kInitialSyncFlag << "$set"
                                << BSON("ts" << time.getTimestamp() << "t" << time.getTerm()
                                             << kBeginFieldName << time.toBSON())));
    txn->recoveryUnit()->waitUntilDurable();
    replCoord->setMyLastDurableOpTime(time);
    LOG(3) << "clearing initial sync flag";
}

void setInitialSyncFlag(OperationContext* txn) {
    updateMinValidDocument(txn, BSON("$set" << kInitialSyncFlag));
    txn->recoveryUnit()->waitUntilDurable();
    LOG(3) << "setting initial sync flag";
}

bool getInitialSyncFlag() {
    OperationContextImpl txn;
    return getInitialSyncFlag(&txn);
}
bool getInitialSyncFlag(OperationContext* txn) {
    const BSONObj doc = getMinValidDocument(txn);
    const auto flag = doc[kInitialSyncFlagFieldName].trueValue();
    LOG(3) << "returning initial sync flag value of " << flag;
    return flag;
}

OpTime getMinValid(OperationContext* txn) {
    const BSONObj doc = getMinValidDocument(txn);
    const auto opTimeStatus = OpTime::parseFromOplogEntry(doc);
    // If any of the keys (fields) are missing from the minvalid document, we return
    // a null OpTime.
    if (opTimeStatus == ErrorCodes::NoSuchKey) {
        return {};
    }

    if (!opTimeStatus.isOK()) {
        severe() << "Error parsing minvalid entry: " << doc
                 << ", with status:" << opTimeStatus.getStatus();
        fassertFailedNoTrace(40052);
    }

    OpTime minValid = opTimeStatus.getValue();
    LOG(3) << "returning minvalid: " << minValid.toString() << "(" << minValid.toBSON() << ")";

    return minValid;
}
void setMinValid(OperationContext* txn, const OpTime& minValid) {
    LOG(3) << "setting minvalid to exactly: " << minValid.toString() << "(" << minValid.toBSON()
           << ")";
    updateMinValidDocument(
        txn, BSON("$set" << BSON("ts" << minValid.getTimestamp() << "t" << minValid.getTerm())));
}

void setMinValidToAtLeast(OperationContext* txn, const OpTime& minValid) {
    LOG(3) << "setting minvalid to at least: " << minValid.toString() << "(" << minValid.toBSON()
           << ")";
    updateMinValidDocument(
        txn, BSON("$max" << BSON("ts" << minValid.getTimestamp() << "t" << minValid.getTerm())));
}

void setOplogDeleteFromPoint(OperationContext* txn, const Timestamp& timestamp) {
    LOG(3) << "setting oplog delete from point to: " << timestamp.toStringPretty();
    updateMinValidDocument(txn, BSON("$set" << BSON(kOplogDeleteFromPointFieldName << timestamp)));
}

Timestamp getOplogDeleteFromPoint(OperationContext* txn) {
    const BSONObj doc = getMinValidDocument(txn);
    Timestamp out = {};
    if (auto field = doc[kOplogDeleteFromPointFieldName]) {
        out = field.timestamp();
    }

    LOG(3) << "returning oplog delete from point: " << out;
    return out;
}

void setAppliedThrough(OperationContext* txn, const OpTime& optime) {
    LOG(3) << "setting appliedThrough to: " << optime.toString() << "(" << optime.toBSON() << ")";
    if (optime.isNull()) {
        updateMinValidDocument(txn, BSON("$unset" << BSON(kBeginFieldName << 1)));
    } else {
        updateMinValidDocument(txn, BSON("$set" << BSON(kBeginFieldName << optime.toBSON())));
    }
}

OpTime getAppliedThrough(OperationContext* txn) {
    const BSONObj doc = getMinValidDocument(txn);
    const auto opTimeStatus = OpTime::parseFromOplogEntry(doc.getObjectField(kBeginFieldName));
    if (!opTimeStatus.isOK()) {
        // Return null OpTime on any parse failure, including if "begin" is missing.
        return {};
    }

    OpTime appliedThrough = opTimeStatus.getValue();
    LOG(3) << "returning appliedThrough: " << appliedThrough.toString() << "("
           << appliedThrough.toBSON() << ")";

    return appliedThrough;
}

}  // namespace repl
}  // namespace mongo


<|start_filename|>jstests/hooks/validate_collections.js<|end_filename|>
// Wrapper around the validate command that can be used to validate index key counts.
'use strict';

function validateCollections(db, obj) {
    function dumpCollection(coll, limit) {
        print('Printing indexes in: ' + coll.getFullName());
        printjson(coll.getIndexes());

        print('Printing the first ' + limit + ' documents in: ' + coll.getFullName());
        var res = coll.find().limit(limit);
        while (res.hasNext()) {
            printjson(res.next());
        }
    }

    assert.eq(typeof db, 'object', 'Invalid `db` object, is the shell connected to a mongod?');
    assert.eq(typeof obj, 'object', 'The `obj` argument must be an object');
    assert(obj.hasOwnProperty('full'), 'Please specify whether to use full validation');

    var full = obj.full;

    var success = true;
    var collNames = db.getCollectionNames();
    for (var collName of collNames) {
        var coll = db.getCollection(collName);
        var res = coll.validate(full);

        if (!res.ok || !res.valid) {
            print('Collection validation failed with response: ' + tojson(res));
            dumpCollection(coll, 100);
            success = false;
        }
    }
    return success;
}


<|start_filename|>src/mongo/s/balance.cpp<|end_filename|>
/**
*    Copyright (C) 2008 10gen 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 <http://www.gnu.org/licenses/>.
*
*    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.
*/

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding

#include "mongo/platform/basic.h"

#include "mongo/s/balance.h"

#include <algorithm>

#include "mongo/client/dbclientcursor.h"
#include "mongo/client/remote_command_targeter.h"
#include "mongo/db/client.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/namespace_string.h"
#include "mongo/db/server_options.h"
#include "mongo/db/write_concern.h"
#include "mongo/db/write_concern_options.h"
#include "mongo/s/balancer_policy.h"
#include "mongo/s/catalog/catalog_cache.h"
#include "mongo/s/catalog/catalog_manager.h"
#include "mongo/s/catalog/type_chunk.h"
#include "mongo/s/catalog/type_collection.h"
#include "mongo/s/catalog/type_mongos.h"
#include "mongo/s/catalog/type_settings.h"
#include "mongo/s/catalog/type_tags.h"
#include "mongo/s/chunk_manager.h"
#include "mongo/s/config.h"
#include "mongo/s/grid.h"
#include "mongo/s/client/shard.h"
#include "mongo/s/client/shard_registry.h"
#include "mongo/util/exit.h"
#include "mongo/util/fail_point_service.h"
#include "mongo/util/log.h"
#include "mongo/util/timer.h"
#include "mongo/util/version.h"

namespace mongo {

using std::map;
using std::set;
using std::shared_ptr;
using std::string;
using std::unique_ptr;
using std::vector;

namespace {

/**
 * Utility class to generate timing and statistics for a single balancer round.
 */
class BalanceRoundDetails {
public:
    BalanceRoundDetails() : _executionTimer() {}

    void setSucceeded(int candidateChunks, int chunksMoved) {
        invariant(!_errMsg);
        _candidateChunks = candidateChunks;
        _chunksMoved = chunksMoved;
    }

    void setFailed(const string& errMsg) {
        _errMsg = errMsg;
    }

    BSONObj toBSON() const {
        BSONObjBuilder builder;
        builder.append("executionTimeMillis", _executionTimer.millis());
        builder.append("errorOccured", _errMsg.is_initialized());

        if (_errMsg) {
            builder.append("errmsg", *_errMsg);
        } else {
            builder.append("candidateChunks", _candidateChunks);
            builder.append("chunksMoved", _chunksMoved);
        }

        return builder.obj();
    }

private:
    const Timer _executionTimer;

    // Set only on success
    int _candidateChunks{0};
    int _chunksMoved{0};

    // Set only on failure
    boost::optional<std::string> _errMsg;
};

}  // namespace

MONGO_FP_DECLARE(skipBalanceRound);
MONGO_FP_DECLARE(balancerRoundIntervalSetting);

namespace {
const Seconds kBalanceRoundDefaultInterval(10);
const Seconds kShortBalanceRoundInterval(1);
}

Balancer balancer;

Balancer::Balancer() : _balancedLastTime(0), _policy(new BalancerPolicy()) {}

Balancer::~Balancer() = default;

int Balancer::_moveChunks(OperationContext* txn,
                          const vector<shared_ptr<MigrateInfo>>& candidateChunks,
                          const WriteConcernOptions* writeConcern,
                          bool waitForDelete) {
    int movedCount = 0;

    for (const auto& migrateInfo : candidateChunks) {
        // If the balancer was disabled since we started this round, don't start new chunks
        // moves.
        const auto balSettingsResult =
            grid.catalogManager(txn)->getGlobalSettings(txn, SettingsType::BalancerDocKey);

        const bool isBalSettingsAbsent =
            balSettingsResult.getStatus() == ErrorCodes::NoMatchingDocument;

        if (!balSettingsResult.isOK() && !isBalSettingsAbsent) {
            warning() << balSettingsResult.getStatus();
            return movedCount;
        }

        const SettingsType& balancerConfig =
            isBalSettingsAbsent ? SettingsType{} : balSettingsResult.getValue();

        if ((!isBalSettingsAbsent && !grid.shouldBalance(balancerConfig)) ||
            MONGO_FAIL_POINT(skipBalanceRound)) {
            LOG(1) << "Stopping balancing round early as balancing was disabled";
            return movedCount;
        }

        // Changes to metadata, borked metadata, and connectivity problems between shards
        // should cause us to abort this chunk move, but shouldn't cause us to abort the entire
        // round of chunks.
        //
        // TODO(spencer): We probably *should* abort the whole round on issues communicating
        // with the config servers, but its impossible to distinguish those types of failures
        // at the moment.
        //
        // TODO: Handle all these things more cleanly, since they're expected problems

        const NamespaceString nss(migrateInfo->ns);

        try {
            shared_ptr<DBConfig> cfg =
                uassertStatusOK(grid.catalogCache()->getDatabase(txn, nss.db().toString()));

            // NOTE: We purposely do not reload metadata here, since _doBalanceRound already
            // tried to do so once.
            shared_ptr<ChunkManager> cm = cfg->getChunkManager(txn, migrateInfo->ns);
            uassert(28628,
                    str::stream()
                        << "Collection " << migrateInfo->ns
                        << " was deleted while balancing was active. Aborting balancing round.",
                    cm);

            ChunkPtr c = cm->findIntersectingChunk(txn, migrateInfo->chunk.min);

            if (c->getMin().woCompare(migrateInfo->chunk.min) ||
                c->getMax().woCompare(migrateInfo->chunk.max)) {
                // Likely a split happened somewhere, so force reload the chunk manager
                cm = cfg->getChunkManager(txn, migrateInfo->ns, true);
                invariant(cm);

                c = cm->findIntersectingChunk(txn, migrateInfo->chunk.min);

                if (c->getMin().woCompare(migrateInfo->chunk.min) ||
                    c->getMax().woCompare(migrateInfo->chunk.max)) {
                    log() << "chunk mismatch after reload, ignoring will retry issue "
                          << migrateInfo->chunk.toString();

                    continue;
                }
            }

            BSONObj res;
            if (c->moveAndCommit(txn,
                                 migrateInfo->to,
                                 Chunk::MaxChunkSize,
                                 writeConcern,
                                 waitForDelete,
                                 0, /* maxTimeMS */
                                 res)) {
                movedCount++;
                continue;
            }

            // The move requires acquiring the collection metadata's lock, which can fail.
            log() << "balancer move failed: " << res << " from: " << migrateInfo->from
                  << " to: " << migrateInfo->to << " chunk: " << migrateInfo->chunk;

            if (res["chunkTooBig"].trueValue()) {
                // Reload just to be safe
                cm = cfg->getChunkManager(txn, migrateInfo->ns);
                invariant(cm);

                c = cm->findIntersectingChunk(txn, migrateInfo->chunk.min);

                log() << "performing a split because migrate failed for size reasons";

                Status status = c->split(txn, Chunk::normal, NULL, NULL);
                log() << "split results: " << status;

                if (!status.isOK()) {
                    log() << "marking chunk as jumbo: " << c->toString();

                    c->markAsJumbo(txn);

                    // We increment moveCount so we do another round right away
                    movedCount++;
                }
            }
        } catch (const DBException& ex) {
            warning() << "could not move chunk " << migrateInfo->chunk.toString()
                      << ", continuing balancing round" << causedBy(ex);
        }
    }

    return movedCount;
}

void Balancer::_ping(OperationContext* txn, bool waiting) {
    MongosType mType;
    mType.setName(_myid);
    mType.setPing(jsTime());
    mType.setUptime(static_cast<int>(time(0) - _started));
    mType.setWaiting(waiting);
    mType.setMongoVersion(versionString);

    grid.catalogManager(txn)->updateConfigDocument(txn,
                                                   MongosType::ConfigNS,
                                                   BSON(MongosType::name(_myid)),
                                                   BSON("$set" << mType.toBSON()),
                                                   true);
}

bool Balancer::_checkOIDs(OperationContext* txn) {
    vector<ShardId> all;
    grid.shardRegistry()->getAllShardIds(&all);

    // map of OID machine ID => shardId
    map<int, string> oids;

    for (const ShardId& shardId : all) {
        const auto s = grid.shardRegistry()->getShard(txn, shardId);
        if (!s) {
            continue;
        }

        BSONObj f = uassertStatusOK(grid.shardRegistry()->runIdempotentCommandOnShard(
            txn,
            s,
            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
            "admin",
            BSON("features" << 1)));
        if (f["oidMachine"].isNumber()) {
            int x = f["oidMachine"].numberInt();
            if (oids.count(x) == 0) {
                oids[x] = shardId;
            } else {
                log() << "error: 2 machines have " << x << " as oid machine piece: " << shardId
                      << " and " << oids[x];

                uassertStatusOK(grid.shardRegistry()->runIdempotentCommandOnShard(
                    txn,
                    s,
                    ReadPreferenceSetting{ReadPreference::PrimaryOnly},
                    "admin",
                    BSON("features" << 1 << "oidReset" << 1)));

                const auto otherShard = grid.shardRegistry()->getShard(txn, oids[x]);
                if (otherShard) {
                    uassertStatusOK(grid.shardRegistry()->runIdempotentCommandOnShard(
                        txn,
                        otherShard,
                        ReadPreferenceSetting{ReadPreference::PrimaryOnly},
                        "admin",
                        BSON("features" << 1 << "oidReset" << 1)));
                }

                return false;
            }
        } else {
            log() << "warning: oidMachine not set on: " << s->toString();
        }
    }

    return true;
}

/**
 * Occasionally prints a log message with shard versions if the versions are not the same
 * in the cluster.
 */
void warnOnMultiVersion(const ShardInfoMap& shardInfo) {
    bool isMultiVersion = false;
    for (ShardInfoMap::const_iterator i = shardInfo.begin(); i != shardInfo.end(); ++i) {
        if (!isSameMajorVersion(i->second.getMongoVersion().c_str())) {
            isMultiVersion = true;
            break;
        }
    }

    // If we're all the same version, don't message
    if (!isMultiVersion)
        return;

    warning() << "multiVersion cluster detected, my version is " << versionString;
    for (ShardInfoMap::const_iterator i = shardInfo.begin(); i != shardInfo.end(); ++i) {
        log() << i->first << " is at version " << i->second.getMongoVersion();
    }
}

void Balancer::_doBalanceRound(OperationContext* txn,
                               ForwardingCatalogManager::ScopedDistLock* distLock,
                               vector<shared_ptr<MigrateInfo>>* candidateChunks) {
    invariant(candidateChunks);

    vector<CollectionType> collections;
    Status collsStatus =
        grid.catalogManager(txn)->getCollections(txn, nullptr, &collections, nullptr);
    if (!collsStatus.isOK()) {
        warning() << "Failed to retrieve the set of collections during balancing round "
                  << collsStatus;
        return;
    }

    if (collections.empty()) {
        LOG(1) << "no collections to balance";
        return;
    }

    // Get a list of all the shards that are participating in this balance round along with any
    // maximum allowed quotas and current utilization. We get the latter by issuing
    // db.serverStatus() (mem.mapped) to all shards.
    //
    // TODO: skip unresponsive shards and mark information as stale.
    ShardInfoMap shardInfo;
    Status loadStatus = DistributionStatus::populateShardInfoMap(txn, &shardInfo);
    if (!loadStatus.isOK()) {
        warning() << "failed to load shard metadata" << causedBy(loadStatus);
        return;
    }

    if (shardInfo.size() < 2) {
        LOG(1) << "can't balance without more active shards";
        return;
    }

    OCCASIONALLY warnOnMultiVersion(shardInfo);

    // For each collection, check if the balancing policy recommends moving anything around.
    for (const auto& coll : collections) {
        uassertStatusOK(distLock->checkForPendingCatalogChange());

        // Skip collections for which balancing is disabled
        const NamespaceString& nss = coll.getNs();

        if (!coll.getAllowBalance()) {
            LOG(1) << "Not balancing collection " << nss << "; explicitly disabled.";
            continue;
        }

        std::vector<ChunkType> allNsChunks;
        Status status = grid.catalogManager(txn)->getChunks(txn,
                                                            BSON(ChunkType::ns(nss.ns())),
                                                            BSON(ChunkType::min() << 1),
                                                            boost::none,  // all chunks
                                                            &allNsChunks,
                                                            nullptr);
        if (!status.isOK()) {
            warning() << "failed to load chunks for ns " << nss.ns() << causedBy(status);
            continue;
        }

        set<BSONObj> allChunkMinimums;
        map<string, vector<ChunkType>> shardToChunksMap;

        for (const ChunkType& chunk : allNsChunks) {
            allChunkMinimums.insert(chunk.getMin().getOwned());

            vector<ChunkType>& chunksList = shardToChunksMap[chunk.getShard()];
            chunksList.push_back(chunk);
        }

        if (shardToChunksMap.empty()) {
            LOG(1) << "skipping empty collection (" << nss.ns() << ")";
            continue;
        }

        for (ShardInfoMap::const_iterator i = shardInfo.begin(); i != shardInfo.end(); ++i) {
            // This loop just makes sure there is an entry in shardToChunksMap for every shard
            shardToChunksMap[i->first];
        }

        DistributionStatus distStatus(shardInfo, shardToChunksMap);

        // TODO: TagRange contains all the information from TagsType except for the namespace,
        //       so maybe the two can be merged at some point in order to avoid the
        //       transformation below.
        vector<TagRange> ranges;

        {
            vector<TagsType> collectionTags;
            uassertStatusOK(
                grid.catalogManager(txn)->getTagsForCollection(txn, nss.ns(), &collectionTags));
            for (const auto& tt : collectionTags) {
                ranges.push_back(
                    TagRange(tt.getMinKey().getOwned(), tt.getMaxKey().getOwned(), tt.getTag()));
                uassert(16356,
                        str::stream() << "tag ranges not valid for: " << nss.ns(),
                        distStatus.addTagRange(ranges.back()));
            }
        }

        auto statusGetDb = grid.catalogCache()->getDatabase(txn, nss.db().toString());
        if (!statusGetDb.isOK()) {
            warning() << "could not load db config to balance collection [" << nss.ns()
                      << "]: " << statusGetDb.getStatus();
            continue;
        }

        shared_ptr<DBConfig> cfg = statusGetDb.getValue();

        // This line reloads the chunk manager once if this process doesn't know the collection
        // is sharded yet.
        shared_ptr<ChunkManager> cm = cfg->getChunkManagerIfExists(txn, nss.ns(), true);
        if (!cm) {
            warning() << "could not load chunks to balance " << nss.ns() << " collection";
            continue;
        }

        // Loop through tags to make sure no chunk spans tags. Split on tag min for all chunks.
        bool didAnySplits = false;

        for (const TagRange& range : ranges) {
            BSONObj min =
                cm->getShardKeyPattern().getKeyPattern().extendRangeBound(range.min, false);

            if (allChunkMinimums.count(min) > 0) {
                continue;
            }

            didAnySplits = true;

            log() << "nss: " << nss.ns() << " need to split on " << min
                  << " because there is a range there";

            ChunkPtr c = cm->findIntersectingChunk(txn, min);

            vector<BSONObj> splitPoints;
            splitPoints.push_back(min);

            Status status = c->multiSplit(txn, splitPoints, NULL);
            if (!status.isOK()) {
                error() << "split failed: " << status;
            } else {
                LOG(1) << "split worked";
            }

            break;
        }

        if (didAnySplits) {
            // State change, just wait till next round
            continue;
        }

        shared_ptr<MigrateInfo> migrateInfo(
            _policy->balance(nss.ns(), distStatus, _balancedLastTime));
        if (migrateInfo) {
            candidateChunks->push_back(migrateInfo);
        }
    }
}

bool Balancer::_init(OperationContext* txn) {
    try {
        log() << "about to contact config servers and shards";

        // contact the config server and refresh shard information
        // checks that each shard is indeed a different process (no hostname mixup)
        // these checks are redundant in that they're redone at every new round but we want to do
        // them initially here so to catch any problem soon
        grid.shardRegistry()->reload(txn);
        if (!_checkOIDs(txn)) {
            return false;
        }

        log() << "config servers and shards contacted successfully";

        StringBuilder buf;
        buf << getHostNameCached() << ":" << serverGlobalParams.port;
        _myid = buf.str();
        _started = time(0);

        log() << "balancer id: " << _myid << " started";

        return true;

    } catch (std::exception& e) {
        warning() << "could not initialize balancer, please check that all shards and config "
                     "servers are up: " << e.what();
        return false;
    }
}

void Balancer::run() {
    Client::initThread("Balancer");

    // This is the body of a BackgroundJob so if we throw here we're basically ending the balancer
    // thread prematurely.
    while (!inShutdown()) {
        auto txn = cc().makeOperationContext();
        if (!_init(txn.get())) {
            log() << "will retry to initialize balancer in one minute";
            sleepsecs(60);
            continue;
        }

        break;
    }

    Seconds balanceRoundInterval(kBalanceRoundDefaultInterval);

    while (!inShutdown()) {
        auto txn = cc().makeOperationContext();

        BalanceRoundDetails roundDetails;

        try {
            // ping has to be first so we keep things in the config server in sync
            _ping(txn.get());

            MONGO_FAIL_POINT_BLOCK(balancerRoundIntervalSetting, scopedBalancerRoundInterval) {
                const BSONObj& data = scopedBalancerRoundInterval.getData();
                balanceRoundInterval = Seconds(data["sleepSecs"].numberInt());
            }

            BSONObj balancerResult;

            // use fresh shard state
            grid.shardRegistry()->reload(txn.get());

            // refresh chunk size (even though another balancer might be active)
            Chunk::refreshChunkSize(txn.get());

            auto balSettingsResult = grid.catalogManager(txn.get())->getGlobalSettings(
                txn.get(), SettingsType::BalancerDocKey);
            const bool isBalSettingsAbsent =
                balSettingsResult.getStatus() == ErrorCodes::NoMatchingDocument;
            if (!balSettingsResult.isOK() && !isBalSettingsAbsent) {
                warning() << balSettingsResult.getStatus();
                return;
            }
            const SettingsType& balancerConfig =
                isBalSettingsAbsent ? SettingsType{} : balSettingsResult.getValue();

            // now make sure we should even be running
            if ((!isBalSettingsAbsent && !grid.shouldBalance(balancerConfig)) ||
                MONGO_FAIL_POINT(skipBalanceRound)) {
                LOG(1) << "skipping balancing round because balancing is disabled";

                // Ping again so scripts can determine if we're active without waiting
                _ping(txn.get(), true);

                sleepFor(balanceRoundInterval);
                continue;
            }

            uassert(13258, "oids broken after resetting!", _checkOIDs(txn.get()));

            {
                auto scopedDistLock = grid.forwardingCatalogManager()->distLock(
                    txn.get(), "balancer", "doing balance round");

                if (!scopedDistLock.isOK()) {
                    LOG(1) << "skipping balancing round" << causedBy(scopedDistLock.getStatus());

                    // Ping again so scripts can determine if we're active without waiting
                    _ping(txn.get(), true);

                    sleepFor(balanceRoundInterval);  // no need to wake up soon
                    continue;
                }

                if (!grid.catalogManager(txn.get())->isMetadataConsistentFromLastCheck(txn.get())) {
                    warning() << "Skipping balancing round because data inconsistency"
                              << " was detected amongst the config servers";
                    sleepFor(balanceRoundInterval);
                    continue;
                }

                const bool waitForDelete =
                    (balancerConfig.isWaitForDeleteSet() ? balancerConfig.getWaitForDelete()
                                                         : false);

                std::unique_ptr<WriteConcernOptions> writeConcern;
                if (balancerConfig.isKeySet()) {  // if balancer doc exists.
                    writeConcern = balancerConfig.getWriteConcern();
                }

                LOG(1) << "*** start balancing round. "
                       << "waitForDelete: " << waitForDelete << ", secondaryThrottle: "
                       << (writeConcern.get() ? writeConcern->toBSON().toString() : "default");

                vector<shared_ptr<MigrateInfo>> candidateChunks;
                _doBalanceRound(txn.get(), &scopedDistLock.getValue(), &candidateChunks);

                if (candidateChunks.size() == 0) {
                    LOG(1) << "no need to move any chunk";
                    _balancedLastTime = 0;
                } else {
                    _balancedLastTime =
                        _moveChunks(txn.get(), candidateChunks, writeConcern.get(), waitForDelete);

                    roundDetails.setSucceeded(static_cast<int>(candidateChunks.size()),
                                              _balancedLastTime);

                    grid.catalogManager(txn.get())
                        ->logAction(txn.get(), "balancer.round", "", roundDetails.toBSON());
                }

                LOG(1) << "*** End of balancing round";
            }

            // Ping again so scripts can determine if we're active without waiting
            _ping(txn.get(), true);

            sleepFor(_balancedLastTime ? kShortBalanceRoundInterval : balanceRoundInterval);
        } catch (const std::exception& e) {
            log() << "caught exception while doing balance: " << e.what();

            // Just to match the opening statement if in log level 1
            LOG(1) << "*** End of balancing round";

            // This round failed, tell the world!
            roundDetails.setFailed(e.what());

            grid.catalogManager(txn.get())
                ->logAction(txn.get(), "balancer.round", "", roundDetails.toBSON());

            // Sleep a fair amount before retrying because of the error
            sleepFor(balanceRoundInterval);
        }
    }
}

}  // namespace mongo


<|start_filename|>jstests/multiVersion/large_document_sync.js<|end_filename|>
/**
 * Tests syncing large objects from a 3.0 node to a 3.2 node. In particular, this test was designed
 * to stress the logic for upconverting an OP_REPLY that contains more than 16MB worth of documents.
 *
 * This test was designed to reproduce SERVER-26182.
 */
(function() {
    // Create a replica set with one "3.0" node and one "3.2" node.
    var replSetName = "testset";
    var nodes = [{binVersion: "3.0"}, {binVersion: "latest"}];

    var rst = ReplSetTest({name: replSetName, nodes: nodes, nodeOptions: {vv: ''}});
    rst.startSet();

    // Rig the election so that the 3.0 node becomes primary.
    var replSetConfig = rst.getReplSetConfig();
    replSetConfig.members[1].priority = 0;
    rst.initiate(replSetConfig);

    var primaryDB = rst.getPrimary().getDB("test");

    primaryDB.c.drop();

    var docCloseTo1MB = {
        x: new Array(900 * 1024).join("x")
    };
    assert.gte(Object.bsonsize(docCloseTo1MB), 0.5 * 1024 * 1024);
    assert.lt(Object.bsonsize(docCloseTo1MB), 1 * 1024 * 1024);

    var docCloseTo4MB = {
        x: new Array(3.5 * 1024 * 1024).join("x")
    };
    assert.gte(Object.bsonsize(docCloseTo4MB), 3 * 1024 * 1024);
    assert.lt(Object.bsonsize(docCloseTo4MB), 4 * 1024 * 1024);

    var docCloseTo16MB = {
        x: new Array(15.5 * 1024 * 1024).join("x")
    };
    assert.gte(Object.bsonsize(docCloseTo16MB), 15 * 1024 * 1024);
    assert.lt(Object.bsonsize(docCloseTo16MB), 16 * 1024 * 1024);

    assert.gt(Object.bsonsize(docCloseTo4MB) + Object.bsonsize(docCloseTo16MB), 16 * 1024 * 1024);

    assert.gt(Object.bsonsize(docCloseTo1MB) + Object.bsonsize(docCloseTo16MB), 16 * 1024 * 1024);

    rst.getPrimary().forceWriteMode("commands");
    // The first find has a threshold of 1MB, so put almost 1MB in, then a huge
    // document to make the total over 16MB.
    assert.writeOK(primaryDB.c.insert(docCloseTo1MB));
    assert.writeOK(primaryDB.c.insert(docCloseTo16MB));

    // The first getMore batch should contain both the ~4 MB and the ~16 MB
    // document, leading to more than 16 MB of user data in the batch.
    assert.writeOK(primaryDB.c.insert(docCloseTo4MB));
    assert.writeOK(primaryDB.c.insert(docCloseTo16MB));

    rst.awaitReplication();
    assert.commandWorked(rst.getSecondary().adminCommand({isMaster: 1}),
                         "expected secondary to survive syncing large documents");

    rst.stopSet();
})();


<|start_filename|>src/mongo/db/catalog/collection_options.h<|end_filename|>
/**
 *    Copyright (C) 2014 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 <http://www.gnu.org/licenses/>.
 *
 *    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.
 */

#pragma once

#include <string>

#include "mongo/base/status.h"
#include "mongo/db/jsobj.h"

namespace mongo {

struct CollectionOptions {
    CollectionOptions() {
        reset();
    }

    void reset();

    /**
     * Returns true if collection options validates successfully.
     */
    bool isValid() const;

    /**
     * Confirms that collection options can be converted to BSON and back without errors.
     */
    Status validate() const;

    /**
     * Parses the "options" subfield of the collection info object.
     */
    Status parse(const BSONObj& obj);

    BSONObj toBSON() const;

    /**
     * @param max in and out, will be adjusted
     * @return if the value is valid at all
     */
    static bool validMaxCappedDocs(long long* max);

    // ----

    bool capped;
    long long cappedSize;
    long long cappedMaxDocs;

    // following 2 are mutually exclusive, can only have one set
    long long initialNumExtents;
    std::vector<long long> initialExtentSizes;

    // behavior of _id index creation when collection created
    void setNoIdIndex() {
        autoIndexId = NO;
    }
    enum {
        DEFAULT,  // currently yes for most collections, NO for some system ones
        YES,      // create _id index
        NO        // do not create _id index
    } autoIndexId;

    // user flags
    enum UserFlags {
        Flag_UsePowerOf2Sizes = 1 << 0,
        Flag_NoPadding = 1 << 1,
    };
    int flags;  // a bitvector of UserFlags
    bool flagsSet;

    bool temp;

    // Storage engine collection options. Always owned or empty.
    BSONObj storageEngine;

    // Default options for indexes created on the collection. Always owned or empty.
    BSONObj indexOptionDefaults;

    // Always owned or empty.
    BSONObj validator;
    std::string validationAction;
    std::string validationLevel;

    // Collation information produced on a newer version, which supports the collation feature. We
    // do the work to parse this information to ensure that we properly recognize when collation
    // metadata is in the catalog, and downgrade must fail.
    BSONObj collation;
};
}


<|start_filename|>src/mongo/db/pipeline/document_source_lookup.cpp<|end_filename|>
/**
 *    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 <http://www.gnu.org/licenses/>.
 *
 *    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 "document_source.h"

#include "mongo/base/init.h"
#include "mongo/db/client.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/pipeline/document.h"
#include "mongo/db/pipeline/expression.h"
#include "mongo/db/pipeline/expression_context.h"
#include "mongo/db/pipeline/value.h"
#include "mongo/db/query/lite_parsed_query.h"
#include "mongo/db/server_parameters.h"
#include "mongo/stdx/memory.h"

namespace mongo {

using boost::intrusive_ptr;

MONGO_EXPORT_STARTUP_SERVER_PARAMETER(internalAggregationLookupBatchSize,
                                      int,
                                      LiteParsedQuery::kDefaultBatchSize);

DocumentSourceLookUp::DocumentSourceLookUp(NamespaceString fromNs,
                                           std::string as,
                                           std::string localField,
                                           std::string foreignField,
                                           const boost::intrusive_ptr<ExpressionContext>& pExpCtx)
    : DocumentSource(pExpCtx),
      _fromNs(std::move(fromNs)),
      _as(std::move(as)),
      _localField(std::move(localField)),
      _foreignField(foreignField),
      _foreignFieldFieldName(std::move(foreignField)) {}

DocumentSourceLookUp::~DocumentSourceLookUp() {
    DESTRUCTOR_GUARD(
        // A DBClientCursor will issue a killCursors command through its parent DBDirectClient when
        // it goes out of scope. To issue a killCursors command, a DBDirectClient needs a valid
        // OperationContext. So here we set the OperationContext on the DBDirectClient, then
        // aggressively destroy the DBClientCursor.
        // Note that we cannot rely on any sort of callback from above to provide a valid
        // OperationContext, since we might be destroyed from the destructor of a CursorManager,
        // which does not have an OperationContext. Thus, we unfortunately have to make a new one or
        // use the one on our thread's Client.
        if (_mongod && _cursor) {
            auto& client = cc();
            if (auto opCtx = client.getOperationContext()) {
                pExpCtx->opCtx = opCtx;
                _mongod->setOperationContext(opCtx);
                _cursor.reset();
            } else {
                auto newOpCtx = client.makeOperationContext();
                pExpCtx->opCtx = newOpCtx.get();
                _mongod->setOperationContext(newOpCtx.get());
                _cursor.reset();
            }
        });
}

REGISTER_DOCUMENT_SOURCE(lookup, DocumentSourceLookUp::createFromBson);

const char* DocumentSourceLookUp::getSourceName() const {
    return "$lookup";
}

std::unique_ptr<DBClientCursor> DocumentSourceLookUp::doQuery(const Document& docToLookUp) const {
    auto query = DocumentSourceLookUp::queryForInput(docToLookUp);

    // Defaults for everything except batch size.
    const int nToReturn = 0;
    const int nToSkip = 0;
    const BSONObj* fieldsToReturn = nullptr;
    const int queryOptions = 0;

    const int batchSize = internalAggregationLookupBatchSize;
    return _mongod->directClient()->query(_fromNs.ns(),
                                          std::move(query),
                                          nToReturn,
                                          nToSkip,
                                          fieldsToReturn,
                                          queryOptions,
                                          batchSize);
}

boost::optional<Document> DocumentSourceLookUp::getNext() {
    pExpCtx->checkForInterrupt();

    uassert(4567, "from collection cannot be sharded", !_mongod->isSharded(_fromNs));

    if (_handlingUnwind) {
        return unwindResult();
    }

    boost::optional<Document> input = pSource->getNext();
    if (!input)
        return {};
    auto cursor = doQuery(*input);

    std::vector<Value> results;
    int objsize = 0;
    while (cursor->more()) {
        BSONObj result = cursor->nextSafe();
        objsize += result.objsize();
        uassert(4568,
                str::stream() << "Total size of documents in " << _fromNs.coll() << " matching "
                              << DocumentSourceLookUp::queryForInput(*input)
                              << " exceeds maximum document size",
                objsize <= BSONObjMaxInternalSize);
        results.push_back(Value(result));
    }

    MutableDocument output(std::move(*input));
    output.setNestedField(_as, Value(std::move(results)));
    return output.freeze();
}

bool DocumentSourceLookUp::coalesce(const intrusive_ptr<DocumentSource>& pNextSource) {
    if (_handlingUnwind) {
        return false;
    }

    auto unwindSrc = dynamic_cast<DocumentSourceUnwind*>(pNextSource.get());
    if (!unwindSrc || unwindSrc->getUnwindPath() != _as.getPath(false)) {
        return false;
    }
    _unwindSrc = std::move(unwindSrc);
    _handlingUnwind = true;
    return true;
}

void DocumentSourceLookUp::dispose() {
    _cursor.reset();
    pSource->dispose();
}

BSONObj DocumentSourceLookUp::queryForInput(const Document& input) const {
    Value localFieldVal = input.getNestedField(_localField);
    if (localFieldVal.missing()) {
        localFieldVal = Value(BSONNULL);
    }

    // { _foreignFieldFiedlName : { "$eq" : localFieldValue } }
    BSONObjBuilder query;
    BSONObjBuilder subObj(query.subobjStart(_foreignFieldFieldName));
    subObj << "$eq" << localFieldVal;
    subObj.doneFast();
    return query.obj();
}

boost::optional<Document> DocumentSourceLookUp::unwindResult() {
    const boost::optional<FieldPath> indexPath(_unwindSrc->indexPath());

    // Loop until we get a document that has at least one match.
    // Note we may return early from this loop if our source stage is exhausted or if the unwind
    // source was asked to return empty arrays and we get a document without a match.
    while (!_cursor || !_cursor->more()) {
        _input = pSource->getNext();
        if (!_input)
            return {};

        _cursorIndex = 0;
        _cursor = doQuery(*_input);

        if (_unwindSrc->preserveNullAndEmptyArrays() && !_cursor->more()) {
            // There were no results for this cursor, but the $unwind was asked to preserve empty
            // arrays, so we should return a document without the array.
            MutableDocument output(std::move(*_input));
            // Note this will correctly objects in the prefix of '_as', to act as if we had created
            // an empty array and then removed it.
            output.setNestedField(_as, Value());
            if (indexPath) {
                output.setNestedField(*indexPath, Value(BSONNULL));
            }
            return output.freeze();
        }
    }
    invariant(_cursor->more() && bool(_input));
    auto nextVal = Value(_cursor->nextSafe());

    // Move input document into output if this is the last or only result, otherwise perform a copy.
    MutableDocument output(_cursor->more() ? *_input : std::move(*_input));
    output.setNestedField(_as, nextVal);

    if (indexPath) {
        output.setNestedField(*indexPath, Value(_cursorIndex));
    }

    _cursorIndex++;
    return output.freeze();
}

void DocumentSourceLookUp::serializeToArray(std::vector<Value>& array, bool explain) const {
    MutableDocument output(
        DOC(getSourceName() << DOC("from" << _fromNs.coll() << "as" << _as.getPath(false)
                                          << "localField" << _localField.getPath(false)
                                          << "foreignField" << _foreignField.getPath(false))));
    if (_handlingUnwind && explain) {
        const boost::optional<FieldPath> indexPath = _unwindSrc->indexPath();
        output[getSourceName()]["unwinding"] =
            Value(DOC("preserveNullAndEmptyArrays"
                      << _unwindSrc->preserveNullAndEmptyArrays() << "includeArrayIndex"
                      << (indexPath ? Value((*indexPath).getPath(false)) : Value())));
    }
    array.push_back(Value(output.freeze()));
    if (_handlingUnwind && !explain) {
        _unwindSrc->serializeToArray(array);
    }
}

DocumentSource::GetDepsReturn DocumentSourceLookUp::getDependencies(DepsTracker* deps) const {
    deps->fields.insert(_localField.getPath(false));
    return SEE_NEXT;
}

boost::intrusive_ptr<DocumentSource> DocumentSourceLookUp::createFromBson(
    BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx) {
    uassert(4569, "the $lookup specification must be an Object", elem.type() == Object);

    NamespaceString fromNs;
    std::string as;
    std::string localField;
    std::string foreignField;

    for (auto&& argument : elem.Obj()) {
        uassert(4570,
                str::stream() << "arguments to $lookup must be strings, " << argument << " is type "
                              << argument.type(),
                argument.type() == String);
        const auto argName = argument.fieldNameStringData();

        if (argName == "from") {
            fromNs = NamespaceString(pExpCtx->ns.db().toString() + '.' + argument.String());
        } else if (argName == "as") {
            as = argument.String();
        } else if (argName == "localField") {
            localField = argument.String();
        } else if (argName == "foreignField") {
            foreignField = argument.String();
        } else {
            uasserted(4571,
                      str::stream() << "unknown argument to $lookup: " << argument.fieldName());
        }
    }

    uassert(4572,
            "need to specify fields from, as, localField, and foreignField for a $lookup",
            !fromNs.ns().empty() && !as.empty() && !localField.empty() && !foreignField.empty());

    return new DocumentSourceLookUp(
        std::move(fromNs), std::move(as), std::move(localField), std::move(foreignField), pExpCtx);
}
}


<|start_filename|>jstests/sharding/bouncing_count.js<|end_filename|>
// Tests whether new sharding is detected on insert by mongos
(function() {

    var st = new ShardingTest({name: "test", shards: 10, mongos: 3});

    var mongosA = st.s0;
    var mongosB = st.s1;
    var mongosC = st.s2;

    var admin = mongosA.getDB("admin");
    var config = mongosA.getDB("config");

    var collA = mongosA.getCollection("foo.bar");
    var collB = mongosB.getCollection("" + collA);
    var collC = mongosB.getCollection("" + collA);

    admin.runCommand({enableSharding: "" + collA.getDB()});
    st.ensurePrimaryShard(collA.getDB().getName(), 'shard0001');
    admin.runCommand({shardCollection: "" + collA, key: {_id: 1}});

    var shards = config.shards.find().sort({_id: 1}).toArray();

    jsTestLog("Splitting up the collection...");

    // Split up the collection
    for (var i = 0; i < shards.length; i++) {
        printjson(admin.runCommand({split: "" + collA, middle: {_id: i}}));
        printjson(admin.runCommand({moveChunk: "" + collA, find: {_id: i}, to: shards[i]._id}));
    }

    mongosB.getDB("admin").runCommand({flushRouterConfig: 1});
    mongosC.getDB("admin").runCommand({flushRouterConfig: 1});
    printjson(collB.count());
    printjson(collC.count());

    // Change up all the versions...
    for (var i = 0; i < shards.length; i++) {
        printjson(admin.runCommand(
            {moveChunk: "" + collA, find: {_id: i}, to: shards[(i + 1) % shards.length]._id}));
    }

    // Make sure mongos A is up-to-date
    mongosA.getDB("admin").runCommand({flushRouterConfig: 1});

    jsTestLog("Running count!");

    printjson(collB.count());
    printjson(collC.find().toArray());

    st.stop();

})();


<|start_filename|>jstests/aggregation/bugs/lookup_unwind_killcursor.js<|end_filename|>
/**
 * Tests that the server can successfully kill the cursor of an aggregation pipeline which is
 * using a $lookup stage with its own cursor.
 *
 * This test was designed to reproduce SERVER-24386.
 */
(function() {
    'use strict';

    // Use a low batch size for the aggregation commands to ensure that the mongo shell does not
    // exhaust cursors during the first batch.
    const batchSize = 2;

    // Use a small batch in the $lookup stage to ensure it needs to issue a getMore. This will help
    // ensure that the $lookup stage has an open cursor when the first batch is returned.
    var conn =
        MongoRunner.runMongod({setParameter: "internalAggregationLookupBatchSize=" + batchSize});
    assert.neq(null, conn, 'mongod was unable to start up');
    const testDB = conn.getDB("test");

    function setup() {
        testDB.source.drop();
        testDB.dest.drop();

        assert.writeOK(testDB.source.insert({local: 1}));

        // The cursor batching logic actually requests one more document than it needs to fill the
        // first batch, so if we want to leave the $lookup stage paused with a cursor open we'll
        // need two more matching documents than the batch size.
        const numMatches = batchSize + 2;
        for (var i = 0; i < numMatches; ++i) {
            assert.writeOK(testDB.dest.insert({foreign: 1}));
        }
    }

    setup();

    const cmdObj = {
        aggregate: 'source',
        pipeline: [
            {
              $lookup: {
                  from: 'dest',
                  localField: 'local',
                  foreignField: 'foreign',
                  as: 'matches',
              }
            },
            {
              $unwind: {
                  path: '$matches',
              },
            },
        ],
        cursor: {
            batchSize: batchSize,
        },
    };

    var res = testDB.runCommand(cmdObj);
    assert.commandWorked(res);

    var cursor = new DBCommandCursor(conn, res, batchSize);
    cursor.close();  // Closing the cursor will issue a killCursor command.

    // Ensure the $lookup stage can be killed by dropping the collection.
    res = testDB.runCommand(cmdObj);
    assert.commandWorked(res);

    cursor = new DBCommandCursor(conn, res, batchSize);
    testDB.source.drop();

    assert.throws(function() {
        cursor.itcount();
    }, [], "expected cursor to have been destroyed during collection drop");

    // Ensure the $lookup stage can be killed by dropping the database.
    setup();
    res = testDB.runCommand(cmdObj);
    assert.commandWorked(res);

    cursor = new DBCommandCursor(conn, res, batchSize);
    assert.commandWorked(testDB.dropDatabase());

    assert.throws(function() {
        cursor.itcount();
    }, [], "expected cursor to have been destroyed during database drop");

    // Ensure the $lookup stage can be killed by the ClientCursorMonitor.
    setup();
    res = testDB.runCommand(cmdObj);
    assert.commandWorked(res);
    cursor = new DBCommandCursor(conn, res, batchSize);

    var serverStatus = assert.commandWorked(testDB.serverStatus());
    const expectedNumTimedOutCursors = serverStatus.metrics.cursor.timedOut + 1;

    // Wait until the idle cursor background job has killed the aggregation cursor.
    assert.commandWorked(testDB.adminCommand({setParameter: 1, cursorTimeoutMillis: 1000}));
    const cursorTimeoutFrequencySeconds = 4;
    assert.soon(
        function() {
            serverStatus = assert.commandWorked(testDB.serverStatus());
            // Use >= here since we may time out the $lookup stage's cursor as well.
            return serverStatus.metrics.cursor.timedOut >= expectedNumTimedOutCursors;
        },
        function() {
            return "aggregation cursor failed to time out: " + tojson(serverStatus);
        },
        cursorTimeoutFrequencySeconds * 1000 * 5);

    assert.eq(0, serverStatus.metrics.cursor.open.total, tojson(serverStatus));

    // We attempt to exhaust the aggregation cursor to verify that sending a getMore returns an
    // error due to the cursor being killed.
    var err = assert.throws(function() {
        cursor.itcount();
    });
    assert.eq(ErrorCodes.CursorNotFound, err.code, tojson(err));

    // Ensure the $lookup stage can be killed by shutting down the server.
    setup();
    res = testDB.runCommand(cmdObj);
    assert.commandWorked(res);

    assert.eq(0, MongoRunner.stopMongod(conn), "expected mongod to shutdown cleanly");
})();


<|start_filename|>jstests/multiVersion/initialize_from_old_node.js<|end_filename|>
/*
 * This is a regression test for SERVER-16189, to make sure a replica set with both current and
 * prior version nodes can be initialized from the prior version node.
 */

(function() {
    "use strict";
    var name = "initialize_from_old";
    var oldVersion = 'last-stable';
    var newVersion = 'latest';
    var nodes = {
        n0: {binVersion: oldVersion},
        n1: {binVersion: newVersion},
        n2: {binVersion: newVersion}
    };
    var rst = new ReplSetTest({nodes: nodes, name: name});
    var conns = rst.startSet();
    var oldNode = conns[0];
    var config = rst.getReplSetConfig();
    var response = oldNode.getDB("admin").runCommand({replSetInitiate: config});
    assert.eq(response.ok, 1);
    rst.awaitReplication();
})();


<|start_filename|>jstests/core/sorth.js<|end_filename|>
// Tests for the $in/sort/limit optimization combined with inequality bounds.  SERVER-5777

(function() {
    "use strict";

    var t = db.jstests_sorth;
    t.drop();

    // These can be set to modify the query run by the helper find().
    var _sort;
    var _limit;
    var _hint;

    /**
     * Generate a cursor using global parameters '_sort', '_hint', and '_limit'.
     */
    function find(query) {
        return t.find(query, {_id: 0}).sort(_sort).limit(_limit).hint(_hint);
    }

    /**
     * Returns true if the elements of 'expectedMatches' match element by element with
     * 'actualMatches', only considering the fields 'a' and 'b'.
     *
     * @param {Array} expectedMatches - expected results from a query.
     * @param {Array} actualMatches - the actual results from that query.
     */
    function resultsMatch(expectedMatches, actualMatches) {
        if (expectedMatches.length !== actualMatches.length) {
            return false;
        }

        for (var i = 0; i < expectedMatches.length; ++i) {
            if ((expectedMatches[i].a !== actualMatches[i].a) ||
                (expectedMatches[i].b !== actualMatches[i].b)) {
                return false;
            }
        }
        return true;
    }

    /**
     * Asserts that the given query returns results that are expected.
     *
     * @param {Object} options.query - the query to run.
     * @param {Array.<Object>} options.expectedQueryResults - the expected results from the query.
     * @param {Array.<Array>} [options.acceptableQueryResults=[options.expectedQueryResults]] - An
     * array of acceptable outcomes of the query. This can be used if there are multiple results
     * that are considered correct for the query.
     */
    function assertMatches(options) {
        const results = find(options.query).toArray();
        const acceptableQueryResults =
            options.acceptableQueryResults || [options.expectedQueryResults];
        assert.gte(acceptableQueryResults.length, 1);
        for (var i = 0; i < acceptableQueryResults.length; ++i) {
            const validResultSet = acceptableQueryResults[i];

            // All results should have the same number of results.
            assert.eq(validResultSet.length,
                      results.length,
                      "Expected " + results.length + " results from query " +
                          tojson(options.query) + " but found " + validResultSet.length);

            if (resultsMatch(validResultSet, results)) {
                return;
            }
        }
        throw new Error("Unexpected results for query " + tojson(options.query) + ": " +
                        tojson(results) + ", acceptable results were: " +
                        tojson(acceptableQueryResults));
    }

    /**
     * Reset data, index, and _sort and _hint globals.
     */
    function reset(sort, index) {
        t.drop();
        t.save({a: 1, b: 1});
        t.save({a: 1, b: 2});
        t.save({a: 1, b: 3});
        t.save({a: 2, b: 0});
        t.save({a: 2, b: 3});
        t.save({a: 2, b: 5});
        t.ensureIndex(index);
        _sort = sort;
        _hint = index;
    }

    function checkForwardDirection(options) {
        // All callers specify a sort that is prefixed by b, ascending.
        assert.eq(Object.keys(options.sort)[0], "b");
        assert.eq(options.sort.b, 1);

        // None of the callers specify a sort on "a".
        assert(!options.sort.hasOwnProperty("a"));

        reset(options.sort, options.index);

        _limit = -1;

        // Lower bound checks.
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 0}], query: {a: {$in: [1, 2]}, b: {$gte: 0}}});
        assertMatches(
            {expectedQueryResults: [{a: 1, b: 1}], query: {a: {$in: [1, 2]}, b: {$gt: 0}}});
        assertMatches(
            {expectedQueryResults: [{a: 1, b: 1}], query: {a: {$in: [1, 2]}, b: {$gte: 1}}});
        assertMatches(
            {expectedQueryResults: [{a: 1, b: 2}], query: {a: {$in: [1, 2]}, b: {$gt: 1}}});
        assertMatches(
            {expectedQueryResults: [{a: 1, b: 2}], query: {a: {$in: [1, 2]}, b: {$gte: 2}}});

        // Since we are sorting on the field "b", and the sort specification doesn't include the
        // field "a", any query that is expected to result in a document with a value of 3 for "b"
        // has two acceptable results, since there are two documents with a value of 3 for "b". The
        // same argument applies for all assertions below involving a result with a value of 3 for
        // the field "b".
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$gt: 2}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$gte: 3}}
        });
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$gt: 3}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$gte: 4}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$gt: 4}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$gte: 5}}});

        // Upper bound checks.
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 0}], query: {a: {$in: [1, 2]}, b: {$lte: 0}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 0}], query: {a: {$in: [1, 2]}, b: {$lt: 1}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 0}], query: {a: {$in: [1, 2]}, b: {$lte: 1}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 0}], query: {a: {$in: [1, 2]}, b: {$lt: 3}}});

        // Lower and upper bounds checks.
        assertMatches({
            expectedQueryResults: [{a: 2, b: 0}],
            query: {a: {$in: [1, 2]}, b: {$gte: 0, $lte: 0}}
        });
        assertMatches({
            expectedQueryResults: [{a: 2, b: 0}],
            query: {a: {$in: [1, 2]}, b: {$gte: 0, $lt: 1}}
        });
        assertMatches({
            expectedQueryResults: [{a: 2, b: 0}],
            query: {a: {$in: [1, 2]}, b: {$gte: 0, $lte: 1}}
        });
        assertMatches({
            expectedQueryResults: [{a: 1, b: 1}],
            query: {a: {$in: [1, 2]}, b: {$gt: 0, $lte: 1}}
        });
        assertMatches({
            expectedQueryResults: [{a: 1, b: 2}],
            query: {a: {$in: [1, 2]}, b: {$gte: 2, $lt: 3}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$gte: 2.5, $lte: 3}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$gt: 2.5, $lte: 3}}
        });

        // Limit is -2.
        _limit = -2;
        assertMatches({
            expectedQueryResults: [{a: 2, b: 0}, {a: 1, b: 1}],
            query: {a: {$in: [1, 2]}, b: {$gte: 0}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 2}, {a: 2, b: 3}], [{a: 1, b: 2}, {a: 1, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$gt: 1}}
        });
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$gt: 4}}});

        // With an additional document between the $in values.
        t.save({a: 1.5, b: 3});
        assertMatches({
            expectedQueryResults: [{a: 2, b: 0}, {a: 1, b: 1}],
            query: {a: {$in: [1, 2]}, b: {$gte: 0}}
        });
    }

    // Basic test with an index suffix order.
    checkForwardDirection({sort: {b: 1}, index: {a: 1, b: 1}});
    // With an additional index field.
    checkForwardDirection({sort: {b: 1}, index: {a: 1, b: 1, c: 1}});
    // With an additional reverse direction index field.
    checkForwardDirection({sort: {b: 1}, index: {a: 1, b: 1, c: -1}});
    // With an additional ordered index field.
    checkForwardDirection({sort: {b: 1, c: 1}, index: {a: 1, b: 1, c: 1}});
    // With an additional reverse direction ordered index field.
    checkForwardDirection({sort: {b: 1, c: -1}, index: {a: 1, b: 1, c: -1}});

    function checkReverseDirection(options) {
        // All callers specify a sort that is prefixed by "b", descending.
        assert.eq(Object.keys(options.sort)[0], "b");
        assert.eq(options.sort.b, -1);
        // None of the callers specify a sort on "a".
        assert(!options.sort.hasOwnProperty("a"));

        reset(options.sort, options.index);
        _limit = -1;

        // For matching documents, highest value of 'b' is 5.
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$gte: 0}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$gte: 5}}});
        assertMatches(
            {expectedQueryResults: [{a: 2, b: 5}], query: {a: {$in: [1, 2]}, b: {$lte: 5}}});
        assertMatches({
            expectedQueryResults: [{a: 2, b: 5}],
            query: {a: {$in: [1, 2]}, b: {$lte: 5, $gte: 5}}
        });

        // For matching documents, highest value of 'b' is 2.
        assertMatches(
            {expectedQueryResults: [{a: 1, b: 2}], query: {a: {$in: [1, 2]}, b: {$lt: 3}}});
        assertMatches({
            expectedQueryResults: [{a: 1, b: 2}],
            query: {a: {$in: [1, 2]}, b: {$lt: 3, $gt: 1}}
        });

        // For matching documents, highest value of 'b' is 1.
        assertMatches({
            expectedQueryResults: [{a: 1, b: 1}],
            query: {a: {$in: [1, 2]}, b: {$lt: 2, $gte: 1}}
        });

        // These queries expect 3 as the highest value of 'b' among matching documents, but there
        // are two documents with a value of 3 for the field 'b'. Either document is acceptable,
        // since there is no sort order on any other existing fields.
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$lt: 5}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$lt: 3.1}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$lt: 3.5}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$lte: 3}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$lt: 3.5, $gte: 3}}
        });
        assertMatches({
            acceptableQueryResults: [[{a: 1, b: 3}], [{a: 2, b: 3}]],
            query: {a: {$in: [1, 2]}, b: {$lte: 3, $gt: 0}}
        });
    }

    // With a descending order index.
    checkReverseDirection({sort: {b: -1}, index: {a: 1, b: -1}});
    checkReverseDirection({sort: {b: -1}, index: {a: 1, b: -1, c: 1}});
    checkReverseDirection({sort: {b: -1}, index: {a: 1, b: -1, c: -1}});
    checkReverseDirection({sort: {b: -1, c: 1}, index: {a: 1, b: -1, c: 1}});
    checkReverseDirection({sort: {b: -1, c: -1}, index: {a: 1, b: -1, c: -1}});
}());


<|start_filename|>jstests/noPassthrough/wt_malformed_creation_string.js<|end_filename|>
/**
 * Tests that a null embedded malformed string is rejected gracefully.
 */
(function() {
    'use strict';

    var engine = 'wiredTiger';
    if (jsTest.options().storageEngine) {
        engine = jsTest.options().storageEngine;
    }

    // Skip this test if not running with the right storage engine.
    if (engine !== 'wiredTiger' && engine !== 'inMemory') {
        jsTest.log('Skipping test because storageEngine is not "wiredTiger" or "inMemory"');
        return;
    }

    // Build an array of malformed strings to test
    var malformedStrings = ["\u0000000", "\0,", "bl\0ah", "split_pct=30,\0split_pct=35,"];

    // Start up a mongod.
    // Test that collection and index creation with malformed creation strings fail gracefully.
    runTest();

    function runTest() {
        var dbpath = MongoRunner.dataPath + 'wt_malformed_creation_string';
        resetDbpath(dbpath);

        // Start a mongod
        var conn = MongoRunner.runMongod({
            dbpath: dbpath,
            noCleanData: true,
        });
        assert.neq(null, conn, 'mongod was unable to start up');

        var testDB = conn.getDB('test');

        // Collection creation with malformed string should fail
        for (var i = 0; i < malformedStrings.length; i++) {
            assert.commandFailedWithCode(
                testDB.createCollection(
                    'coll', {storageEngine: {[engine]: {configString: malformedStrings[i]}}}),
                ErrorCodes.FailedToParse);
        }

        // Create collection to test index creation on
        assert.commandWorked(testDB.createCollection('coll'));

        // Index creation with malformed string should fail
        for (var i = 0; i < malformedStrings.length; i++) {
            assert.commandFailedWithCode(
                testDB.coll.createIndex(
                    {a: 1},
                    {
                      name: 'with_malformed_str',
                      storageEngine: {[engine]: {configString: malformedStrings[i]}}
                    }),
                ErrorCodes.FailedToParse);
        }

        MongoRunner.stopMongod(conn);
    }
})();


<|start_filename|>jstests/tool/tool1.js<|end_filename|>
// mongo tool tests, very basic to start with

baseName = "jstests_tool_tool1";
dbPath = MongoRunner.dataPath + baseName + "/";
externalPath = MongoRunner.dataPath + baseName + "_external/";
externalBaseName = "export.json";
externalFile = externalPath + externalBaseName;

function fileSize() {
    var l = listFiles(externalPath);
    for (var i = 0; i < l.length; i++) {
        if (l[i].baseName == externalBaseName)
            return l[i].size;
    }
    return -1;
}

resetDbpath(externalPath);

var m = MongoRunner.runMongod({dbpath: dbPath, noprealloc: "", bind_ip: "127.0.0.1"});
c = m.getDB(baseName).getCollection(baseName);
c.save({a: 1});
assert(c.findOne());

var exitCode = MongoRunner.runMongoTool("mongodump",
                                        {
                                          host: "127.0.0.1:" + m.port,
                                          out: externalPath,
                                        });
assert.eq(0, exitCode, "mongodump failed to dump data from mongod");

c.drop();

exitCode = MongoRunner.runMongoTool("mongorestore",
                                    {
                                      host: "127.0.0.1:" + m.port,
                                      dir: externalPath,
                                    });
assert.eq(0, exitCode, "mongorestore failed to restore data to mongod");

assert.soon("c.findOne()", "mongodump then restore has no data w/sleep");
assert(c.findOne(), "mongodump then restore has no data");
assert.eq(1, c.findOne().a, "mongodump then restore has no broken data");

resetDbpath(externalPath);

assert.eq(-1, fileSize(), "mongoexport prep invalid");

exitCode = MongoRunner.runMongoTool("mongoexport",
                                    {
                                      host: "127.0.0.1:" + m.port,
                                      db: baseName,
                                      collection: baseName,
                                      out: externalFile,
                                    });
assert.eq(0,
          exitCode,
          "mongoexport failed to export collection '" + c.getFullName() + "' from mongod");

assert.lt(10, fileSize(), "file size changed");

c.drop();

exitCode = MongoRunner.runMongoTool("mongoimport",
                                    {
                                      host: "127.0.0.1:" + m.port,
                                      db: baseName,
                                      collection: baseName,
                                      file: externalFile,
                                    });
assert.eq(0,
          exitCode,
          "mongoimport failed to import collection '" + c.getFullName() + "' into mongod");

assert.soon("c.findOne()", "mongo import json A");
assert(c.findOne() && 1 == c.findOne().a, "mongo import json B");
