Update upstream source from tag 'upstream/6.0.10'debian/6.0.10-1

Update to upstream version '6.0.10'
with Debian dir 2d176fa254eee97b139f712fec5709641335a8c3

1 files changed, 350 insertions, 64 deletions

diff --git a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
index 547080ff4ea..89d8168a295 100644
--- a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
+++ b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
@@ -33,19 +33,26 @@
 
 #include "mongo/db/s/migration_chunk_cloner_source_legacy.h"
 
+#include <fmt/format.h>
+
 #include "mongo/base/status.h"
+#include "mongo/bson/bsonobj.h"
 #include "mongo/client/read_preference.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog_raii.h"
+#include "mongo/db/db_raii.h"
+#include "mongo/db/dbdirectclient.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/exec/working_set_common.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/ops/write_ops_retryability.h"
+#include "mongo/db/query/get_executor.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/replication_process.h"
 #include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/migration_source_manager.h"
+#include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/db/s/sharding_runtime_d_params_gen.h"
 #include "mongo/db/s/sharding_statistics.h"
@@ -68,6 +75,8 @@
 namespace mongo {
 namespace {
 
+using namespace fmt::literals;
+
 const char kRecvChunkStatus[] = "_recvChunkStatus";
 const char kRecvChunkCommit[] = "_recvChunkCommit";
 const char kRecvChunkAbort[] = "_recvChunkAbort";
@@ -76,13 +85,24 @@ const int kMaxObjectPerChunk{250000};
 const Hours kMaxWaitToCommitCloneForJumboChunk(6);
 
 MONGO_FAIL_POINT_DEFINE(failTooMuchMemoryUsed);
+MONGO_FAIL_POINT_DEFINE(hangAfterProcessingDeferredXferMods);
+
+/**
+ * Returns true if the given BSON object in the shard key value pair format is within the given
+ * range.
+ */
+bool isShardKeyValueInRange(const BSONObj& shardKeyValue, const BSONObj& min, const BSONObj& max) {
+    return shardKeyValue.woCompare(min) >= 0 && shardKeyValue.woCompare(max) < 0;
+}
 
-bool isInRange(const BSONObj& obj,
-               const BSONObj& min,
-               const BSONObj& max,
-               const ShardKeyPattern& shardKeyPattern) {
-    BSONObj k = shardKeyPattern.extractShardKeyFromDoc(obj);
-    return k.woCompare(min) >= 0 && k.woCompare(max) < 0;
+/**
+ * Returns true if the given BSON document is within the given chunk range.
+ */
+bool isDocInRange(const BSONObj& obj,
+                  const BSONObj& min,
+                  const BSONObj& max,
+                  const ShardKeyPattern& shardKeyPattern) {
+    return isShardKeyValueInRange(shardKeyPattern.extractShardKeyFromDoc(obj), min, max);
 }
 
 BSONObj createRequestWithSessionId(StringData commandName,
@@ -96,9 +116,8 @@ BSONObj createRequestWithSessionId(StringData commandName,
     return builder.obj();
 }
 
-BSONObj getDocumentKeyFromReplOperation(repl::ReplOperation replOperation,
-                                        repl::OpTypeEnum opType) {
-    switch (opType) {
+BSONObj getDocumentKeyFromReplOperation(repl::ReplOperation replOperation) {
+    switch (replOperation.getOpType()) {
         case repl::OpTypeEnum::kInsert:
         case repl::OpTypeEnum::kDelete:
             return replOperation.getObject();
@@ -157,6 +176,30 @@ private:
     const repl::OpTime _opTime;
 };
 
+LogTransactionOperationsForShardingHandler::LogTransactionOperationsForShardingHandler(
+    LogicalSessionId lsid,
+    const std::vector<repl::OplogEntry>& stmts,
+    repl::OpTime prepareOrCommitOpTime)
+    : _lsid(std::move(lsid)), _prepareOrCommitOpTime(std::move(prepareOrCommitOpTime)) {
+    _stmts.reserve(stmts.size());
+    _ownedReplBSONObj.reserve(stmts.size());
+
+    for (const auto& op : stmts) {
+        auto ownedBSON = op.getDurableReplOperation().toBSON().getOwned();
+        _ownedReplBSONObj.push_back(ownedBSON);
+        _stmts.push_back(
+            repl::ReplOperation::parse({"MigrationChunkClonerSource_toReplOperation"}, ownedBSON));
+    }
+}
+
+LogTransactionOperationsForShardingHandler::LogTransactionOperationsForShardingHandler(
+    LogicalSessionId lsid,
+    const std::vector<repl::ReplOperation>& stmts,
+    repl::OpTime prepareOrCommitOpTime)
+    : _lsid(std::move(lsid)),
+      _stmts(stmts),
+      _prepareOrCommitOpTime(std::move(prepareOrCommitOpTime)) {}
+
 void LogTransactionOperationsForShardingHandler::commit(boost::optional<Timestamp>) {
     std::set<NamespaceString> namespacesTouchedByTransaction;
 
@@ -214,39 +257,45 @@ void LogTransactionOperationsForShardingHandler::commit(boost::optional<Timestam
             continue;
         }
 
-        auto documentKey = getDocumentKeyFromReplOperation(stmt, opType);
+        auto preImageDocKey = getDocumentKeyFromReplOperation(stmt);
 
-        auto idElement = documentKey["_id"];
+        auto idElement = preImageDocKey["_id"];
         if (idElement.eoo()) {
             LOGV2_WARNING(21994,
-                          "Received a document without an _id field, ignoring: {documentKey}",
                           "Received a document without an _id and will ignore that document",
-                          "documentKey"_attr = redact(documentKey));
+                          "documentKey"_attr = redact(preImageDocKey));
             continue;
         }
 
-        auto const& minKey = cloner->_args.getMin().get();
-        auto const& maxKey = cloner->_args.getMax().get();
-        auto const& shardKeyPattern = cloner->_shardKeyPattern;
-
-        if (!isInRange(documentKey, minKey, maxKey, shardKeyPattern)) {
-            // If the preImageDoc is not in range but the postImageDoc was, we know that the
-            // document has changed shard keys and no longer belongs in the chunk being cloned.
-            // We will model the deletion of the preImage document so that the destination chunk
-            // does not receive an outdated version of this document.
-            if (opType == repl::OpTypeEnum::kUpdate &&
-                isInRange(stmt.getPreImageDocumentKey(), minKey, maxKey, shardKeyPattern) &&
-                !stmt.getPreImageDocumentKey()["_id"].eoo()) {
-                opType = repl::OpTypeEnum::kDelete;
-                idElement = stmt.getPreImageDocumentKey()["id"];
+        if (opType == repl::OpTypeEnum::kUpdate) {
+            auto const& shardKeyPattern = cloner->_shardKeyPattern;
+            auto preImageShardKeyValues =
+                shardKeyPattern.extractShardKeyFromDocumentKey(preImageDocKey);
+
+            // If prepare was performed from another term, we will not have the post image doc key
+            // since it is not persisted in the oplog.
+            auto postImageDocKey = stmt.getPostImageDocumentKey();
+            if (!postImageDocKey.isEmpty()) {
+                if (!cloner->_processUpdateForXferMod(preImageDocKey, postImageDocKey)) {
+                    // We don't need to add this op to session migration if neither post or pre
+                    // image doc falls within the chunk range.
+                    continue;
+                }
             } else {
-                continue;
+                // We can't perform reads here using the same recovery unit because the transaction
+                // is already committed. We instead defer performing the reads when xferMods command
+                // is called. Also allow this op to be added to session migration since we can't
+                // tell whether post image doc will fall within the chunk range. If it turns out
+                // both preImage and postImage doc don't fall into the chunk range, it is not wrong
+                // for this op to be added to session migration, but it will result in wasted work
+                // and unneccesary extra oplog storage on the destination.
+                cloner->_deferProcessingForXferMod(preImageDocKey);
             }
+        } else {
+            cloner->_addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {});
         }
 
         addToSessionMigrationOptimeQueueIfNeeded(cloner, nss, _prepareOrCommitOpTime);
-
-        cloner->_addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {});
     }
 }
 
@@ -379,7 +428,7 @@ StatusWith<BSONObj> MigrationChunkClonerSourceLegacy::commitClone(OperationConte
             }
         } else {
             invariant(PlanExecutor::IS_EOF == _jumboChunkCloneState->clonerState);
-            invariant(_cloneLocs.empty());
+            invariant(!_cloneList.hasMore());
         }
     }
 
@@ -443,7 +492,7 @@ void MigrationChunkClonerSourceLegacy::cancelClone(OperationContext* opCtx) noex
 }
 
 bool MigrationChunkClonerSourceLegacy::isDocumentInMigratingChunk(const BSONObj& doc) {
-    return isInRange(doc, getMin(), getMax(), _shardKeyPattern);
+    return isDocInRange(doc, getMin(), getMax(), _shardKeyPattern);
 }
 
 void MigrationChunkClonerSourceLegacy::onInsertOp(OperationContext* opCtx,
@@ -462,7 +511,7 @@ void MigrationChunkClonerSourceLegacy::onInsertOp(OperationContext* opCtx,
         return;
     }
 
-    if (!isInRange(insertedDoc, getMin(), getMax(), _shardKeyPattern)) {
+    if (!isDocInRange(insertedDoc, getMin(), getMax(), _shardKeyPattern)) {
         return;
     }
 
@@ -497,12 +546,12 @@ void MigrationChunkClonerSourceLegacy::onUpdateOp(OperationContext* opCtx,
         return;
     }
 
-    if (!isInRange(postImageDoc, getMin(), getMax(), _shardKeyPattern)) {
+    if (!isDocInRange(postImageDoc, getMin(), getMax(), _shardKeyPattern)) {
         // If the preImageDoc is not in range but the postImageDoc was, we know that the document
         // has changed shard keys and no longer belongs in the chunk being cloned. We will model
         // the deletion of the preImage document so that the destination chunk does not receive an
         // outdated version of this document.
-        if (preImageDoc && isInRange(*preImageDoc, getMin(), getMax(), _shardKeyPattern)) {
+        if (preImageDoc && isDocInRange(*preImageDoc, getMin(), getMax(), _shardKeyPattern)) {
             onDeleteOp(opCtx, *preImageDoc, opTime, prePostImageOpTime);
         }
         return;
@@ -664,6 +713,7 @@ void MigrationChunkClonerSourceLegacy::_nextCloneBatchFromIndexScan(OperationCon
             lk.unlock();
 
             ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1);
+            ShardingStatistics::get(opCtx).countBytesClonedOnDonor.addAndFetch(obj.objsize());
         }
     } catch (DBException& exception) {
         exception.addContext("Executor error while scanning for documents belonging to chunk");
@@ -685,38 +735,57 @@ void MigrationChunkClonerSourceLegacy::_nextCloneBatchFromCloneLocs(OperationCon
                            internalQueryExecYieldIterations.load(),
                            Milliseconds(internalQueryExecYieldPeriodMS.load()));
 
-    stdx::unique_lock<Latch> lk(_mutex);
-    auto iter = _cloneLocs.begin();
+    while (true) {
+        int recordsNoLongerExist = 0;
+        auto docInFlight = _cloneList.getNextDoc(opCtx, collection, &recordsNoLongerExist);
+
+        if (recordsNoLongerExist) {
+            stdx::lock_guard lk(_mutex);
+            _numRecordsPassedOver += recordsNoLongerExist;
+        }
+
+        const auto& doc = docInFlight->getDoc();
+        if (!doc) {
+            break;
+        }
 
-    for (; iter != _cloneLocs.end(); ++iter) {
         // We must always make progress in this method by at least one document because empty
         // return indicates there is no more initial clone data.
         if (arrBuilder->arrSize() && tracker.intervalHasElapsed()) {
+            _cloneList.insertOverflowDoc(*doc);
             break;
         }
 
-        auto nextRecordId = *iter;
-
-        lk.unlock();
-
-        Snapshotted<BSONObj> doc;
-        if (collection->findDoc(opCtx, nextRecordId, &doc)) {
-            // Use the builder size instead of accumulating the document sizes directly so
-            // that we take into consideration the overhead of BSONArray indices.
-            if (arrBuilder->arrSize() &&
-                (arrBuilder->len() + doc.value().objsize() + 1024) > BSONObjMaxUserSize) {
-
-                break;
+        // Do not send documents that are no longer in the chunk range being moved. This can
+        // happen when document shard key value of the document changed after the initial
+        // index scan during cloning. This is needed because the destination is very
+        // conservative in processing xferMod deletes and won't delete docs that are not in
+        // the range of the chunk being migrated.
+        if (!isDocInRange(
+                doc->value(), _args.getMin().value(), _args.getMax().value(), _shardKeyPattern)) {
+            {
+                stdx::lock_guard lk(_mutex);
+                _numRecordsPassedOver++;
             }
+            continue;
+        }
 
-            arrBuilder->append(doc.value());
-            ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1);
+        // Use the builder size instead of accumulating the document sizes directly so
+        // that we take into consideration the overhead of BSONArray indices.
+        if (arrBuilder->arrSize() &&
+            (arrBuilder->len() + doc->value().objsize() + 1024) > BSONObjMaxUserSize) {
+            _cloneList.insertOverflowDoc(*doc);
+            break;
         }
 
-        lk.lock();
+        {
+            stdx::lock_guard lk(_mutex);
+            _numRecordsCloned++;
+        }
+        arrBuilder->append(doc->value());
+        ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1);
+        ShardingStatistics::get(opCtx).countBytesClonedOnDonor.addAndFetch(doc->value().objsize());
     }
-
-    _cloneLocs.erase(_cloneLocs.begin(), iter);
 }
 
 uint64_t MigrationChunkClonerSourceLegacy::getCloneBatchBufferAllocationSize() {
@@ -725,7 +794,7 @@ uint64_t MigrationChunkClonerSourceLegacy::getCloneBatchBufferAllocationSize() {
         return static_cast<uint64_t>(BSONObjMaxUserSize);
 
     return std::min(static_cast<uint64_t>(BSONObjMaxUserSize),
-                    _averageObjectSizeForCloneLocs * _cloneLocs.size());
+                    _averageObjectSizeForCloneLocs * _cloneList.size());
 }
 
 Status MigrationChunkClonerSourceLegacy::nextCloneBatch(OperationContext* opCtx,
@@ -748,18 +817,91 @@ Status MigrationChunkClonerSourceLegacy::nextCloneBatch(OperationContext* opCtx,
     return Status::OK();
 }
 
+bool MigrationChunkClonerSourceLegacy::_processUpdateForXferMod(const BSONObj& preImageDocKey,
+                                                                const BSONObj& postImageDocKey) {
+    auto const& minKey = _args.getMin().value();
+    auto const& maxKey = _args.getMax().value();
+
+    auto postShardKeyValues = _shardKeyPattern.extractShardKeyFromDocumentKey(postImageDocKey);
+    fassert(6836100, !postShardKeyValues.isEmpty());
+
+    auto opType = repl::OpTypeEnum::kUpdate;
+    auto idElement = preImageDocKey["_id"];
+
+    if (!isShardKeyValueInRange(postShardKeyValues, minKey, maxKey)) {
+        // If the preImageDoc is not in range but the postImageDoc was, we know that the
+        // document has changed shard keys and no longer belongs in the chunk being cloned.
+        // We will model the deletion of the preImage document so that the destination chunk
+        // does not receive an outdated version of this document.
+
+        auto preImageShardKeyValues =
+            _shardKeyPattern.extractShardKeyFromDocumentKey(preImageDocKey);
+        fassert(6836101, !preImageShardKeyValues.isEmpty());
+
+        if (!isShardKeyValueInRange(preImageShardKeyValues, minKey, maxKey)) {
+            return false;
+        }
+
+        opType = repl::OpTypeEnum::kDelete;
+        idElement = postImageDocKey["_id"];
+    }
+
+    _addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {});
+
+    return true;
+}
+
+void MigrationChunkClonerSourceLegacy::_deferProcessingForXferMod(const BSONObj& preImageDocKey) {
+    stdx::lock_guard<Latch> sl(_mutex);
+    _deferredReloadOrDeletePreImageDocKeys.push_back(preImageDocKey.getOwned());
+    _deferredUntransferredOpsCounter++;
+}
+
+void MigrationChunkClonerSourceLegacy::_processDeferredXferMods(OperationContext* opCtx,
+                                                                Database* db) {
+    std::vector<BSONObj> deferredReloadOrDeletePreImageDocKeys;
+
+    {
+        stdx::unique_lock lk(_mutex);
+        deferredReloadOrDeletePreImageDocKeys.swap(_deferredReloadOrDeletePreImageDocKeys);
+    }
+
+    for (const auto& preImageDocKey : deferredReloadOrDeletePreImageDocKeys) {
+        auto idElement = preImageDocKey["_id"];
+        BSONObj newerVersionDoc;
+        if (!Helpers::findById(opCtx, db, nss().ns(), BSON("_id" << idElement), newerVersionDoc)) {
+            // If the document can no longer be found, this means that another later op must have
+            // deleted it. That delete would have been captured by the xferMods so nothing else to
+            // do here.
+            continue;
+        }
+
+        auto postImageDocKey =
+            CollectionMetadata::extractDocumentKey(&_shardKeyPattern, newerVersionDoc);
+        static_cast<void>(_processUpdateForXferMod(preImageDocKey, postImageDocKey));
+    }
+
+    hangAfterProcessingDeferredXferMods.execute([&](const auto& data) {
+        if (!deferredReloadOrDeletePreImageDocKeys.empty()) {
+            hangAfterProcessingDeferredXferMods.pauseWhileSet();
+        }
+    });
+}
+
 Status MigrationChunkClonerSourceLegacy::nextModsBatch(OperationContext* opCtx,
                                                        Database* db,
                                                        BSONObjBuilder* builder) {
     dassert(opCtx->lockState()->isCollectionLockedForMode(nss(), MODE_IS));
 
+    _processDeferredXferMods(opCtx, db);
+
     std::list<BSONObj> deleteList;
     std::list<BSONObj> updateList;
 
     {
         // All clone data must have been drained before starting to fetch the incremental changes.
         stdx::unique_lock<Latch> lk(_mutex);
-        invariant(_cloneLocs.empty());
+        invariant(!_cloneList.hasMore());
 
         // The "snapshot" for delete and update list must be taken under a single lock. This is to
         // ensure that we will preserve the causal order of writes. Always consume the delete
@@ -771,6 +913,11 @@ Status MigrationChunkClonerSourceLegacy::nextModsBatch(OperationContext* opCtx,
         updateList.splice(updateList.cbegin(), _reload);
     }
 
+    // It's important to abandon any open snapshots before processing updates so that we are sure
+    // that our snapshot is at least as new as those updates. It's possible for a stale snapshot to
+    // still be open from reads performed by _processDeferredXferMods(), above.
+    opCtx->recoveryUnit()->abandonSnapshot();
+
     StringData ns = nss().ns().c_str();
     BSONArrayBuilder arrDel(builder->subarrayStart("deleted"));
     auto noopFn = [](BSONObj idDoc, BSONObj* fullDoc) {
@@ -797,6 +944,7 @@ Status MigrationChunkClonerSourceLegacy::nextModsBatch(OperationContext* opCtx,
     _untransferredDeletesCounter = _deleted.size();
     _reload.splice(_reload.cbegin(), updateList);
     _untransferredUpsertsCounter = _reload.size();
+    _deferredUntransferredOpsCounter = _deferredReloadOrDeletePreImageDocKeys.size();
 
     return Status::OK();
 }
@@ -811,6 +959,8 @@ void MigrationChunkClonerSourceLegacy::_cleanup() {
     _untransferredUpsertsCounter = 0;
     _deleted.clear();
     _untransferredDeletesCounter = 0;
+    _deferredReloadOrDeletePreImageDocKeys.clear();
+    _deferredUntransferredOpsCounter = 0;
 }
 
 StatusWith<BSONObj> MigrationChunkClonerSourceLegacy::_callRecipient(OperationContext* opCtx,
@@ -938,6 +1088,8 @@ Status MigrationChunkClonerSourceLegacy::_storeCurrentLocs(OperationContext* opC
     try {
         BSONObj obj;
         RecordId recordId;
+        RecordIdSet recordIdSet;
+
         while (PlanExecutor::ADVANCED == exec->getNext(&obj, &recordId)) {
             Status interruptStatus = opCtx->checkForInterruptNoAssert();
             if (!interruptStatus.isOK()) {
@@ -945,19 +1097,20 @@ Status MigrationChunkClonerSourceLegacy::_storeCurrentLocs(OperationContext* opC
             }
 
             if (!isLargeChunk) {
-                stdx::lock_guard<Latch> lk(_mutex);
-                _cloneLocs.insert(recordId);
+                recordIdSet.insert(recordId);
             }
 
             if (++recCount > maxRecsWhenFull) {
                 isLargeChunk = true;
 
                 if (_forceJumbo) {
-                    _cloneLocs.clear();
+                    recordIdSet.clear();
                     break;
                 }
             }
         }
+
+        _cloneList.populateList(std::move(recordIdSet));
     } catch (DBException& exception) {
         exception.addContext("Executor error while scanning for documents belonging to chunk");
         throw;
@@ -1055,9 +1208,9 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC
 
         stdx::lock_guard<Latch> sl(_mutex);
 
-        const std::size_t cloneLocsRemaining = _cloneLocs.size();
         int64_t untransferredModsSizeBytes = _untransferredDeletesCounter * _averageObjectIdSize +
-            _untransferredUpsertsCounter * _averageObjectSizeForCloneLocs;
+            (_untransferredUpsertsCounter + _deferredUntransferredOpsCounter) *
+                _averageObjectSizeForCloneLocs;
 
         if (_forceJumbo && _jumboChunkCloneState) {
             LOGV2(21992,
@@ -1077,13 +1230,14 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC
                   "moveChunk data transfer progress",
                   "response"_attr = redact(res),
                   "memoryUsedBytes"_attr = _memoryUsed,
-                  "docsRemainingToClone"_attr = cloneLocsRemaining,
+                  "docsRemainingToClone"_attr =
+                      _cloneList.size() - _numRecordsCloned - _numRecordsPassedOver,
                   "untransferredModsSizeBytes"_attr = untransferredModsSizeBytes);
         }
 
         if (res["state"].String() == "steady" && sessionCatalogSourceInCatchupPhase &&
             estimateUntransferredSessionsSize == 0) {
-            if (cloneLocsRemaining != 0 ||
+            if (_cloneList.hasMore() ||
                 (_jumboChunkCloneState && _forceJumbo &&
                  PlanExecutor::IS_EOF != _jumboChunkCloneState->clonerState)) {
                 return {ErrorCodes::OperationIncomplete,
@@ -1122,6 +1276,7 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC
                       "moveChunk data transfer within threshold to allow write blocking",
                       "_untransferredUpsertsCounter"_attr = _untransferredUpsertsCounter,
                       "_untransferredDeletesCounter"_attr = _untransferredDeletesCounter,
+                      "_deferredUntransferredOpsCounter"_attr = _deferredUntransferredOpsCounter,
                       "_averageObjectSizeForCloneLocs"_attr = _averageObjectSizeForCloneLocs,
                       "_averageObjectIdSize"_attr = _averageObjectIdSize,
                       "untransferredModsSizeBytes"_attr = untransferredModsSizeBytes,
@@ -1227,4 +1382,135 @@ MigrationChunkClonerSourceLegacy::getNotificationForNextSessionMigrationBatch()
     return _sessionCatalogSource->getNotificationForNewOplog();
 }
 
+MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWithLock::DocumentInFlightWithLock(
+    WithLock lock, MigrationChunkClonerSourceLegacy::CloneList& clonerList)
+    : _inProgressReadToken(
+          std::make_unique<MigrationChunkClonerSourceLegacy::CloneList::InProgressReadToken>(
+              lock, clonerList)) {}
+
+void MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWithLock::setDoc(
+    boost::optional<Snapshotted<BSONObj>> doc) {
+    _doc = std::move(doc);
+}
+
+std::unique_ptr<MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock>
+MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWithLock::release() {
+    invariant(_inProgressReadToken);
+
+    return std::make_unique<
+        MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock>(
+        std::move(_inProgressReadToken), std::move(_doc));
+}
+
+MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock::
+    DocumentInFlightWhileNotInLock(
+        std::unique_ptr<CloneList::InProgressReadToken> inProgressReadToken,
+        boost::optional<Snapshotted<BSONObj>> doc)
+    : _inProgressReadToken(std::move(inProgressReadToken)), _doc(std::move(doc)) {}
+
+void MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock::setDoc(
+    boost::optional<Snapshotted<BSONObj>> doc) {
+    _doc = std::move(doc);
+}
+
+const boost::optional<Snapshotted<BSONObj>>&
+MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock::getDoc() {
+    return _doc;
+}
+
+MigrationChunkClonerSourceLegacy::CloneList::InProgressReadToken::InProgressReadToken(
+    WithLock withLock, CloneList& cloneList)
+    : _cloneList(cloneList) {
+    _cloneList._startedOneInProgressRead(withLock);
+}
+
+MigrationChunkClonerSourceLegacy::CloneList::InProgressReadToken::~InProgressReadToken() {
+    _cloneList._finishedOneInProgressRead();
+}
+
+MigrationChunkClonerSourceLegacy::CloneList::CloneList() {
+    _recordIdsIter = _recordIds.begin();
+}
+
+void MigrationChunkClonerSourceLegacy::CloneList::populateList(RecordIdSet recordIds) {
+    stdx::lock_guard lk(_mutex);
+    _recordIds = std::move(recordIds);
+    _recordIdsIter = _recordIds.begin();
+}
+
+void MigrationChunkClonerSourceLegacy::CloneList::insertOverflowDoc(Snapshotted<BSONObj> doc) {
+    stdx::lock_guard lk(_mutex);
+    invariant(_inProgressReads >= 1);
+    _overflowDocs.push_back(std::move(doc));
+}
+
+bool MigrationChunkClonerSourceLegacy::CloneList::hasMore() const {
+    stdx::lock_guard lk(_mutex);
+    return _recordIdsIter != _recordIds.cend() && _inProgressReads > 0;
+}
+
+std::unique_ptr<MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock>
+MigrationChunkClonerSourceLegacy::CloneList::getNextDoc(OperationContext* opCtx,
+                                                        const CollectionPtr& collection,
+                                                        int* numRecordsNoLongerExist) {
+    while (true) {
+        stdx::unique_lock lk(_mutex);
+        invariant(_inProgressReads >= 0);
+        RecordId nextRecordId;
+
+        opCtx->waitForConditionOrInterrupt(_moreDocsCV, lk, [&]() {
+            return _recordIdsIter != _recordIds.end() || !_overflowDocs.empty() ||
+                _inProgressReads == 0;
+        });
+
+        DocumentInFlightWithLock docInFlight(lk, *this);
+
+        // One of the following must now be true (corresponding to the three if conditions):
+        //   1.  There is a document in the overflow set
+        //   2.  The iterator has not reached the end of the record id set
+        //   3.  The overflow set is empty, the iterator is at the end, and
+        //       no threads are holding a document.  This condition indicates
+        //       that there are no more docs to return for the cloning phase.
+        if (!_overflowDocs.empty()) {
+            docInFlight.setDoc(std::move(_overflowDocs.front()));
+            _overflowDocs.pop_front();
+            return docInFlight.release();
+        } else if (_recordIdsIter != _recordIds.end()) {
+            nextRecordId = *_recordIdsIter;
+            ++_recordIdsIter;
+        } else {
+            return docInFlight.release();
+        }
+
+        lk.unlock();
+
+        auto docInFlightWhileNotLocked = docInFlight.release();
+
+        Snapshotted<BSONObj> doc;
+        if (collection->findDoc(opCtx, nextRecordId, &doc)) {
+            docInFlightWhileNotLocked->setDoc(std::move(doc));
+            return docInFlightWhileNotLocked;
+        }
+
+        if (numRecordsNoLongerExist) {
+            (*numRecordsNoLongerExist)++;
+        }
+    }
+}
+
+size_t MigrationChunkClonerSourceLegacy::CloneList::size() const {
+    stdx::unique_lock lk(_mutex);
+    return _recordIds.size();
+}
+
+void MigrationChunkClonerSourceLegacy::CloneList::_startedOneInProgressRead(WithLock) {
+    _inProgressReads++;
+}
+
+void MigrationChunkClonerSourceLegacy::CloneList::_finishedOneInProgressRead() {
+    stdx::lock_guard lk(_mutex);
+    _inProgressReads--;
+    _moreDocsCV.notify_one();
+}
+
 }  // namespace mongo