SERVER-37408 Add afterClusterTime to initial sync collection scans

(cherry picked from commit cbd0a1a3df662c54da23d5def4ccc10dd1c1f88e)

2 files changed, 34 insertions, 0 deletions

diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 4870893af20..0e35184d422 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -109,6 +109,16 @@ MONGO_INITIALIZER(InitializeParseValidationActionImpl)(InitializerContext* const
 // Used below to fail during inserts.
 MONGO_FP_DECLARE(failCollectionInserts);
 
+// Used to pause after inserting collection data and calling the opObservers.  Inserts to
+// replicated collections that are not part of a multi-statement transaction will have generated
+// their OpTime and oplog entry. Supports parameters to limit pause by namespace and by _id
+// of first data item in an insert (must be of type string):
+//  data: {
+//      collectionNS: <fully-qualified collection namespace>,
+//      first_id: <string>
+//  }
+MONGO_FP_DECLARE(hangAfterCollectionInserts);
+
 // Uses the collator factory to convert the BSON representation of a collator to a
 // CollatorInterface. Returns null if the BSONObj is empty. We expect the stored collation to be
 // valid, since it gets validated on collection create.
@@ -383,6 +393,25 @@ Status CollectionImpl::insertDocuments(OperationContext* opCtx,
 
     opCtx->recoveryUnit()->onCommit([this]() { notifyCappedWaitersIfNeeded(); });
 
+    MONGO_FAIL_POINT_BLOCK(hangAfterCollectionInserts, extraData) {
+        const BSONObj& data = extraData.getData();
+        const auto collElem = data["collectionNS"];
+        const auto firstIdElem = data["first_id"];
+        // If the failpoint specifies no collection or matches the existing one, hang.
+        if ((!collElem || _ns.ns() == collElem.str()) &&
+            (!firstIdElem || (begin != end && firstIdElem.type() == mongo::String &&
+                              begin->doc["_id"].str() == firstIdElem.str()))) {
+            string whenFirst =
+                firstIdElem ? (string(" when first _id is ") + firstIdElem.str()) : "";
+            while (MONGO_FAIL_POINT(hangAfterCollectionInserts)) {
+                log() << "hangAfterCollectionInserts fail point enabled for " << _ns.toString()
+                      << whenFirst << ". Blocking until fail point is disabled.";
+                mongo::sleepsecs(1);
+                opCtx->checkForInterrupt();
+            }
+        }
+    }
+
     return Status::OK();
 }
 
diff --git a/src/mongo/db/repl/initial_syncer.cpp b/src/mongo/db/repl/initial_syncer.cpp
index 35c2e3c36c4..85427fb5329 100644
--- a/src/mongo/db/repl/initial_syncer.cpp
+++ b/src/mongo/db/repl/initial_syncer.cpp
@@ -620,6 +620,11 @@ void InitialSyncer::_lastOplogEntryFetcherCallbackForBeginTimestamp(
     auto filterBob = BSONObjBuilder(queryBob.subobjStart("filter"));
     filterBob.append("_id", FeatureCompatibilityVersion::kParameterName);
     filterBob.done();
+    // As part of reading the FCV, we ensure the source node "all committed" timestamp has advanced
+    // to at least the timestamp of the last optime that we found in the lastOplogEntryFetcher.
+    // When document locking is used, there could be oplog "holes" which would result in
+    // inconsistent initial sync data if we didn't do this.
+    queryBob.append("readConcern", BSON("afterOpTime" << lastOpTimeWithHash.opTime));
 
     _fCVFetcher = stdx::make_unique<Fetcher>(
         _exec,