Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

1 files changed, 121 insertions, 377 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 910c00aaf0d..547080ff4ea 100644
--- a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
+++ b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
@@ -33,26 +33,19 @@
 
 #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"
@@ -75,8 +68,6 @@
 namespace mongo {
 namespace {
 
-using namespace fmt::literals;
-
 const char kRecvChunkStatus[] = "_recvChunkStatus";
 const char kRecvChunkCommit[] = "_recvChunkCommit";
 const char kRecvChunkAbort[] = "_recvChunkAbort";
@@ -85,24 +76,13 @@ 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;
-}
-
-/**
- * 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);
+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;
 }
 
 BSONObj createRequestWithSessionId(StringData commandName,
@@ -116,8 +96,9 @@ BSONObj createRequestWithSessionId(StringData commandName,
     return builder.obj();
 }
 
-BSONObj getDocumentKeyFromReplOperation(repl::ReplOperation replOperation) {
-    switch (replOperation.getOpType()) {
+BSONObj getDocumentKeyFromReplOperation(repl::ReplOperation replOperation,
+                                        repl::OpTypeEnum opType) {
+    switch (opType) {
         case repl::OpTypeEnum::kInsert:
         case repl::OpTypeEnum::kDelete:
             return replOperation.getObject();
@@ -145,29 +126,36 @@ char getOpCharForCrudOpType(repl::OpTypeEnum opType) {
 
 }  // namespace
 
-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));
+/**
+ * Used to commit work for LogOpForSharding. Used to keep track of changes in documents that are
+ * part of a chunk being migrated.
+ */
+class LogOpForShardingHandler final : public RecoveryUnit::Change {
+public:
+    /**
+     * Invariant: idObj should belong to a document that is part of the active chunk being migrated
+     */
+    LogOpForShardingHandler(MigrationChunkClonerSourceLegacy* cloner,
+                            const BSONObj& idObj,
+                            const char op,
+                            const repl::OpTime& opTime)
+        : _cloner(cloner), _idObj(idObj.getOwned()), _op(op), _opTime(opTime) {}
+
+    void commit(boost::optional<Timestamp>) override {
+        _cloner->_addToTransferModsQueue(_idObj, _op, _opTime);
+        _cloner->_decrementOutstandingOperationTrackRequests();
+    }
+
+    void rollback() override {
+        _cloner->_decrementOutstandingOperationTrackRequests();
     }
-}
 
-LogTransactionOperationsForShardingHandler::LogTransactionOperationsForShardingHandler(
-    LogicalSessionId lsid,
-    const std::vector<repl::ReplOperation>& stmts,
-    repl::OpTime prepareOrCommitOpTime)
-    : _lsid(std::move(lsid)),
-      _stmts(stmts),
-      _prepareOrCommitOpTime(std::move(prepareOrCommitOpTime)) {}
+private:
+    MigrationChunkClonerSourceLegacy* const _cloner;
+    const BSONObj _idObj;
+    const char _op;
+    const repl::OpTime _opTime;
+};
 
 void LogTransactionOperationsForShardingHandler::commit(boost::optional<Timestamp>) {
     std::set<NamespaceString> namespacesTouchedByTransaction;
@@ -226,45 +214,39 @@ void LogTransactionOperationsForShardingHandler::commit(boost::optional<Timestam
             continue;
         }
 
-        auto preImageDocKey = getDocumentKeyFromReplOperation(stmt);
+        auto documentKey = getDocumentKeyFromReplOperation(stmt, opType);
 
-        auto idElement = preImageDocKey["_id"];
+        auto idElement = documentKey["_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(preImageDocKey));
+                          "documentKey"_attr = redact(documentKey));
             continue;
         }
 
-        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;
-                }
+        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"];
             } else {
-                // 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);
+                continue;
             }
-        } else {
-            cloner->_addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {});
         }
 
         addToSessionMigrationOptimeQueueIfNeeded(cloner, nss, _prepareOrCommitOpTime);
