path: root/src/mongo/s/d_migrate.cpp

/**
 *    Copyright (C) 2008-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::kSharding

#include "mongo/platform/basic.h"

#include <algorithm>
#include <chrono>
#include <condition_variable>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include "mongo/client/connpool.h"
#include "mongo/db/auth/action_set.h"
#include "mongo/db/auth/action_type.h"
#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/auth/privilege.h"
#include "mongo/db/catalog/document_validation.h"
#include "mongo/db/catalog/index_create.h"
#include "mongo/db/clientcursor.h"
#include "mongo/db/commands.h"
#include "mongo/db/concurrency/lock_state.h"
#include "mongo/db/db_raii.h"
#include "mongo/db/dbhelpers.h"
#include "mongo/db/exec/plan_stage.h"
#include "mongo/db/field_parser.h"
#include "mongo/db/service_context.h"
#include "mongo/db/hasher.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/namespace_string.h"
#include "mongo/db/op_observer.h"
#include "mongo/db/operation_context_impl.h"
#include "mongo/db/ops/delete.h"
#include "mongo/db/query/internal_plans.h"
#include "mongo/db/query/query_knobs.h"
#include "mongo/db/range_deleter_service.h"
#include "mongo/db/repl/repl_client_info.h"
#include "mongo/db/repl/replication_coordinator_global.h"
#include "mongo/db/storage/mmap_v1/dur.h"
#include "mongo/db/write_concern.h"
#include "mongo/logger/ramlog.h"
#include "mongo/s/catalog/catalog_manager.h"
#include "mongo/s/catalog/type_chunk.h"
#include "mongo/s/chunk.h"
#include "mongo/s/chunk_version.h"
#include "mongo/s/config.h"
#include "mongo/s/d_state.h"
#include "mongo/s/catalog/dist_lock_manager.h"
#include "mongo/s/client/shard.h"
#include "mongo/s/client/shard_registry.h"
#include "mongo/s/grid.h"
#include "mongo/s/shard_key_pattern.h"
#include "mongo/stdx/memory.h"
#include "mongo/util/assert_util.h"
#include "mongo/util/elapsed_tracker.h"
#include "mongo/util/exit.h"
#include "mongo/util/fail_point_service.h"
#include "mongo/util/log.h"
#include "mongo/util/processinfo.h"
#include "mongo/util/startup_test.h"

// Pause while a fail point is enabled.
#define MONGO_FP_PAUSE_WHILE(symbol)   \
    while (MONGO_FAIL_POINT(symbol)) { \
        sleepmillis(100);              \
    }

namespace mongo {

using namespace std::chrono;
using std::list;
using std::set;
using std::string;
using std::unique_ptr;
using std::vector;

namespace {

const int kDefaultWTimeoutMs = 60 * 1000;
const WriteConcernOptions DefaultWriteConcern(2, WriteConcernOptions::NONE, kDefaultWTimeoutMs);

/**
 * Returns the default write concern for migration cleanup (at donor shard) and
 * cloning documents (at recipient shard).
 */
WriteConcernOptions getDefaultWriteConcern() {
    repl::ReplicationCoordinator* replCoordinator = repl::getGlobalReplicationCoordinator();
    if (replCoordinator->getReplicationMode() == mongo::repl::ReplicationCoordinator::modeReplSet) {
        Status status = replCoordinator->checkIfWriteConcernCanBeSatisfied(DefaultWriteConcern);
        if (status.isOK()) {
            return DefaultWriteConcern;
        }
    }

    return WriteConcernOptions(1, WriteConcernOptions::NONE, 0);
}

}  // namespace

MONGO_FP_DECLARE(failMigrationCommit);
MONGO_FP_DECLARE(failMigrationConfigWritePrepare);
MONGO_FP_DECLARE(failMigrationApplyOps);

Tee* migrateLog = RamLog::get("migrate");

class MoveTimingHelper {
public:
    MoveTimingHelper(OperationContext* txn,
                     const string& where,
                     const string& ns,
                     BSONObj min,
                     BSONObj max,
                     int total,
                     string* cmdErrmsg,
                     string toShard,
                     string fromShard)
        : _txn(txn),
          _where(where),
          _ns(ns),
          _to(toShard),
          _from(fromShard),
          _next(0),
          _total(total),
          _cmdErrmsg(cmdErrmsg) {
        _b.append("min", min);
        _b.append("max", max);
    }

    ~MoveTimingHelper() {
        // even if logChange doesn't throw, bson does
        // sigh
        try {
            if (!_to.empty()) {
                _b.append("to", _to);
            }
            if (!_from.empty()) {
                _b.append("from", _from);
            }
            if (_next != _total) {
                _b.append("note", "aborted");
            } else {
                _b.append("note", "success");
            }
            if (!_cmdErrmsg->empty()) {
                _b.append("errmsg", *_cmdErrmsg);
            }

            grid.catalogManager()->logChange(_txn->getClient()->clientAddress(true),
                                             (string) "moveChunk." + _where,
                                             _ns,
                                             _b.obj());
        } catch (const std::exception& e) {
            warning() << "couldn't record timing for moveChunk '" << _where << "': " << e.what()
                      << migrateLog;
        }
    }

    void done(int step) {
        invariant(step == ++_next);
        invariant(step <= _total);

        const string s = str::stream() << "step " << step << " of " << _total;

        CurOp* op = CurOp::get(_txn);
        {
            std::lock_guard<Client> lk(*_txn->getClient());
            op->setMessage_inlock(s.c_str());
        }

        _b.appendNumber(s, _t.millis());
        _t.reset();
    }

private:
    OperationContext* const _txn;
    Timer _t;

    string _where;
    string _ns;
    string _to;
    string _from;

    int _next;
    int _total;  // expected # of steps

    const string* _cmdErrmsg;

    BSONObjBuilder _b;
};

class ChunkCommandHelper : public Command {
public:
    ChunkCommandHelper(const char* name) : Command(name) {}

    virtual void help(std::stringstream& help) const {
        help << "internal - should not be called directly";
    }
    virtual bool slaveOk() const {
        return false;
    }
    virtual bool adminOnly() const {
        return true;
    }
    virtual bool isWriteCommandForConfigServer() const {
        return false;
    }
};

bool isInRange(const BSONObj& obj,
               const BSONObj& min,
               const BSONObj& max,
               const BSONObj& shardKeyPattern) {
    ShardKeyPattern shardKey(shardKeyPattern);
    BSONObj k = shardKey.extractShardKeyFromDoc(obj);
    return k.woCompare(min) >= 0 && k.woCompare(max) < 0;
}

class MigrateFromStatus {
public:
    MigrateFromStatus() : _inCriticalSection(false), _memoryUsed(0), _active(false) {}

    /**
     * @return false if cannot start. One of the reason for not being able to
     *     start is there is already an existing migration in progress.
     */
    bool start(OperationContext* txn,
               const std::string& ns,
               const BSONObj& min,
               const BSONObj& max,
               const BSONObj& shardKeyPattern) {
        verify(!min.isEmpty());
        verify(!max.isEmpty());
        verify(!ns.empty());

        // Get global shared to synchronize with logOp. Also see comments in the class
        // members declaration for more details.
        Lock::GlobalRead globalShared(txn->lockState());
        std::lock_guard<std::mutex> lk(_mutex);

        if (_active) {
            return false;
        }

        _ns = ns;
        _min = min;
        _max = max;
        _shardKeyPattern = shardKeyPattern;

        verify(_deleted.size() == 0);
        verify(_reload.size() == 0);
        verify(_memoryUsed == 0);

        _active = true;

        std::lock_guard<std::mutex> tLock(_cloneLocsMutex);
        verify(_cloneLocs.size() == 0);

        return true;
    }

    void done(OperationContext* txn) {
        log() << "MigrateFromStatus::done About to acquire global lock to exit critical "
                 "section";

        // Get global shared to synchronize with logOp. Also see comments in the class
        // members declaration for more details.
        Lock::GlobalRead globalShared(txn->lockState());
        std::lock_guard<std::mutex> lk(_mutex);

        _active = false;
        _deleteNotifyExec.reset(NULL);
        _inCriticalSection = false;
        _inCriticalSectionCV.notify_all();

        _deleted.clear();
        _reload.clear();
        _memoryUsed = 0;

        std::lock_guard<std::mutex> cloneLock(_cloneLocsMutex);
        _cloneLocs.clear();
    }

    void logOp(OperationContext* txn,
               const char* opstr,
               const char* ns,
               const BSONObj& obj,
               BSONObj* patt,
               bool notInActiveChunk) {
        ensureShardVersionOKOrThrow(txn->getClient(), ns);

        const char op = opstr[0];

        if (notInActiveChunk) {
            // Ignore writes that came from the migration process like cleanup so they
            // won't be transferred to the recipient shard. Also ignore ops from
            // _migrateClone and _transferMods since it is impossible to move a chunk
            // to self.
            return;
        }

        dassert(txn->lockState()->isWriteLocked());  // Must have Global IX.

        if (!_active)
            return;

        if (_ns != ns)
            return;

        // no need to log if this is not an insertion, an update, or an actual deletion
        // note: opstr 'db' isn't a deletion but a mention that a database exists
        // (for replication machinery mostly).
        if (op == 'n' || op == 'c' || (op == 'd' && opstr[1] == 'b'))
            return;

        BSONElement ide;
        if (patt)
            ide = patt->getField("_id");
        else
            ide = obj["_id"];

        if (ide.eoo()) {
            warning() << "logOpForSharding got mod with no _id, ignoring  obj: " << obj
                      << migrateLog;
            return;
        }

        if (op == 'i' && (!isInRange(obj, _min, _max, _shardKeyPattern))) {
            return;
        }

        BSONObj idObj(ide.wrap());

        if (op == 'u') {
            BSONObj fullDoc;
            OldClientContext ctx(txn, _ns, false);
            if (!Helpers::findById(txn, ctx.db(), _ns.c_str(), idObj, fullDoc)) {
                warning() << "logOpForSharding couldn't find: " << idObj
                          << " even though should have" << migrateLog;
                dassert(false);  // TODO: Abort the migration.
                return;
            }

            if (!isInRange(fullDoc, _min, _max, _shardKeyPattern)) {
                return;
            }
        }

        // Note: can't check if delete is in active chunk since the document is gone!

        txn->recoveryUnit()->registerChange(new LogOpForShardingHandler(this, idObj, op));
    }

    /**
     * Insert items from docIdList to a new array with the given fieldName in the given
     * builder. If explode is true, the inserted object will be the full version of the
     * document. Note that the whenever an item from the docList is inserted to the array,
     * it will also be removed from docList.
     *
     * Should be holding the collection lock for ns if explode is true.
     */
    void xfer(OperationContext* txn,
              const string& ns,
              Database* db,
              list<BSONObj>* docIdList,
              BSONObjBuilder& builder,
              const char* fieldName,
              long long& size,
              bool explode) {
        const long long maxSize = 1024 * 1024;

        if (docIdList->size() == 0 || size > maxSize)
            return;

        BSONArrayBuilder arr(builder.subarrayStart(fieldName));

        list<BSONObj>::iterator docIdIter = docIdList->begin();
        while (docIdIter != docIdList->end() && size < maxSize) {
            BSONObj idDoc = *docIdIter;
            if (explode) {
                BSONObj fullDoc;
                if (Helpers::findById(txn, db, ns.c_str(), idDoc, fullDoc)) {
                    arr.append(fullDoc);
                    size += fullDoc.objsize();
                }
            } else {
                arr.append(idDoc);
                size += idDoc.objsize();
            }

            docIdIter = docIdList->erase(docIdIter);
        }

        arr.done();
    }

    /**
     * called from the dest of a migrate
     * transfers mods from src to dest
     */
    bool transferMods(OperationContext* txn, string& errmsg, BSONObjBuilder& b) {
        long long size = 0;

        {
            AutoGetCollectionForRead ctx(txn, getNS());

            std::lock_guard<std::mutex> sl(_mutex);
            if (!_active) {
                errmsg = "no active migration!";
                return false;
            }

            // TODO: fix SERVER-16540 race
            xfer(txn, _ns, ctx.getDb(), &_deleted, b, "deleted", size, false);
            xfer(txn, _ns, ctx.getDb(), &_reload, b, "reload", size, true);
        }

        b.append("size", size);

        return true;
    }

    /**
     * Get the disklocs that belong to the chunk migrated and sort them in _cloneLocs
     * (to avoid seeking disk later).
     *
     * @param maxChunkSize number of bytes beyond which a chunk's base data (no indices)
     *                     is considered too large to move.
     * @param errmsg filled with textual description of error if this call return false.
     * @return false if approximate chunk size is too big to move or true otherwise.
     */
    bool storeCurrentLocs(OperationContext* txn,
                          long long maxChunkSize,
                          string& errmsg,
                          BSONObjBuilder& result) {
        AutoGetCollectionForRead ctx(txn, getNS());
        Collection* collection = ctx.getCollection();
        if (!collection) {
            errmsg = "ns not found, should be impossible";
            return false;
        }

        // Allow multiKey based on the invariant that shard keys must be single-valued.
        // Therefore, any multi-key index prefixed by shard key cannot be multikey over
        // the shard key fields.
        IndexDescriptor* idx =
            collection->getIndexCatalog()->findShardKeyPrefixedIndex(txn,
                                                                     _shardKeyPattern,
                                                                     false);  // requireSingleKey

        if (idx == NULL) {
            errmsg = str::stream() << "can't find index with prefix " << _shardKeyPattern
                                   << " in storeCurrentLocs for " << _ns;
            return false;
        }

        // Assume both min and max non-empty, append MinKey's to make them fit chosen index
        BSONObj min;
        BSONObj max;
        KeyPattern kp(idx->keyPattern());

        {
            // It's alright not to lock _mutex all the way through based on the assumption
            // that this is only called by the main thread that drives the migration and
            // only it can start and stop the current migration.
            std::lock_guard<std::mutex> sl(_mutex);

            invariant(_deleteNotifyExec.get() == NULL);
            unique_ptr<WorkingSet> ws = stdx::make_unique<WorkingSet>();
            unique_ptr<DeleteNotificationStage> dns = stdx::make_unique<DeleteNotificationStage>();
            // Takes ownership of 'ws' and 'dns'.
            auto statusWithPlanExecutor = PlanExecutor::make(
                txn, std::move(ws), std::move(dns), collection, PlanExecutor::YIELD_MANUAL);
            invariant(statusWithPlanExecutor.isOK());
            _deleteNotifyExec = std::move(statusWithPlanExecutor.getValue());
            _deleteNotifyExec->registerExec();

            min = Helpers::toKeyFormat(kp.extendRangeBound(_min, false));
            max = Helpers::toKeyFormat(kp.extendRangeBound(_max, false));
        }

        unique_ptr<PlanExecutor> exec(
            InternalPlanner::indexScan(txn, collection, idx, min, max, false));
        // We can afford to yield here because any change to the base data that we might
        // miss is already being queued and will migrate in the 'transferMods' stage.
        exec->setYieldPolicy(PlanExecutor::YIELD_AUTO);

        // use the average object size to estimate how many objects a full chunk would carry
        // do that while traversing the chunk's range using the sharding index, below
        // there's a fair amount of slack before we determine a chunk is too large because object
        // sizes will vary
        unsigned long long maxRecsWhenFull;
        long long avgRecSize;
        const long long totalRecs = collection->numRecords(txn);
        if (totalRecs > 0) {
            avgRecSize = collection->dataSize(txn) / totalRecs;
            maxRecsWhenFull = maxChunkSize / avgRecSize;
            maxRecsWhenFull = std::min((unsigned long long)(Chunk::MaxObjectPerChunk + 1),
                                       130 * maxRecsWhenFull / 100 /* slack */);
        } else {
            avgRecSize = 0;
            maxRecsWhenFull = Chunk::MaxObjectPerChunk + 1;
        }

        // do a full traversal of the chunk and don't stop even if we think it is a large chunk
        // we want the number of records to better report, in that case
        bool isLargeChunk = false;
        unsigned long long recCount = 0;
        ;
        RecordId dl;
        while (PlanExecutor::ADVANCED == exec->getNext(NULL, &dl)) {
            if (!isLargeChunk) {
                std::lock_guard<std::mutex> lk(_cloneLocsMutex);
                _cloneLocs.insert(dl);
            }

            if (++recCount > maxRecsWhenFull) {
                isLargeChunk = true;
                // continue on despite knowing that it will fail,
                // just to get the correct value for recCount.
            }
        }
        exec.reset();

        if (isLargeChunk) {
            std::lock_guard<std::mutex> sl(_mutex);
            warning() << "cannot move chunk: the maximum number of documents for a chunk is "
                      << maxRecsWhenFull << " , the maximum chunk size is " << maxChunkSize
                      << " , average document size is " << avgRecSize << ". Found " << recCount
                      << " documents in chunk "
                      << " ns: " << _ns << " " << _min << " -> " << _max << migrateLog;

            result.appendBool("chunkTooBig", true);
            result.appendNumber("estimatedChunkSize", (long long)(recCount * avgRecSize));
            errmsg = "chunk too big to move";
            return false;
        }

        log() << "moveChunk number of documents: " << cloneLocsRemaining() << migrateLog;

        txn->recoveryUnit()->abandonSnapshot();
        return true;
    }

    bool clone(OperationContext* txn, string& errmsg, BSONObjBuilder& result) {
        ElapsedTracker tracker(internalQueryExecYieldIterations, internalQueryExecYieldPeriodMS);

        int allocSize = 0;
        {
            AutoGetCollectionForRead ctx(txn, getNS());

            std::lock_guard<std::mutex> sl(_mutex);
            if (!_active) {
                errmsg = "not active";
                return false;
            }

            Collection* collection = ctx.getCollection();
            if (!collection) {
                errmsg = str::stream() << "collection " << _ns << " does not exist";
                return false;
            }

            allocSize = std::min(
                BSONObjMaxUserSize,
                static_cast<int>((12 + collection->averageObjectSize(txn)) * cloneLocsRemaining()));
        }

        bool isBufferFilled = false;
        BSONArrayBuilder clonedDocsArrayBuilder(allocSize);
        while (!isBufferFilled) {
            AutoGetCollectionForRead ctx(txn, getNS());

            std::lock_guard<std::mutex> sl(_mutex);
            if (!_active) {
                errmsg = "not active";
                return false;
            }

            // TODO: fix SERVER-16540 race

            Collection* collection = ctx.getCollection();

            if (!collection) {
                errmsg = str::stream() << "collection " << _ns << " does not exist";
                return false;
            }

            std::lock_guard<std::mutex> lk(_cloneLocsMutex);
            set<RecordId>::iterator cloneLocsIter = _cloneLocs.begin();
            for (; cloneLocsIter != _cloneLocs.end(); ++cloneLocsIter) {
                if (tracker.intervalHasElapsed())  // should I yield?
                    break;

                RecordId dl = *cloneLocsIter;
                Snapshotted<BSONObj> doc;
                if (!collection->findDoc(txn, dl, &doc)) {
                    // doc was deleted
                    continue;
                }

                // Use the builder size instead of accumulating 'doc's size so that we take
                // into consideration the overhead of BSONArray indices, and *always*
                // append one doc.
                if (clonedDocsArrayBuilder.arrSize() != 0 &&
                    (clonedDocsArrayBuilder.len() + doc.value().objsize() + 1024) >
                        BSONObjMaxUserSize) {
                    isBufferFilled = true;  // break out of outer while loop
                    break;
                }

                clonedDocsArrayBuilder.append(doc.value());
            }

            _cloneLocs.erase(_cloneLocs.begin(), cloneLocsIter);

            // Note: must be holding _cloneLocsMutex, don't move this inside while condition!
            if (_cloneLocs.empty()) {
                break;
            }
        }

        result.appendArray("objects", clonedDocsArrayBuilder.arr());
        return true;
    }

    void aboutToDelete(const RecordId& dl) {
        // Even though above we call findDoc to check for existance
        // that check only works for non-mmapv1 engines, and this is needed
        // for mmapv1.

        std::lock_guard<std::mutex> lk(_cloneLocsMutex);
        _cloneLocs.erase(dl);
    }

    std::size_t cloneLocsRemaining() {
        std::lock_guard<std::mutex> lk(_cloneLocsMutex);
        return _cloneLocs.size();
    }

    long long mbUsed() const {
        std::lock_guard<std::mutex> lk(_mutex);
        return _memoryUsed / (1024 * 1024);
    }

    bool getInCriticalSection() const {
        std::lock_guard<std::mutex> lk(_mutex);
        return _inCriticalSection;
    }

    void setInCriticalSection(bool b) {
        std::lock_guard<std::mutex> lk(_mutex);
        _inCriticalSection = b;
        _inCriticalSectionCV.notify_all();
    }

    std::string getNS() const {
        std::lock_guard<std::mutex> sl(_mutex);
        return _ns;
    }

    /**
     * @return true if we are NOT in the critical section
     */
    bool waitTillNotInCriticalSection(int maxSecondsToWait) {
        const auto deadline = system_clock::now() + seconds(maxSecondsToWait);
        std::unique_lock<std::mutex> lk(_mutex);
        while (_inCriticalSection) {
            if (std::cv_status::timeout == _inCriticalSectionCV.wait_until(lk, deadline))
                return false;
        }

        return true;
    }

    bool isActive() const {
        return _getActive();
    }

private:
    bool _getActive() const {
        std::lock_guard<std::mutex> lk(_mutex);
        return _active;
    }
    void _setActive(bool b) {
        std::lock_guard<std::mutex> lk(_mutex);
        _active = b;
    }

    /**
     * Used to commit work for LogOpForSharding. Used to keep track of changes in documents
     * that are part of a chunk being migrated.
     */
    class LogOpForShardingHandler : public RecoveryUnit::Change {
    public:
        /**
         * Invariant: idObj should belong to a document that is part of the active chunk
         * being migrated.
         */
        LogOpForShardingHandler(MigrateFromStatus* migrateFromStatus,
                                const BSONObj& idObj,
                                const char op)
            : _migrateFromStatus(migrateFromStatus), _idObj(idObj.getOwned()), _op(op) {}