+
+        cloner->_addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {});
     }
 }
 
@@ -397,7 +379,7 @@ StatusWith<BSONObj> MigrationChunkClonerSourceLegacy::commitClone(OperationConte
             }
         } else {
             invariant(PlanExecutor::IS_EOF == _jumboChunkCloneState->clonerState);
-            invariant(!_cloneList.hasMore());
+            invariant(_cloneLocs.empty());
         }
     }
 
@@ -460,6 +442,10 @@ void MigrationChunkClonerSourceLegacy::cancelClone(OperationContext* opCtx) noex
     }
 }
 
+bool MigrationChunkClonerSourceLegacy::isDocumentInMigratingChunk(const BSONObj& doc) {
+    return isInRange(doc, getMin(), getMax(), _shardKeyPattern);
+}
+
 void MigrationChunkClonerSourceLegacy::onInsertOp(OperationContext* opCtx,
                                                   const BSONObj& insertedDoc,
                                                   const repl::OpTime& opTime) {
@@ -476,7 +462,7 @@ void MigrationChunkClonerSourceLegacy::onInsertOp(OperationContext* opCtx,
         return;
     }
 
-    if (!isDocInRange(insertedDoc, getMin(), getMax(), _shardKeyPattern)) {
+    if (!isInRange(insertedDoc, getMin(), getMax(), _shardKeyPattern)) {
         return;
     }
 
@@ -484,8 +470,13 @@ void MigrationChunkClonerSourceLegacy::onInsertOp(OperationContext* opCtx,
         return;
     }
 
-    _addToTransferModsQueue(idElement.wrap(), 'i', opCtx->getTxnNumber() ? opTime : repl::OpTime());
-    _decrementOutstandingOperationTrackRequests();
+    if (opCtx->getTxnNumber()) {
+        opCtx->recoveryUnit()->registerChange(
+            std::make_unique<LogOpForShardingHandler>(this, idElement.wrap(), 'i', opTime));
+    } else {
+        opCtx->recoveryUnit()->registerChange(
+            std::make_unique<LogOpForShardingHandler>(this, idElement.wrap(), 'i', repl::OpTime()));
+    }
 }
 
 void MigrationChunkClonerSourceLegacy::onUpdateOp(OperationContext* opCtx,
@@ -506,16 +497,13 @@ void MigrationChunkClonerSourceLegacy::onUpdateOp(OperationContext* opCtx,
         return;
     }
 
-    if (!isDocInRange(postImageDoc, getMin(), getMax(), _shardKeyPattern)) {
+    if (!isInRange(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 && isDocInRange(*preImageDoc, getMin(), getMax(), _shardKeyPattern)) {
-            onDeleteOp(opCtx,
-                       repl::getDocumentKey(_shardKeyPattern, *preImageDoc),
-                       opTime,
-                       prePostImageOpTime);
+        if (preImageDoc && isInRange(*preImageDoc, getMin(), getMax(), _shardKeyPattern)) {
+            onDeleteOp(opCtx, *preImageDoc, opTime, prePostImageOpTime);
         }
         return;
     }
@@ -524,40 +512,29 @@ void MigrationChunkClonerSourceLegacy::onUpdateOp(OperationContext* opCtx,
         return;
     }
 
-    _addToTransferModsQueue(idElement.wrap(), 'u', opCtx->getTxnNumber() ? opTime : repl::OpTime());
-    _decrementOutstandingOperationTrackRequests();
+    if (opCtx->getTxnNumber()) {
+        opCtx->recoveryUnit()->registerChange(
+            std::make_unique<LogOpForShardingHandler>(this, idElement.wrap(), 'u', opTime));
+    } else {
+        opCtx->recoveryUnit()->registerChange(
+            std::make_unique<LogOpForShardingHandler>(this, idElement.wrap(), 'u', repl::OpTime()));
+    }
 }
 
 void MigrationChunkClonerSourceLegacy::onDeleteOp(OperationContext* opCtx,
-                                                  const repl::DocumentKey& documentKey,
+                                                  const BSONObj& deletedDocId,
                                                   const repl::OpTime& opTime,
                                                   const repl::OpTime&) {
     dassert(opCtx->lockState()->isCollectionLockedForMode(nss(), MODE_IX));
 
-    const auto shardKeyAndId = documentKey.getShardKeyAndId();
-
-    BSONElement idElement = documentKey.getId()["_id"];
+    BSONElement idElement = deletedDocId["_id"];
     if (idElement.eoo()) {
         LOGV2_WARNING(
             21997,
             "logDeleteOp received a document without an _id field, ignoring deleted doc: "
-            "{shardKeyAndId}",
+            "{deletedDocId}",
             "logDeleteOp received a document without an _id field and will ignore that document",
-            "deletedDocShardKeyAndId"_attr = redact(shardKeyAndId));
-        return;
-    }
-
-    if (!documentKey.getShardKey()) {
-        LOGV2_WARNING(8023600,
-                      "logDeleteOp received a document without the shard key field and will ignore "
-                      "that document",
-                      "deletedDocShardKeyAndId"_attr = redact(shardKeyAndId));
-        return;
-    }
-
-    const auto shardKeyValue =
-        _shardKeyPattern.extractShardKeyFromDocumentKey(*documentKey.getShardKey());
-    if (!isShardKeyValueInRange(shardKeyValue, getMin(), getMax())) {
+            "deletedDocId"_attr = redact(deletedDocId));
         return;
     }
 
@@ -565,9 +542,13 @@ void MigrationChunkClonerSourceLegacy::onDeleteOp(OperationContext* opCtx,
         return;
     }
 
-    _addToTransferModsQueue(
-        documentKey.getId(), 'd', opCtx->getTxnNumber() ? opTime : repl::OpTime());
-    _decrementOutstandingOperationTrackRequests();
+    if (opCtx->getTxnNumber()) {
+        opCtx->recoveryUnit()->registerChange(
+            std::make_unique<LogOpForShardingHandler>(this, idElement.wrap(), 'd', opTime));
+    } else {
+        opCtx->recoveryUnit()->registerChange(
+            std::make_unique<LogOpForShardingHandler>(this, idElement.wrap(), 'd', repl::OpTime()));
+    }
 }
 
 void MigrationChunkClonerSourceLegacy::_addToSessionMigrationOptimeQueue(
@@ -683,7 +664,6 @@ 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");
@@ -705,57 +685,38 @@ void MigrationChunkClonerSourceLegacy::_nextCloneBatchFromCloneLocs(OperationCon
                            internalQueryExecYieldIterations.load(),
                            Milliseconds(internalQueryExecYieldPeriodMS.load()));
 
-    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;
-        }
+    stdx::unique_lock<Latch> lk(_mutex);
+    auto iter = _cloneLocs.begin();
 
+    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;
         }
 
-        // 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++;
+        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;
             }