        virtual void commit() {
            switch (_op) {
                case 'd': {
                    std::lock_guard<std::mutex> sl(_migrateFromStatus->_mutex);
                    _migrateFromStatus->_deleted.push_back(_idObj);
                    _migrateFromStatus->_memoryUsed += _idObj.firstElement().size() + 5;
                    break;
                }

                case 'i':
                case 'u': {
                    std::lock_guard<std::mutex> sl(_migrateFromStatus->_mutex);
                    _migrateFromStatus->_reload.push_back(_idObj);
                    _migrateFromStatus->_memoryUsed += _idObj.firstElement().size() + 5;
                    break;
                }

                default:
                    invariant(false);
            }
        }

        virtual void rollback() {}

    private:
        MigrateFromStatus* _migrateFromStatus;
        const BSONObj _idObj;
        const char _op;
    };

    /**
     * Used to receive invalidation notifications.
     *
     * XXX: move to the exec/ directory.
     */
    class DeleteNotificationStage : public PlanStage {
    public:
        virtual void invalidate(OperationContext* txn, const RecordId& dl, InvalidationType type);

        virtual StageState work(WorkingSetID* out) {
            invariant(false);
        }
        virtual bool isEOF() {
            invariant(false);
            return false;
        }
        virtual void kill() {}
        virtual void saveState() {
            invariant(false);
        }
        virtual void restoreState(OperationContext* opCtx) {
            invariant(false);
        }
        virtual unique_ptr<PlanStageStats> getStats() {
            invariant(false);
            return NULL;
        }
        virtual CommonStats* getCommonStats() const {
            invariant(false);
            return NULL;
        }
        virtual SpecificStats* getSpecificStats() const {
            invariant(false);
            return NULL;
        }
        virtual std::vector<PlanStage*> getChildren() const {
            vector<PlanStage*> empty;
            return empty;
        }
        virtual StageType stageType() const {
            return STAGE_NOTIFY_DELETE;
        }
    };

    //
    // All member variables are labeled with one of the following codes indicating the
    // synchronization rules for accessing them.
    //
    // (M)  Must hold _mutex for access.
    // (MG) For reads, _mutex *OR* Global IX Lock must be held.
    //      For writes, the _mutex *AND* (Global Shared or Exclusive Lock) must be held.
    // (C)  Must hold _cloneLocsMutex for access.
    //
    // Locking order:
    //
    // Global Lock -> _mutex -> _cloneLocsMutex

    mutable std::mutex _mutex;

    std::condition_variable _inCriticalSectionCV;  // (M)

    // Is migration currently in critical section. This can be used to block new writes.
    bool _inCriticalSection;  // (M)

    unique_ptr<PlanExecutor> _deleteNotifyExec;  // (M)

    // List of _id of documents that were modified that must be re-cloned.
    list<BSONObj> _reload;  // (M)

    // List of _id of documents that were deleted during clone that should be deleted later.
    list<BSONObj> _deleted;  // (M)

    // bytes in _reload + _deleted
    long long _memoryUsed;  // (M)

    // If a migration is currently active.
    bool _active;  // (MG)

    string _ns;                // (MG)
    BSONObj _min;              // (MG)
    BSONObj _max;              // (MG)
    BSONObj _shardKeyPattern;  // (MG)

    mutable std::mutex _cloneLocsMutex;

    // List of record id that needs to be transferred from here to the other side.
    set<RecordId> _cloneLocs;  // (C)

} migrateFromStatus;

void MigrateFromStatus::DeleteNotificationStage::invalidate(OperationContext* txn,
                                                            const RecordId& dl,
                                                            InvalidationType type) {
    if (type == INVALIDATION_DELETION) {
        migrateFromStatus.aboutToDelete(dl);
    }
}

struct MigrateStatusHolder {
    MigrateStatusHolder(OperationContext* txn,
                        const std::string& ns,
                        const BSONObj& min,
                        const BSONObj& max,
                        const BSONObj& shardKeyPattern)
        : _txn(txn) {
        _isAnotherMigrationActive = !migrateFromStatus.start(txn, ns, min, max, shardKeyPattern);
    }
    ~MigrateStatusHolder() {
        if (!_isAnotherMigrationActive) {
            migrateFromStatus.done(_txn);
        }
    }

    bool isAnotherMigrationActive() const {
        return _isAnotherMigrationActive;
    }

private:
    OperationContext* _txn;
    bool _isAnotherMigrationActive;
};

void logOpForSharding(OperationContext* txn,
                      const char* opstr,
                      const char* ns,
                      const BSONObj& obj,
                      BSONObj* patt,
                      bool notInActiveChunk) {
    migrateFromStatus.logOp(txn, opstr, ns, obj, patt, notInActiveChunk);
}

class TransferModsCommand : public ChunkCommandHelper {
public:
    void help(std::stringstream& h) const {
        h << "internal";
    }
    TransferModsCommand() : ChunkCommandHelper("_transferMods") {}
    virtual void addRequiredPrivileges(const std::string& dbname,
                                       const BSONObj& cmdObj,
                                       std::vector<Privilege>* out) {
        ActionSet actions;
        actions.addAction(ActionType::internal);
        out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
    }
    bool run(OperationContext* txn,
             const string&,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        return migrateFromStatus.transferMods(txn, errmsg, result);
    }
} transferModsCommand;


class InitialCloneCommand : public ChunkCommandHelper {
public:
    void help(std::stringstream& h) const {
        h << "internal";
    }
    InitialCloneCommand() : ChunkCommandHelper("_migrateClone") {}
    virtual void addRequiredPrivileges(const std::string& dbname,
                                       const BSONObj& cmdObj,
                                       std::vector<Privilege>* out) {
        ActionSet actions;
        actions.addAction(ActionType::internal);
        out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
    }
    bool run(OperationContext* txn,
             const string&,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        return migrateFromStatus.clone(txn, errmsg, result);
    }
} initialCloneCommand;

// Tests can pause / resume moveChunk's progress at each step by enabling / disabling each fail
// point.
MONGO_FP_DECLARE(moveChunkHangAtStep1);
MONGO_FP_DECLARE(moveChunkHangAtStep2);
MONGO_FP_DECLARE(moveChunkHangAtStep3);
MONGO_FP_DECLARE(moveChunkHangAtStep4);
MONGO_FP_DECLARE(moveChunkHangAtStep5);
MONGO_FP_DECLARE(moveChunkHangAtStep6);

/**
 * this is the main entry for moveChunk
 * called to initial a move
 * usually by a mongos
 * this is called on the "from" side
 *
 * Format:
 * {
 *   moveChunk: "namespace",
 *   from: "hostAndPort",
 *   fromShard: "shardName",
 *   to: "hostAndPort",
 *   toShard: "shardName",
 *   min: {},
 *   max: {},
 *   maxChunkBytes: numeric,
 *   configdb: "hostAndPort",
 *
 *   // optional
 *   secondaryThrottle: bool, //defaults to true.
 *   writeConcern: {} // applies to individual writes.
 * }
 */
class MoveChunkCommand : public Command {
public:
    MoveChunkCommand() : Command("moveChunk") {}
    virtual void help(std::stringstream& help) const {
        help << "should not be calling this directly";
    }

    virtual bool slaveOk() const {
        return false;
    }
    virtual bool adminOnly() const {
        return true;
    }
    virtual bool isWriteCommandForConfigServer() const {
        return false;
    }
    virtual Status checkAuthForCommand(ClientBasic* client,
                                       const std::string& dbname,
                                       const BSONObj& cmdObj) {
        if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
                ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
                ActionType::moveChunk)) {
            return Status(ErrorCodes::Unauthorized, "Unauthorized");
        }
        return Status::OK();
    }
    virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
        return parseNsFullyQualified(dbname, cmdObj);
    }

    bool run(OperationContext* txn,
             const string& dbname,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        // 1. Parse options
        // 2. Make sure my view is complete and lock the distributed lock to ensure shard
        //    metadata stability.
        // 3. Migration
        //    Retrieve all RecordIds, which need to be migrated in order to do as little seeking
        //    as possible during transfer. Retrieval of the RecordIds happens under a collection
        //    lock, but then the collection lock is dropped. This opens up an opportunity for
        //    repair or compact to invalidate these RecordIds, because these commands do not
        //    synchronized with migration. Note that data modifications are not a problem,
        //    because we are registered for change notifications.
        //
        // 4. pause till migrate caught up
        // 5. LOCK
        //    a) update my config, essentially locking
        //    b) finish migrate
        //    c) update config server
        //    d) logChange to config server
        // 6. wait for all current cursors to expire
        // 7. remove data locally

        // -------------------------------

        // 1.
        string ns = parseNs(dbname, cmdObj);

        // The shard addresses, redundant, but allows for validation
        string toShardHost = cmdObj["to"].str();
        string fromShardHost = cmdObj["from"].str();

        // The shard names
        string toShardName = cmdObj["toShard"].str();
        string fromShardName = cmdObj["fromShard"].str();

        // Process secondary throttle settings and assign defaults if necessary.
        BSONObj secThrottleObj;
        WriteConcernOptions writeConcern;
        Status status = writeConcern.parseSecondaryThrottle(cmdObj, &secThrottleObj);

        if (!status.isOK()) {
            if (status.code() != ErrorCodes::WriteConcernNotDefined) {
                warning() << status.toString();
                return appendCommandStatus(result, status);
            }

            writeConcern = getDefaultWriteConcern();
        } else {
            repl::ReplicationCoordinator* replCoordinator = repl::getGlobalReplicationCoordinator();

            if (replCoordinator->getReplicationMode() ==
                    repl::ReplicationCoordinator::modeMasterSlave &&
                writeConcern.shouldWaitForOtherNodes()) {
                warning() << "moveChunk cannot check if secondary throttle setting "
                          << writeConcern.toBSON()
                          << " can be enforced in a master slave configuration";
            }

            Status status = replCoordinator->checkIfWriteConcernCanBeSatisfied(writeConcern);
            if (!status.isOK() && status != ErrorCodes::NoReplicationEnabled) {
                warning() << status.toString();
                return appendCommandStatus(result, status);
            }
        }

        if (writeConcern.shouldWaitForOtherNodes() &&
            writeConcern.wTimeout == WriteConcernOptions::kNoTimeout) {
            // Don't allow no timeout.
            writeConcern.wTimeout = kDefaultWTimeoutMs;
        }

        // Do inline deletion
        bool waitForDelete = cmdObj["waitForDelete"].trueValue();
        if (waitForDelete) {
            log() << "moveChunk waiting for full cleanup after move";
        }

        BSONObj min = cmdObj["min"].Obj();
        BSONObj max = cmdObj["max"].Obj();
        BSONElement maxSizeElem = cmdObj["maxChunkSizeBytes"];

        if (ns.empty()) {
            errmsg = "need to specify namespace in command";
            return false;
        }

        if (toShardName.empty()) {
            errmsg = "need to specify shard to move chunk to";
            return false;
        }
        if (fromShardName.empty()) {
            errmsg = "need to specify shard to move chunk from";
            return false;
        }

        if (min.isEmpty()) {
            errmsg = "need to specify a min";
            return false;
        }

        if (max.isEmpty()) {
            errmsg = "need to specify a max";
            return false;
        }

        if (maxSizeElem.eoo() || !maxSizeElem.isNumber()) {
            errmsg = "need to specify maxChunkSizeBytes";
            return false;
        }
        const long long maxChunkSize = maxSizeElem.numberLong();  // in bytes

        // This could be the first call that enables sharding - make sure we initialize the
        // sharding state for this shard.
        if (!shardingState.enabled()) {
            if (cmdObj["configdb"].type() != String) {
                errmsg = "sharding not enabled";
                warning() << errmsg;
                return false;
            }
            string configdb = cmdObj["configdb"].String();
            ShardingState::initialize(configdb);
        }

        // Initialize our current shard name in the shard state if needed
        shardingState.gotShardName(fromShardName);

        // Make sure we're as up-to-date as possible with shard information
        // This catches the case where we had to previously changed a shard's host by
        // removing/adding a shard with the same name
        Shard::reloadShardInfo();

        ConnectionString configLoc =
            ConnectionString::parse(shardingState.getConfigServer(), errmsg);
        if (!configLoc.isValid()) {
            warning() << errmsg;
            return false;
        }

        MoveTimingHelper timing(
            txn, "from", ns, min, max, 6 /* steps */, &errmsg, toShardName, fromShardName);

        log() << "received moveChunk request: " << cmdObj << migrateLog;

        timing.done(1);
        MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep1);

        // 2.

        if (migrateFromStatus.isActive()) {
            errmsg = "migration already in progress";
            warning() << errmsg;
            return false;
        }

        //
        // Get the distributed lock
        //

        string whyMessage(str::stream() << "migrating chunk [" << minKey << ", " << maxKey
                                        << ") in " << ns);
        auto scopedDistLock = grid.catalogManager()->getDistLockManager()->lock(ns, whyMessage);

        if (!scopedDistLock.isOK()) {
            errmsg = stream() << "could not acquire collection lock for " << ns
                              << " to migrate chunk [" << minKey << "," << maxKey << ")"
                              << causedBy(scopedDistLock.getStatus());

            warning() << errmsg;
            return false;
        }

        BSONObj chunkInfo =
            BSON("min" << min << "max" << max << "from" << fromShardName << "to" << toShardName);

        grid.catalogManager()->logChange(
            txn->getClient()->clientAddress(true), "moveChunk.start", ns, chunkInfo);

        // Always refresh our metadata remotely
        ChunkVersion origShardVersion;
        Status refreshStatus = shardingState.refreshMetadataNow(txn, ns, &origShardVersion);

        if (!refreshStatus.isOK()) {
            errmsg = str::stream() << "moveChunk cannot start migrate of chunk "
                                   << "[" << minKey << "," << maxKey << ")"
                                   << causedBy(refreshStatus.reason());

            warning() << errmsg;
            return false;
        }

        if (origShardVersion.majorVersion() == 0) {
            // It makes no sense to migrate if our version is zero and we have no chunks
            errmsg = str::stream() << "moveChunk cannot start migrate of chunk "
                                   << "[" << minKey << "," << maxKey << ")"
                                   << " with zero shard version";

            warning() << errmsg;
            return false;
        }

        // From mongos >= v3.0.
        BSONElement epochElem(cmdObj["epoch"]);
        if (epochElem.type() == jstOID) {
            OID cmdEpoch = epochElem.OID();

            if (cmdEpoch != origShardVersion.epoch()) {
                errmsg = str::stream() << "moveChunk cannot move chunk "
                                       << "[" << minKey << "," << maxKey << "), "
                                       << "collection may have been dropped. "
                                       << "current epoch: " << origShardVersion.epoch()
                                       << ", cmd epoch: " << cmdEpoch;
                warning() << errmsg;
                return false;
            }
        }

        // Get collection metadata
        const CollectionMetadataPtr origCollMetadata(shardingState.getCollectionMetadata(ns));

        // With nonzero shard version, we must have metadata
        invariant(NULL != origCollMetadata);

        ChunkVersion origCollVersion = origCollMetadata->getCollVersion();
        BSONObj shardKeyPattern = origCollMetadata->getKeyPattern();

        // With nonzero shard version, we must have a coll version >= our shard version
        invariant(origCollVersion >= origShardVersion);

        // With nonzero shard version, we must have a shard key
        invariant(!shardKeyPattern.isEmpty());

        ChunkType origChunk;
        if (!origCollMetadata->getNextChunk(min, &origChunk) || origChunk.getMin().woCompare(min) ||
            origChunk.getMax().woCompare(max)) {
            // Our boundaries are different from those passed in
            errmsg = str::stream() << "moveChunk cannot find chunk "
                                   << "[" << minKey << "," << maxKey << ")"
                                   << " to migrate, the chunk boundaries may be stale";

            warning() << errmsg;
            return false;
        }

        log() << "moveChunk request accepted at version " << origShardVersion;

        timing.done(2);
        MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep2);

        // 3.
        MigrateStatusHolder statusHolder(txn, ns, min, max, shardKeyPattern);

        if (statusHolder.isAnotherMigrationActive()) {
            errmsg = "moveChunk is already in progress from this shard";
            warning() << errmsg;
            return false;
        }

        ConnectionString fromShardCS;
        ConnectionString toShardCS;

        // Resolve the shard connection strings.
        {
            std::shared_ptr<Shard> fromShard = grid.shardRegistry()->getShard(fromShardName);
            uassert(28674,
                    str::stream() << "Source shard " << fromShardName
                                  << " is missing. This indicates metadata corruption.",
                    fromShard);

            fromShardCS = fromShard->getConnString();

            std::shared_ptr<Shard> toShard = grid.shardRegistry()->getShard(toShardName);
            uassert(28675,
                    str::stream() << "Destination shard " << toShardName
                                  << " is missing. This indicates metadata corruption.",
                    toShard);

            toShardCS = toShard->getConnString();
        }

        {
            // See comment at the top of the function for more information on what
            // synchronization is used here.
            if (!migrateFromStatus.storeCurrentLocs(txn, maxChunkSize, errmsg, result)) {
                warning() << errmsg;
                return false;
            }

            ScopedDbConnection connTo(toShardCS);
            BSONObj res;
            bool ok;

            const bool isSecondaryThrottle(writeConcern.shouldWaitForOtherNodes());

            BSONObjBuilder recvChunkStartBuilder;
            recvChunkStartBuilder.append("_recvChunkStart", ns);
            recvChunkStartBuilder.append("from", fromShardCS.toString());
            recvChunkStartBuilder.append("fromShardName", fromShardName);
            recvChunkStartBuilder.append("toShardName", toShardName);
            recvChunkStartBuilder.append("min", min);
            recvChunkStartBuilder.append("max", max);
            recvChunkStartBuilder.append("shardKeyPattern", shardKeyPattern);
            recvChunkStartBuilder.append("configServer", shardingState.getConfigServer());
            recvChunkStartBuilder.append("secondaryThrottle", isSecondaryThrottle);

            // Follow the same convention in moveChunk.
            if (isSecondaryThrottle && !secThrottleObj.isEmpty()) {
                recvChunkStartBuilder.append("writeConcern", secThrottleObj);
            }

            try {
                ok = connTo->runCommand("admin", recvChunkStartBuilder.done(), res);
            } catch (DBException& e) {
                errmsg = str::stream() << "moveChunk could not contact to: shard " << toShardName
                                       << " to start transfer" << causedBy(e);
                warning() << errmsg;
                return false;
            }

            connTo.done();

            if (!ok) {
                errmsg = "moveChunk failed to engage TO-shard in the data transfer: ";
                verify(res["errmsg"].type());
                errmsg += res["errmsg"].String();
                result.append("cause", res);
                warning() << errmsg;
                return false;
            }
        }
        timing.done(3);
        MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep3);

        // 4.

        // Track last result from TO shard for sanity check
        BSONObj res;
        for (int i = 0; i < 86400; i++) {  // don't want a single chunk move to take more than a day
            invariant(!txn->lockState()->isLocked());

            // Exponential sleep backoff, up to 1024ms. Don't sleep much on the first few
            // iterations, since we want empty chunk migrations to be fast.
            sleepmillis(1 << std::min(i, 10));

            ScopedDbConnection conn(toShardCS);
            bool ok;
            res = BSONObj();
            try {
                ok = conn->runCommand("admin", BSON("_recvChunkStatus" << 1), res);
                res = res.getOwned();
            } catch (DBException& e) {
                errmsg = str::stream() << "moveChunk could not contact to: shard " << toShardName
                                       << " to monitor transfer" << causedBy(e);
                warning() << errmsg;
                return false;
            }

            conn.done();

            if (res["ns"].str() != ns || res["from"].str() != fromShardCS.toString() ||
                !res["min"].isABSONObj() || res["min"].Obj().woCompare(min) != 0 ||
                !res["max"].isABSONObj() || res["max"].Obj().woCompare(max) != 0) {
                // This can happen when the destination aborted the migration and
                // received another recvChunk before this thread sees the transition
                // to the abort state. This is currently possible only if multiple migrations
                // are happening at once. This is an unfortunate consequence of the shards not
                // being able to keep track of multiple incoming and outgoing migrations.
                errmsg = str::stream() << "Destination shard aborted migration, "
                                          "now running a new one: " << res;
                warning() << errmsg;
                return false;
            }

            LOG(0) << "moveChunk data transfer progress: " << res
                   << " my mem used: " << migrateFromStatus.mbUsed() << migrateLog;

            if (!ok || res["state"].String() == "fail") {
                warning() << "moveChunk error transferring data caused migration abort: " << res
                          << migrateLog;
                errmsg = "data transfer error";
                result.append("cause", res);
                return false;
            }

            if (res["state"].String() == "steady")
                break;

            if (migrateFromStatus.mbUsed() > (500 * 1024 * 1024)) {
                // This is too much memory for us to use for this so we're going to abort
                // the migrate
                ScopedDbConnection conn(toShardCS);

                BSONObj res;
                if (!conn->runCommand("admin", BSON("_recvChunkAbort" << 1), res)) {
                    warning() << "Error encountered while trying to abort migration on "
                              << "destination shard" << toShardCS;
                }

                res = res.getOwned();
                conn.done();
                error() << "aborting migrate because too much memory used res: " << res
                        << migrateLog;
                errmsg = "aborting migrate because too much memory used";
                result.appendBool("split", true);
                return false;
            }

            txn->checkForInterrupt();
        }

        timing.done(4);
        MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep4);

        // 5.

        // Before we get into the critical section of the migration, let's double check
        // that the docs have been cloned, the config servers are reachable,
        // and the lock is in place.
        log() << "About to check if it is safe to enter critical section";

        // Ensure all cloned docs have actually been transferred
        std::size_t locsRemaining = migrateFromStatus.cloneLocsRemaining();
        if (locsRemaining != 0) {
            errmsg = str::stream() << "moveChunk cannot enter critical section before all data is"
                                   << " cloned, " << locsRemaining << " locs were not transferred"
                                   << " but to-shard reported " << res;

            // Should never happen, but safe to abort before critical section
            error() << errmsg << migrateLog;
            dassert(false);
            return false;
        }

        // Ensure distributed lock still held
        Status lockStatus = scopedDistLock.getValue().checkStatus();
        if (!lockStatus.isOK()) {
            errmsg = str::stream() << "not entering migrate critical section because "
                                   << lockStatus.toString();
            warning() << errmsg;
            return false;
        }

        log() << "About to enter migrate critical section";

        {
            // 5.a
            // we're under the collection lock here, so no other migrate can change maxVersion
            // or CollectionMetadata state
            migrateFromStatus.setInCriticalSection(true);
            ChunkVersion myVersion = origCollVersion;
            myVersion.incMajor();

            {
                ScopedTransaction transaction(txn, MODE_IX);
                Lock::DBLock lk(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);
                verify(myVersion > shardingState.getVersion(ns));

                // bump the metadata's version up and "forget" about the chunk being moved
                // this is not the commit point but in practice the state in this shard won't
                // until the commit it done
                shardingState.donateChunk(txn, ns, min, max, myVersion);
            }

            log() << "moveChunk setting version to: " << myVersion << migrateLog;

            // 5.b
            // we're under the collection lock here, too, so we can undo the chunk donation because
            // no other state change could be ongoing

            BSONObj res;
            bool ok;

            try {
                ScopedDbConnection connTo(toShardCS, 35.0);
                ok = connTo->runCommand("admin", BSON("_recvChunkCommit" << 1), res);
                connTo.done();
            } catch (DBException& e) {
                errmsg = str::stream() << "moveChunk could not contact to: shard "
                                       << toShardCS.toString() << " to commit transfer"
                                       << causedBy(e);
                warning() << errmsg;
                ok = false;
            }

            if (!ok || MONGO_FAIL_POINT(failMigrationCommit)) {
                log() << "moveChunk migrate commit not accepted by TO-shard: " << res
                      << " resetting shard version to: " << origShardVersion << migrateLog;
                {
                    ScopedTransaction transaction(txn, MODE_IX);
                    Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                    Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);

                    log() << "moveChunk collection lock acquired to reset shard version from "
                             "failed migration";

                    // revert the chunk manager back to the state before "forgetting" about the
                    // chunk
                    shardingState.undoDonateChunk(txn, ns, origCollMetadata);
                }
                log() << "Shard version successfully reset to clean up failed migration";

                errmsg = "_recvChunkCommit failed!";
                result.append("cause", res);
                return false;
            }

            log() << "moveChunk migrate commit accepted by TO-shard: " << res << migrateLog;

            // 5.c

            // version at which the next highest lastmod will be set
            // if the chunk being moved is the last in the shard, nextVersion is that chunk's
            // lastmod otherwise the highest version is from the chunk being bumped on the
            // FROM-shard
            ChunkVersion nextVersion;

            // we want to go only once to the configDB but perhaps change two chunks, the one being
            // migrated and another local one (so to bump version for the entire shard)
            // we use the 'applyOps' mechanism to group the two updates and make them safer
            // TODO pull config update code to a module

            BSONArrayBuilder updates;
            {
                // update for the chunk being moved
                BSONObjBuilder op;
                op.append("op", "u");
                op.appendBool("b", false /* no upserting */);
                op.append("ns", ChunkType::ConfigNS);

                BSONObjBuilder n(op.subobjStart("o"));
                n.append(ChunkType::name(), Chunk::genID(ns, min));
                myVersion.addToBSON(n, ChunkType::DEPRECATED_lastmod());
                n.append(ChunkType::ns(), ns);
                n.append(ChunkType::min(), min);
                n.append(ChunkType::max(), max);
                n.append(ChunkType::shard(), toShardName);
                n.done();

                BSONObjBuilder q(op.subobjStart("o2"));
                q.append(ChunkType::name(), Chunk::genID(ns, min));
                q.done();

                updates.append(op.obj());
            }

            nextVersion = myVersion;

            // if we have chunks left on the FROM shard, update the version of one of them as
            // well.  we can figure that out by grabbing the metadata installed on 5.a

            const CollectionMetadataPtr bumpedCollMetadata(shardingState.getCollectionMetadata(ns));
            if (bumpedCollMetadata->getNumChunks() > 0) {
                // get another chunk on that shard
                ChunkType bumpChunk;
                bool chunkRes =
                    bumpedCollMetadata->getNextChunk(bumpedCollMetadata->getMinKey(), &bumpChunk);
                BSONObj bumpMin = bumpChunk.getMin();
                BSONObj bumpMax = bumpChunk.getMax();

                (void)chunkRes;  // for compile warning on non-debug
                dassert(chunkRes);
                dassert(bumpMin.woCompare(min) != 0);

                BSONObjBuilder op;
                op.append("op", "u");
                op.appendBool("b", false);
                op.append("ns", ChunkType::ConfigNS);

                nextVersion.incMinor();  // same as used on donateChunk
                BSONObjBuilder n(op.subobjStart("o"));
                n.append(ChunkType::name(), Chunk::genID(ns, bumpMin));
                nextVersion.addToBSON(n, ChunkType::DEPRECATED_lastmod());
                n.append(ChunkType::ns(), ns);
                n.append(ChunkType::min(), bumpMin);
                n.append(ChunkType::max(), bumpMax);
                n.append(ChunkType::shard(), fromShardName);
                n.done();

                BSONObjBuilder q(op.subobjStart("o2"));
                q.append(ChunkType::name(), Chunk::genID(ns, bumpMin));
                q.done();

                updates.append(op.obj());

                log() << "moveChunk updating self version to: " << nextVersion << " through "
                      << bumpMin << " -> " << bumpMax << " for collection '" << ns << "'"
                      << migrateLog;

            } else {
                log() << "moveChunk moved last chunk out for collection '" << ns << "'"
                      << migrateLog;
            }

            BSONArrayBuilder preCond;
            {
                BSONObjBuilder b;
                b.append("ns", ChunkType::ConfigNS);
                b.append("q",
                         BSON("query" << BSON(ChunkType::ns(ns)) << "orderby"
                                      << BSON(ChunkType::DEPRECATED_lastmod() << -1)));
                {
                    BSONObjBuilder bb(b.subobjStart("res"));
                    // TODO: For backwards compatibility, we can't yet require an epoch here
                    bb.appendTimestamp(ChunkType::DEPRECATED_lastmod(), origCollVersion.toLong());
                    bb.done();
                }
                preCond.append(b.obj());
            }

            Status applyOpsStatus{Status::OK()};
            try {
                // For testing migration failures
                if (MONGO_FAIL_POINT(failMigrationConfigWritePrepare)) {
                    throw DBException("mock migration failure before config write",
                                      PrepareConfigsFailedCode);
                }

                applyOpsStatus =
                    grid.catalogManager()->applyChunkOpsDeprecated(updates.arr(), preCond.arr());

                if (MONGO_FAIL_POINT(failMigrationApplyOps)) {
                    throw SocketException(SocketException::RECV_ERROR,
                                          shardingState.getConfigServer());
                }
            } catch (const DBException& ex) {
                warning() << ex << migrateLog;
                applyOpsStatus = ex.toStatus();
            }

            if (applyOpsStatus == ErrorCodes::PrepareConfigsFailedCode) {
                // In the process of issuing the migrate commit, the SyncClusterConnection
                // checks that the config servers are reachable. If they are not, we are
                // sure that the applyOps command was not sent to any of the configs, so we
                // can safely back out of the migration here, by resetting the shard
                // version that we bumped up to in the donateChunk() call above.

                log() << "About to acquire moveChunk coll lock to reset shard version from "
                      << "failed migration";

                {
                    ScopedTransaction transaction(txn, MODE_IX);
                    Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                    Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);

                    // Revert the metadata back to the state before "forgetting"
                    // about the chunk.
                    shardingState.undoDonateChunk(txn, ns, origCollMetadata);
                }

                log() << "Shard version successfully reset to clean up failed migration";

                errmsg = "Failed to send migrate commit to configs because " + errmsg;
                return false;

            } else if (!applyOpsStatus.isOK()) {
                // this could be a blip in the connectivity
                // wait out a few seconds and check if the commit request made it
                //
                // if the commit made it to the config, we'll see the chunk in the new shard and
                // there's no action
                // if the commit did not make it, currently the only way to fix this state is to
                // bounce the mongod so that the old state (before migrating) be brought in

                warning() << "moveChunk commit outcome ongoing" << migrateLog;
                sleepsecs(10);

                // look for the chunk in this shard whose version got bumped
                // we assume that if that mod made it to the config, the applyOps was successful
                try {
                    std::vector<ChunkType> newestChunk;
                    Status status = grid.catalogManager()->getChunks(
                        BSON(ChunkType::ns(ns)),
                        BSON(ChunkType::DEPRECATED_lastmod() << -1),
                        1,
                        &newestChunk);
                    uassertStatusOK(status);

                    ChunkVersion checkVersion;
                    if (!newestChunk.empty()) {
                        invariant(newestChunk.size() == 1);
                        checkVersion = newestChunk[0].getVersion();
                    }

                    if (checkVersion.equals(nextVersion)) {
                        log() << "moveChunk commit confirmed" << migrateLog;
                        errmsg.clear();

                    } else {
                        error() << "moveChunk commit failed: version is at " << checkVersion
                                << " instead of " << nextVersion << migrateLog;
                        error() << "TERMINATING" << migrateLog;
                        dbexit(EXIT_SHARDING_ERROR);
                    }
                } catch (...) {
                    error() << "moveChunk failed to get confirmation of commit" << migrateLog;
                    error() << "TERMINATING" << migrateLog;
                    dbexit(EXIT_SHARDING_ERROR);
                }
            }

            migrateFromStatus.setInCriticalSection(false);

            // 5.d
            BSONObjBuilder commitInfo;
            commitInfo.appendElements(chunkInfo);
            if (res["counts"].type() == Object) {
                commitInfo.appendElements(res["counts"].Obj());
            }

            grid.catalogManager()->logChange(
                txn->getClient()->clientAddress(true), "moveChunk.commit", ns, commitInfo.obj());
        }

        migrateFromStatus.done(txn);
        timing.done(5);
        MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep5);

        // 6.
        // NOTE: It is important that the distributed collection lock be held for this step.
        RangeDeleter* deleter = getDeleter();
        RangeDeleterOptions deleterOptions(
            KeyRange(ns, min.getOwned(), max.getOwned(), shardKeyPattern));
        deleterOptions.writeConcern = writeConcern;
        deleterOptions.waitForOpenCursors = true;
        deleterOptions.fromMigrate = true;
        deleterOptions.onlyRemoveOrphanedDocs = true;
        deleterOptions.removeSaverReason = "post-cleanup";

        if (waitForDelete) {
            log() << "doing delete inline for cleanup of chunk data" << migrateLog;

            string errMsg;
            // This is an immediate delete, and as a consequence, there could be more
            // deletes happening simultaneously than there are deleter worker threads.
            if (!deleter->deleteNow(txn, deleterOptions, &errMsg)) {
                log() << "Error occured while performing cleanup: " << errMsg;
            }
        } else {
            log() << "forking for cleanup of chunk data" << migrateLog;

            string errMsg;
            if (!deleter->queueDelete(txn,
                                      deleterOptions,
                                      NULL,  // Don't want to be notified.
                                      &errMsg)) {
                log() << "could not queue migration cleanup: " << errMsg;
            }
        }
        timing.done(6);
        MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep6);

        return true;
    }

} moveChunkCmd;