-            continue;
-        }
 
-        // 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;
+            arrBuilder->append(doc.value());
+            ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1);
         }
 
-        {
-            stdx::lock_guard lk(_mutex);
-            _numRecordsCloned++;
-        }
-        arrBuilder->append(doc->value());
-        ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1);
-        ShardingStatistics::get(opCtx).countBytesClonedOnDonor.addAndFetch(doc->value().objsize());
+        lk.lock();
     }
+
+    _cloneLocs.erase(_cloneLocs.begin(), iter);
 }
 
 uint64_t MigrationChunkClonerSourceLegacy::getCloneBatchBufferAllocationSize() {
@@ -764,7 +725,7 @@ uint64_t MigrationChunkClonerSourceLegacy::getCloneBatchBufferAllocationSize() {
         return static_cast<uint64_t>(BSONObjMaxUserSize);
 
     return std::min(static_cast<uint64_t>(BSONObjMaxUserSize),
-                    _averageObjectSizeForCloneLocs * _cloneList.size());
+                    _averageObjectSizeForCloneLocs * _cloneLocs.size());
 }
 
 Status MigrationChunkClonerSourceLegacy::nextCloneBatch(OperationContext* opCtx,
@@ -787,91 +748,18 @@ 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(!_cloneList.hasMore());
+        invariant(_cloneLocs.empty());
 
         // 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
@@ -883,11 +771,6 @@ 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) {
@@ -914,7 +797,6 @@ Status MigrationChunkClonerSourceLegacy::nextModsBatch(OperationContext* opCtx,
     _untransferredDeletesCounter = _deleted.size();
     _reload.splice(_reload.cbegin(), updateList);
     _untransferredUpsertsCounter = _reload.size();
-    _deferredUntransferredOpsCounter = _deferredReloadOrDeletePreImageDocKeys.size();
 
     return Status::OK();
 }
@@ -929,8 +811,6 @@ void MigrationChunkClonerSourceLegacy::_cleanup() {
     _untransferredUpsertsCounter = 0;
     _deleted.clear();
     _untransferredDeletesCounter = 0;
-    _deferredReloadOrDeletePreImageDocKeys.clear();
-    _deferredUntransferredOpsCounter = 0;
 }
 
 StatusWith<BSONObj> MigrationChunkClonerSourceLegacy::_callRecipient(OperationContext* opCtx,
@@ -1058,8 +938,6 @@ 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()) {
@@ -1067,20 +945,19 @@ Status MigrationChunkClonerSourceLegacy::_storeCurrentLocs(OperationContext* opC
             }
 
             if (!isLargeChunk) {
-                recordIdSet.insert(recordId);
+                stdx::lock_guard<Latch> lk(_mutex);
+                _cloneLocs.insert(recordId);
             }
 
             if (++recCount > maxRecsWhenFull) {
                 isLargeChunk = true;
 
                 if (_forceJumbo) {
-                    recordIdSet.clear();
+                    _cloneLocs.clear();
                     break;
                 }
             }
         }
-
-        _cloneList.populateList(std::move(recordIdSet));
     } catch (DBException& exception) {
         exception.addContext("Executor error while scanning for documents belonging to chunk");
         throw;
@@ -1178,9 +1055,9 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC
 
         stdx::lock_guard<Latch> sl(_mutex);
 
+        const std::size_t cloneLocsRemaining = _cloneLocs.size();
         int64_t untransferredModsSizeBytes = _untransferredDeletesCounter * _averageObjectIdSize +
-            (_untransferredUpsertsCounter + _deferredUntransferredOpsCounter) *
-                _averageObjectSizeForCloneLocs;
+            _untransferredUpsertsCounter * _averageObjectSizeForCloneLocs;
 
         if (_forceJumbo && _jumboChunkCloneState) {
             LOGV2(21992,
@@ -1200,14 +1077,13 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC
                   "moveChunk data transfer progress",
                   "response"_attr = redact(res),
                   "memoryUsedBytes"_attr = _memoryUsed,
-                  "docsRemainingToClone"_attr =
-                      _cloneList.size() - _numRecordsCloned - _numRecordsPassedOver,
+                  "docsRemainingToClone"_attr = cloneLocsRemaining,
                   "untransferredModsSizeBytes"_attr = untransferredModsSizeBytes);
         }
 
         if (res["state"].String() == "steady" && sessionCatalogSourceInCatchupPhase &&
             estimateUntransferredSessionsSize == 0) {
-            if (_cloneList.hasMore() ||
+            if (cloneLocsRemaining != 0 ||
                 (_jumboChunkCloneState && _forceJumbo &&
                  PlanExecutor::IS_EOF != _jumboChunkCloneState->clonerState)) {
                 return {ErrorCodes::OperationIncomplete,
@@ -1246,7 +1122,6 @@ 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,
@@ -1352,135 +1227,4 @@ 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