bool ShardingState::inCriticalMigrateSection() {
    return migrateFromStatus.getInCriticalSection();
}

bool ShardingState::waitTillNotInCriticalSection(int maxSecondsToWait) {
    return migrateFromStatus.waitTillNotInCriticalSection(maxSecondsToWait);
}

/* -----
   below this are the "to" side commands

   command to initiate
   worker thread
     does initial clone
     pulls initial change set
     keeps pulling
     keeps state
   command to get state
   commend to "commit"
*/

// Enabling / disabling these fail points pauses / resumes MigrateStatus::_go(), the thread
// that receives a chunk migration from the donor.
MONGO_FP_DECLARE(migrateThreadHangAtStep1);
MONGO_FP_DECLARE(migrateThreadHangAtStep2);
MONGO_FP_DECLARE(migrateThreadHangAtStep3);
MONGO_FP_DECLARE(migrateThreadHangAtStep4);
MONGO_FP_DECLARE(migrateThreadHangAtStep5);

class MigrateStatus {
public:
    enum State { READY, CLONE, CATCHUP, STEADY, COMMIT_START, DONE, FAIL, ABORT };

    MigrateStatus()
        : _active(false),
          _numCloned(0),
          _clonedBytes(0),
          _numCatchup(0),
          _numSteady(0),
          _state(READY) {}

    void setState(State newState) {
        std::lock_guard<std::mutex> sl(_mutex);
        _state = newState;
    }

    State getState() const {
        std::lock_guard<std::mutex> sl(_mutex);
        return _state;
    }

    /**
     * Returns OK if preparation was successful.
     */
    Status prepare(const std::string& ns,
                   const std::string& fromShard,
                   const BSONObj& min,
                   const BSONObj& max,
                   const BSONObj& shardKeyPattern) {
        std::lock_guard<std::mutex> lk(_mutex);

        if (_active) {
            return Status(ErrorCodes::ConflictingOperationInProgress,
                          str::stream() << "Active migration already in progress "
                                        << "ns: " << _ns << ", from: " << _from << ", min: " << _min
                                        << ", max: " << _max);
        }

        _state = READY;
        _errmsg = "";

        _ns = ns;
        _from = fromShard;
        _min = min;
        _max = max;
        _shardKeyPattern = shardKeyPattern;

        _numCloned = 0;
        _clonedBytes = 0;
        _numCatchup = 0;
        _numSteady = 0;

        _active = true;

        return Status::OK();
    }

    void go(OperationContext* txn,
            const std::string& ns,
            BSONObj min,
            BSONObj max,
            BSONObj shardKeyPattern,
            const std::string& fromShard,
            const OID& epoch,
            const WriteConcernOptions& writeConcern) {
        try {
            _go(txn, ns, min, max, shardKeyPattern, fromShard, epoch, writeConcern);
        } catch (std::exception& e) {
            {
                std::lock_guard<std::mutex> sl(_mutex);
                _state = FAIL;
                _errmsg = e.what();
            }

            error() << "migrate failed: " << e.what() << migrateLog;
        } catch (...) {
            {
                std::lock_guard<std::mutex> sl(_mutex);
                _state = FAIL;
                _errmsg = "UNKNOWN ERROR";
            }

            error() << "migrate failed with unknown exception" << migrateLog;
        }

        if (getState() != DONE) {
            // Unprotect the range if needed/possible on unsuccessful TO migration
            ScopedTransaction transaction(txn, MODE_IX);
            Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
            Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);

            string errMsg;
            if (!shardingState.forgetPending(txn, ns, min, max, epoch, &errMsg)) {
                warning() << errMsg;
            }
        }

        setActive(false);
    }

    void _go(OperationContext* txn,
             const std::string& ns,
             BSONObj min,
             BSONObj max,
             BSONObj shardKeyPattern,
             const std::string& fromShard,
             const OID& epoch,
             const WriteConcernOptions& writeConcern) {
        verify(getActive());
        verify(getState() == READY);
        verify(!min.isEmpty());
        verify(!max.isEmpty());

        DisableDocumentValidation validationDisabler(txn);

        log() << "starting receiving-end of migration of chunk " << min << " -> " << max
              << " for collection " << ns << " from " << fromShard << " at epoch "
              << epoch.toString();

        string errmsg;
        MoveTimingHelper timing(txn, "to", ns, min, max, 5 /* steps */, &errmsg, "", "");

        ScopedDbConnection conn(fromShard);
        conn->getLastError();  // just test connection

        NamespaceString nss(ns);
        {
            // 0. copy system.namespaces entry if collection doesn't already exist
            OldClientWriteContext ctx(txn, ns);
            if (!repl::getGlobalReplicationCoordinator()->canAcceptWritesFor(nss)) {
                errmsg = str::stream() << "Not primary during migration: " << ns
                                       << ": checking if collection exists";
                warning() << errmsg;
                setState(FAIL);
                return;
            }

            // Only copy if ns doesn't already exist
            Database* db = ctx.db();
            Collection* collection = db->getCollection(ns);

            if (!collection) {
                list<BSONObj> infos = conn->getCollectionInfos(
                    nsToDatabase(ns), BSON("name" << nsToCollectionSubstring(ns)));

                BSONObj options;
                if (infos.size() > 0) {
                    BSONObj entry = infos.front();
                    if (entry["options"].isABSONObj()) {
                        options = entry["options"].Obj();
                    }
                }

                WriteUnitOfWork wuow(txn);
                Status status = userCreateNS(txn, db, ns, options, false);
                if (!status.isOK()) {
                    warning() << "failed to create collection [" << ns << "] "
                              << " with options " << options << ": " << status;
                }
                wuow.commit();
            }
        }

        {
            // 1. copy indexes

            vector<BSONObj> indexSpecs;
            {
                const std::list<BSONObj> indexes = conn->getIndexSpecs(ns);
                indexSpecs.insert(indexSpecs.begin(), indexes.begin(), indexes.end());
            }

            ScopedTransaction transaction(txn, MODE_IX);
            Lock::DBLock lk(txn->lockState(), nsToDatabaseSubstring(ns), MODE_X);
            OldClientContext ctx(txn, ns);
            if (!repl::getGlobalReplicationCoordinator()->canAcceptWritesFor(nss)) {
                errmsg = str::stream() << "Not primary during migration: " << ns;
                warning() << errmsg;
                setState(FAIL);
                return;
            }

            Database* db = ctx.db();
            Collection* collection = db->getCollection(ns);
            if (!collection) {
                errmsg = str::stream() << "collection dropped during migration: " << ns;
                warning() << errmsg;
                setState(FAIL);
                return;
            }

            MultiIndexBlock indexer(txn, collection);

            indexer.removeExistingIndexes(&indexSpecs);

            if (!indexSpecs.empty()) {
                // Only copy indexes if the collection does not have any documents.
                if (collection->numRecords(txn) > 0) {
                    errmsg = str::stream() << "aborting migration, shard is missing "
                                           << indexSpecs.size() << " indexes and "
                                           << "collection is not empty. Non-trivial "
                                           << "index creation should be scheduled manually";
                    warning() << errmsg;
                    setState(FAIL);
                    return;
                }

                Status status = indexer.init(indexSpecs);
                if (!status.isOK()) {
                    errmsg = str::stream() << "failed to create index before migrating data. "
                                           << " error: " << status.toString();
                    warning() << errmsg;
                    setState(FAIL);
                    return;
                }

                status = indexer.insertAllDocumentsInCollection();
                if (!status.isOK()) {
                    errmsg = str::stream() << "failed to create index before migrating data. "
                                           << " error: " << status.toString();
                    warning() << errmsg;
                    setState(FAIL);
                    return;
                }

                WriteUnitOfWork wunit(txn);
                indexer.commit();

                for (size_t i = 0; i < indexSpecs.size(); i++) {
                    // make sure to create index on secondaries as well
                    getGlobalServiceContext()->getOpObserver()->onCreateIndex(
                        txn, db->getSystemIndexesName(), indexSpecs[i], true /* fromMigrate */);
                }

                wunit.commit();
            }

            timing.done(1);
            MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep1);
        }

        {
            // 2. delete any data already in range
            RangeDeleterOptions deleterOptions(
                KeyRange(ns, min.getOwned(), max.getOwned(), shardKeyPattern));
            deleterOptions.writeConcern = writeConcern;
            // No need to wait since all existing cursors will filter out this range when
            // returning the results.
            deleterOptions.waitForOpenCursors = false;
            deleterOptions.fromMigrate = true;
            deleterOptions.onlyRemoveOrphanedDocs = true;
            deleterOptions.removeSaverReason = "preCleanup";

            string errMsg;

            if (!getDeleter()->deleteNow(txn, deleterOptions, &errMsg)) {
                warning() << "Failed to queue delete for migrate abort: " << errMsg;
                setState(FAIL);
                return;
            }

            {
                // Protect the range by noting that we're now starting a migration to it
                ScopedTransaction transaction(txn, MODE_IX);
                Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);

                if (!shardingState.notePending(txn, ns, min, max, epoch, &errmsg)) {
                    warning() << errmsg;
                    setState(FAIL);
                    return;
                }
            }

            timing.done(2);
            MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep2);
        }

        State currentState = getState();
        if (currentState == FAIL || currentState == ABORT) {
            string errMsg;
            RangeDeleterOptions deleterOptions(
                KeyRange(ns, min.getOwned(), max.getOwned(), shardKeyPattern));
            deleterOptions.writeConcern = writeConcern;
            // No need to wait since all existing cursors will filter out this range when
            // returning the results.
            deleterOptions.waitForOpenCursors = false;
            deleterOptions.fromMigrate = true;
            deleterOptions.onlyRemoveOrphanedDocs = true;

            if (!getDeleter()->queueDelete(txn, deleterOptions, NULL /* notifier */, &errMsg)) {
                warning() << "Failed to queue delete for migrate abort: " << errMsg;
            }
        }

        {
            // 3. initial bulk clone
            setState(CLONE);

            while (true) {
                BSONObj res;
                if (!conn->runCommand("admin",
                                      BSON("_migrateClone" << 1),
                                      res)) {  // gets array of objects to copy, in disk order
                    setState(FAIL);
                    errmsg = "_migrateClone failed: ";
                    errmsg += res.toString();
                    error() << errmsg << migrateLog;
                    conn.done();
                    return;
                }

                BSONObj arr = res["objects"].Obj();
                int thisTime = 0;

                BSONObjIterator i(arr);
                while (i.more()) {
                    txn->checkForInterrupt();

                    if (getState() == ABORT) {
                        errmsg = str::stream() << "Migration abort requested while "
                                               << "copying documents";
                        error() << errmsg << migrateLog;
                        return;
                    }

                    BSONObj docToClone = i.next().Obj();
                    {
                        OldClientWriteContext cx(txn, ns);

                        BSONObj localDoc;
                        if (willOverrideLocalId(txn,
                                                ns,
                                                min,
                                                max,
                                                shardKeyPattern,
                                                cx.db(),
                                                docToClone,
                                                &localDoc)) {
                            string errMsg = str::stream() << "cannot migrate chunk, local document "
                                                          << localDoc << " has same _id as cloned "
                                                          << "remote document " << docToClone;

                            warning() << errMsg;

                            // Exception will abort migration cleanly
                            uasserted(16976, errMsg);
                        }

                        Helpers::upsert(txn, ns, docToClone, true);
                    }
                    thisTime++;

                    {
                        std::lock_guard<std::mutex> statsLock(_mutex);
                        _numCloned++;
                        _clonedBytes += docToClone.objsize();
                    }

                    if (writeConcern.shouldWaitForOtherNodes()) {
                        repl::ReplicationCoordinator::StatusAndDuration replStatus =
                            repl::getGlobalReplicationCoordinator()->awaitReplication(
                                txn,
                                repl::ReplClientInfo::forClient(txn->getClient()).getLastOp(),
                                writeConcern);
                        if (replStatus.status.code() == ErrorCodes::WriteConcernFailed) {
                            warning() << "secondaryThrottle on, but doc insert timed out; "
                                         "continuing";
                        } else {
                            massertStatusOK(replStatus.status);
                        }
                    }
                }

                if (thisTime == 0)
                    break;
            }

            timing.done(3);
            MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep3);
        }

        // If running on a replicated system, we'll need to flush the docs we cloned to the
        // secondaries
        repl::OpTime lastOpApplied = repl::ReplClientInfo::forClient(txn->getClient()).getLastOp();

        {
            // 4. do bulk of mods
            setState(CATCHUP);
            while (true) {
                BSONObj res;
                if (!conn->runCommand("admin", BSON("_transferMods" << 1), res)) {
                    setState(FAIL);
                    errmsg = "_transferMods failed: ";
                    errmsg += res.toString();
                    error() << "_transferMods failed: " << res << migrateLog;
                    conn.done();
                    return;
                }
                if (res["size"].number() == 0)
                    break;

                apply(txn, ns, min, max, shardKeyPattern, res, &lastOpApplied);

                const int maxIterations = 3600 * 50;
                int i;
                for (i = 0; i < maxIterations; i++) {
                    txn->checkForInterrupt();

                    if (getState() == ABORT) {
                        errmsg = str::stream() << "Migration abort requested while waiting "
                                               << "for replication at catch up stage";
                        error() << errmsg << migrateLog;

                        return;
                    }

                    if (opReplicatedEnough(txn, lastOpApplied, writeConcern))
                        break;

                    if (i > 100) {
                        warning() << "secondaries having hard time keeping up with migrate"
                                  << migrateLog;
                    }

                    sleepmillis(20);
                }

                if (i == maxIterations) {
                    errmsg = "secondary can't keep up with migrate";
                    error() << errmsg << migrateLog;
                    conn.done();
                    setState(FAIL);
                    return;
                }
            }

            timing.done(4);
            MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep4);
        }

        {
            // pause to wait for replication
            // this will prevent us from going into critical section until we're ready
            Timer t;
            while (t.minutes() < 600) {
                txn->checkForInterrupt();

                if (getState() == ABORT) {
                    errmsg = "Migration abort requested while waiting for replication";
                    error() << errmsg << migrateLog;
                    return;
                }

                log() << "Waiting for replication to catch up before entering critical section";
                if (flushPendingWrites(txn, ns, min, max, lastOpApplied, writeConcern))
                    break;
                sleepsecs(1);
            }

            if (t.minutes() >= 600) {
                setState(FAIL);
                errmsg = "Cannot go to critical section because secondaries cannot keep up";
                error() << errmsg << migrateLog;
                return;
            }
        }

        {
            // 5. wait for commit

            setState(STEADY);
            bool transferAfterCommit = false;
            while (getState() == STEADY || getState() == COMMIT_START) {
                txn->checkForInterrupt();

                // Make sure we do at least one transfer after recv'ing the commit message
                // If we aren't sure that at least one transfer happens *after* our state
                // changes to COMMIT_START, there could be mods still on the FROM shard that
                // got logged *after* our _transferMods but *before* the critical section.
                if (getState() == COMMIT_START)
                    transferAfterCommit = true;

                BSONObj res;
                if (!conn->runCommand("admin", BSON("_transferMods" << 1), res)) {
                    log() << "_transferMods failed in STEADY state: " << res << migrateLog;
                    errmsg = res.toString();
                    setState(FAIL);
                    conn.done();
                    return;
                }

                if (res["size"].number() > 0 &&
                    apply(txn, ns, min, max, shardKeyPattern, res, &lastOpApplied)) {
                    continue;
                }

                if (getState() == ABORT) {
                    return;
                }

                // We know we're finished when:
                // 1) The from side has told us that it has locked writes (COMMIT_START)
                // 2) We've checked at least one more time for un-transmitted mods
                if (getState() == COMMIT_START && transferAfterCommit == true) {
                    if (flushPendingWrites(txn, ns, min, max, lastOpApplied, writeConcern))
                        break;
                }

                // Only sleep if we aren't committing
                if (getState() == STEADY)
                    sleepmillis(10);
            }

            if (getState() == FAIL) {
                errmsg = "timed out waiting for commit";
                return;
            }

            timing.done(5);
            MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep5);
        }

        setState(DONE);
        conn.done();
    }

    void status(BSONObjBuilder& b) {
        std::lock_guard<std::mutex> sl(_mutex);

        b.appendBool("active", _active);

        b.append("ns", _ns);
        b.append("from", _from);
        b.append("min", _min);
        b.append("max", _max);
        b.append("shardKeyPattern", _shardKeyPattern);

        b.append("state", stateToString(_state));
        if (_state == FAIL) {
            b.append("errmsg", _errmsg);
        }

        BSONObjBuilder bb(b.subobjStart("counts"));
        bb.append("cloned", _numCloned);
        bb.append("clonedBytes", _clonedBytes);
        bb.append("catchup", _numCatchup);
        bb.append("steady", _numSteady);
        bb.done();
    }

    bool apply(OperationContext* txn,
               const string& ns,
               BSONObj min,
               BSONObj max,
               BSONObj shardKeyPattern,
               const BSONObj& xfer,
               repl::OpTime* lastOpApplied) {
        repl::OpTime dummy;
        if (lastOpApplied == NULL) {
            lastOpApplied = &dummy;
        }

        bool didAnything = false;

        if (xfer["deleted"].isABSONObj()) {
            ScopedTransaction transaction(txn, MODE_IX);
            Lock::DBLock dlk(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
            Helpers::RemoveSaver rs("moveChunk", ns, "removedDuring");

            BSONObjIterator i(xfer["deleted"].Obj());
            while (i.more()) {
                Lock::CollectionLock clk(txn->lockState(), ns, MODE_X);
                OldClientContext ctx(txn, ns);

                BSONObj id = i.next().Obj();

                // do not apply deletes if they do not belong to the chunk being migrated
                BSONObj fullObj;
                if (Helpers::findById(txn, ctx.db(), ns.c_str(), id, fullObj)) {
                    if (!isInRange(fullObj, min, max, shardKeyPattern)) {
                        log() << "not applying out of range deletion: " << fullObj << migrateLog;

                        continue;
                    }
                }

                if (serverGlobalParams.moveParanoia) {
                    rs.goingToDelete(fullObj);
                }

                deleteObjects(txn,
                              ctx.db(),
                              ns,
                              id,
                              PlanExecutor::YIELD_MANUAL,
                              true /* justOne */,
                              false /* god */,
                              true /* fromMigrate */);

                *lastOpApplied = repl::ReplClientInfo::forClient(txn->getClient()).getLastOp();
                didAnything = true;
            }
        }

        if (xfer["reload"].isABSONObj()) {
            BSONObjIterator i(xfer["reload"].Obj());
            while (i.more()) {
                OldClientWriteContext cx(txn, ns);

                BSONObj updatedDoc = i.next().Obj();

                BSONObj localDoc;
                if (willOverrideLocalId(
                        txn, ns, min, max, shardKeyPattern, cx.db(), updatedDoc, &localDoc)) {
                    string errMsg = str::stream()
                        << "cannot migrate chunk, local document " << localDoc
                        << " has same _id as reloaded remote document " << updatedDoc;

                    warning() << errMsg;

                    // Exception will abort migration cleanly
                    uasserted(16977, errMsg);
                }

                // We are in write lock here, so sure we aren't killing
                Helpers::upsert(txn, ns, updatedDoc, true);

                *lastOpApplied = repl::ReplClientInfo::forClient(txn->getClient()).getLastOp();
                didAnything = true;
            }
        }

        return didAnything;
    }

    /**
     * Checks if an upsert of a remote document will override a local document with the same _id
     * but in a different range on this shard.
     * Must be in WriteContext to avoid races and DBHelper errors.
     * TODO: Could optimize this check out if sharding on _id.
     */
    bool willOverrideLocalId(OperationContext* txn,
                             const string& ns,
                             BSONObj min,
                             BSONObj max,
                             BSONObj shardKeyPattern,
                             Database* db,
                             BSONObj remoteDoc,
                             BSONObj* localDoc) {
        *localDoc = BSONObj();
        if (Helpers::findById(txn, db, ns.c_str(), remoteDoc, *localDoc)) {
            return !isInRange(*localDoc, min, max, shardKeyPattern);
        }

        return false;
    }

    /**
     * Returns true if the majority of the nodes and the nodes corresponding to the given
     * writeConcern (if not empty) have applied till the specified lastOp.
     */
    bool opReplicatedEnough(OperationContext* txn,
                            const repl::OpTime& lastOpApplied,
                            const WriteConcernOptions& writeConcern) {
        WriteConcernOptions majorityWriteConcern;
        majorityWriteConcern.wTimeout = -1;
        majorityWriteConcern.wMode = WriteConcernOptions::kMajority;
        Status majorityStatus = repl::getGlobalReplicationCoordinator()
                                    ->awaitReplication(txn, lastOpApplied, majorityWriteConcern)
                                    .status;

        if (!writeConcern.shouldWaitForOtherNodes()) {
            return majorityStatus.isOK();
        }

        // Also enforce the user specified write concern after "majority" so it covers
        // the union of the 2 write concerns.

        WriteConcernOptions userWriteConcern(writeConcern);
        userWriteConcern.wTimeout = -1;
        Status userStatus = repl::getGlobalReplicationCoordinator()
                                ->awaitReplication(txn, lastOpApplied, userWriteConcern)
                                .status;

        return majorityStatus.isOK() && userStatus.isOK();
    }

    bool flushPendingWrites(OperationContext* txn,
                            const std::string& ns,
                            BSONObj min,
                            BSONObj max,
                            const repl::OpTime& lastOpApplied,
                            const WriteConcernOptions& writeConcern) {
        if (!opReplicatedEnough(txn, lastOpApplied, writeConcern)) {
            repl::OpTime op(lastOpApplied);
            OCCASIONALLY warning() << "migrate commit waiting for a majority of slaves for '" << ns
                                   << "' " << min << " -> " << max << " waiting for: " << op
                                   << migrateLog;
            return false;
        }

        log() << "migrate commit succeeded flushing to secondaries for '" << ns << "' " << min
              << " -> " << max << migrateLog;

        {
            // Get global lock to wait for write to be commited to journal.
            ScopedTransaction transaction(txn, MODE_S);
            Lock::GlobalRead lk(txn->lockState());

            // if durability is on, force a write to journal
            if (getDur().commitNow(txn)) {
                log() << "migrate commit flushed to journal for '" << ns << "' " << min << " -> "
                      << max << migrateLog;
            }
        }

        return true;
    }

    static string stateToString(State state) {
        switch (state) {
            case READY:
                return "ready";
            case CLONE:
                return "clone";
            case CATCHUP:
                return "catchup";
            case STEADY:
                return "steady";
            case COMMIT_START:
                return "commitStart";
            case DONE:
                return "done";
            case FAIL:
                return "fail";
            case ABORT:
                return "abort";
        }
        verify(0);
        return "";
    }

    bool startCommit() {
        std::unique_lock<std::mutex> lock(_mutex);

        if (_state != STEADY) {
            return false;
        }

        const auto deadline = system_clock::now() + seconds(30);
        _state = COMMIT_START;
        while (_active) {
            if (std::cv_status::timeout == isActiveCV.wait_until(lock, deadline)) {
                _state = FAIL;
                log() << "startCommit never finished!" << migrateLog;
                return false;
            }
        }

        if (_state == DONE) {
            return true;
        }

        log() << "startCommit failed, final data failed to transfer" << migrateLog;
        return false;
    }

    void abort() {
        std::lock_guard<std::mutex> sl(_mutex);
        _state = ABORT;
        _errmsg = "aborted";
    }

    bool getActive() const {
        std::lock_guard<std::mutex> lk(_mutex);
        return _active;
    }
    void setActive(bool b) {
        std::lock_guard<std::mutex> lk(_mutex);
        _active = b;
        isActiveCV.notify_all();
    }

    // Guards all fields.
    mutable std::mutex _mutex;
    bool _active;
    std::condition_variable isActiveCV;

    std::string _ns;
    std::string _from;
    BSONObj _min;
    BSONObj _max;
    BSONObj _shardKeyPattern;

    long long _numCloned;
    long long _clonedBytes;
    long long _numCatchup;
    long long _numSteady;

    State _state;
    std::string _errmsg;

} migrateStatus;

void migrateThread(std::string ns,
                   BSONObj min,
                   BSONObj max,
                   BSONObj shardKeyPattern,
                   std::string fromShard,
                   OID epoch,
                   WriteConcernOptions writeConcern) {
    Client::initThread("migrateThread");
    OperationContextImpl txn;
    if (getGlobalAuthorizationManager()->isAuthEnabled()) {
        ShardedConnectionInfo::addHook();
        AuthorizationSession::get(txn.getClient())->grantInternalAuthorization();
    }

    migrateStatus.go(&txn, ns, min, max, shardKeyPattern, fromShard, epoch, writeConcern);
}

/**
 * Command for initiating the recipient side of the migration to start copying data
 * from the donor shard.
 *
 * {
 *   _recvChunkStart: "namespace",
 *   congfigServer: "hostAndPort",
 *   from: "hostAndPort",
 *   fromShardName: "shardName",
 *   toShardName: "shardName",
 *   min: {},
 *   max: {},
 *   shardKeyPattern: {},
 *
 *   // optional
 *   secondaryThrottle: bool, // defaults to true
 *   writeConcern: {} // applies to individual writes.
 * }
 */
class RecvChunkStartCommand : public ChunkCommandHelper {
public:
    void help(std::stringstream& h) const {
        h << "internal";
    }
    RecvChunkStartCommand() : ChunkCommandHelper("_recvChunkStart") {}

    virtual bool isWriteCommandForConfigServer() const {
        return false;
    }
    virtual void addRequiredPrivileges(const std::string& dbname,
                                       const BSONObj& cmdObj,
                                       std::vector<Privilege>* out) {
        ActionSet actions;
        actions.addAction(ActionType::internal);
        out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
    }
    bool run(OperationContext* txn,
             const string&,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        // Active state of TO-side migrations (MigrateStatus) is serialized by distributed
        // collection lock.
        if (migrateStatus.getActive()) {
            errmsg = "migrate already in progress";
            return false;
        }

        // Pending deletes (for migrations) are serialized by the distributed collection lock,
        // we are sure we registered a delete for a range *before* we can migrate-in a
        // subrange.
        const size_t numDeletes = getDeleter()->getTotalDeletes();
        if (numDeletes > 0) {
            errmsg = str::stream() << "can't accept new chunks because "
                                   << " there are still " << numDeletes
                                   << " deletes from previous migration";

            warning() << errmsg;
            return false;
        }

        if (!shardingState.enabled()) {
            if (!cmdObj["configServer"].eoo()) {
                dassert(cmdObj["configServer"].type() == String);
                ShardingState::initialize(cmdObj["configServer"].String());
            } else {
                errmsg = str::stream()
                    << "cannot start recv'ing chunk, "
                    << "sharding is not enabled and no config server was provided";

                warning() << errmsg;
                return false;
            }
        }

        if (!cmdObj["toShardName"].eoo()) {
            dassert(cmdObj["toShardName"].type() == String);
            shardingState.gotShardName(cmdObj["toShardName"].String());
        }

        string ns = cmdObj.firstElement().String();
        BSONObj min = cmdObj["min"].Obj().getOwned();
        BSONObj max = cmdObj["max"].Obj().getOwned();

        // Refresh our collection manager from the config server, we need a collection manager
        // to start registering pending chunks.
        // We force the remote refresh here to make the behavior consistent and predictable,
        // generally we'd refresh anyway, and to be paranoid.
        ChunkVersion currentVersion;
        Status status = shardingState.refreshMetadataNow(txn, ns, &currentVersion);

        if (!status.isOK()) {
            errmsg = str::stream() << "cannot start recv'ing chunk "
                                   << "[" << min << "," << max << ")" << causedBy(status.reason());

            warning() << errmsg;
            return false;
        }

        // Process secondary throttle settings and assign defaults if necessary.
        WriteConcernOptions writeConcern;
        status = writeConcern.parseSecondaryThrottle(cmdObj, NULL);

        if (!status.isOK()) {
            if (status.code() != ErrorCodes::WriteConcernNotDefined) {
                warning() << status.toString();
                return appendCommandStatus(result, status);
            }

            writeConcern = getDefaultWriteConcern();
        } else {
            repl::ReplicationCoordinator* replCoordinator = repl::getGlobalReplicationCoordinator();

            if (replCoordinator->getReplicationMode() ==
                    repl::ReplicationCoordinator::modeMasterSlave &&
                writeConcern.shouldWaitForOtherNodes()) {
                warning() << "recvChunk cannot check if secondary throttle setting "
                          << writeConcern.toBSON()
                          << " can be enforced in a master slave configuration";
            }

            Status status = replCoordinator->checkIfWriteConcernCanBeSatisfied(writeConcern);
            if (!status.isOK() && status != ErrorCodes::NoReplicationEnabled) {
                warning() << status.toString();
                return appendCommandStatus(result, status);
            }
        }

        if (writeConcern.shouldWaitForOtherNodes() &&
            writeConcern.wTimeout == WriteConcernOptions::kNoTimeout) {
            // Don't allow no timeout.
            writeConcern.wTimeout = kDefaultWTimeoutMs;
        }

        BSONObj shardKeyPattern;
        if (cmdObj.hasField("shardKeyPattern")) {
            shardKeyPattern = cmdObj["shardKeyPattern"].Obj().getOwned();
        } else {
            // shardKeyPattern may not be provided if another shard is from pre 2.2
            // In that case, assume the shard key pattern is the same as the range
            // specifiers provided.
            BSONObj keya = Helpers::inferKeyPattern(min);
            BSONObj keyb = Helpers::inferKeyPattern(max);
            verify(keya == keyb);

            warning()
                << "No shard key pattern provided by source shard for migration."
                   " This is likely because the source shard is running a version prior to 2.2."
                   " Falling back to assuming the shard key matches the pattern of the min and max"
                   " chunk range specifiers.  Inferred shard key: " << keya;

            shardKeyPattern = keya.getOwned();
        }

        const string fromShard(cmdObj["from"].String());

        // Set the TO-side migration to active
        Status prepareStatus = migrateStatus.prepare(ns, fromShard, min, max, shardKeyPattern);

        if (!prepareStatus.isOK()) {
            return appendCommandStatus(result, prepareStatus);
        }

        std::thread m(migrateThread,
                      ns,
                      min,
                      max,
                      shardKeyPattern,
                      fromShard,
                      currentVersion.epoch(),
                      writeConcern);

        m.detach();
        result.appendBool("started", true);
        return true;
    }

} recvChunkStartCmd;

class RecvChunkStatusCommand : public ChunkCommandHelper {
public:
    void help(std::stringstream& h) const {
        h << "internal";
    }
    RecvChunkStatusCommand() : ChunkCommandHelper("_recvChunkStatus") {}
    virtual void addRequiredPrivileges(const std::string& dbname,
                                       const BSONObj& cmdObj,
                                       std::vector<Privilege>* out) {
        ActionSet actions;
        actions.addAction(ActionType::internal);
        out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
    }
    bool run(OperationContext* txn,
             const string&,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        migrateStatus.status(result);
        return 1;
    }

} recvChunkStatusCommand;

class RecvChunkCommitCommand : public ChunkCommandHelper {
public:
    void help(std::stringstream& h) const {
        h << "internal";
    }
    RecvChunkCommitCommand() : ChunkCommandHelper("_recvChunkCommit") {}
    virtual void addRequiredPrivileges(const std::string& dbname,
                                       const BSONObj& cmdObj,
                                       std::vector<Privilege>* out) {
        ActionSet actions;
        actions.addAction(ActionType::internal);
        out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
    }
    bool run(OperationContext* txn,
             const string&,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        bool ok = migrateStatus.startCommit();
        migrateStatus.status(result);
        return ok;
    }

} recvChunkCommitCommand;

class RecvChunkAbortCommand : public ChunkCommandHelper {
public:
    void help(std::stringstream& h) const {
        h << "internal";
    }
    RecvChunkAbortCommand() : ChunkCommandHelper("_recvChunkAbort") {}
    virtual void addRequiredPrivileges(const std::string& dbname,
                                       const BSONObj& cmdObj,
                                       std::vector<Privilege>* out) {
        ActionSet actions;
        actions.addAction(ActionType::internal);
        out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
    }
    bool run(OperationContext* txn,
             const string&,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        migrateStatus.abort();
        migrateStatus.status(result);
        return true;
    }

} recvChunkAboortCommand;


class IsInRangeTest : public StartupTest {
public:
    void run() {
        BSONObj min = BSON("x" << 1);
        BSONObj max = BSON("x" << 5);
        BSONObj skey = BSON("x" << 1);

        verify(!isInRange(BSON("x" << 0), min, max, skey));
        verify(isInRange(BSON("x" << 1), min, max, skey));
        verify(isInRange(BSON("x" << 3), min, max, skey));
        verify(isInRange(BSON("x" << 4), min, max, skey));
        verify(!isInRange(BSON("x" << 5), min, max, skey));
        verify(!isInRange(BSON("x" << 6), min, max, skey));

        BSONObj obj = BSON("n" << 3);
        BSONObj min2 = BSON("x" << BSONElementHasher::hash64(obj.firstElement(), 0) - 2);
        BSONObj max2 = BSON("x" << BSONElementHasher::hash64(obj.firstElement(), 0) + 2);
        BSONObj hashedKey = BSON("x"
                                 << "hashed");

        verify(isInRange(BSON("x" << 3), min2, max2, hashedKey));
        verify(!isInRange(BSON("x" << 3), min, max, hashedKey));
        verify(!isInRange(BSON("x" << 4), min2, max2, hashedKey));

        LOG(1) << "isInRangeTest passed" << migrateLog;
    }
} isInRangeTest;
}