SERVER-33258: Guarantee replication tables are durable before use.

8 files changed, 46 insertions, 31 deletions

diff --git a/src/mongo/base/error_codes.err b/src/mongo/base/error_codes.err
index 8b63d5f77b7..f333f5cc016 100644
--- a/src/mongo/base/error_codes.err
+++ b/src/mongo/base/error_codes.err
@@ -239,6 +239,7 @@ error_code("NotImplemented", 238)
 error_code("SnapshotTooOld", 239)
 error_code("DNSRecordTypeMismatch", 240)
 error_code("ConversionFailure", 241)
+error_code("CannotCreateCollection", 242)
 
 # Error codes 4000-8999 are reserved.
 
diff --git a/src/mongo/db/repl/replication_consistency_markers.h b/src/mongo/db/repl/replication_consistency_markers.h
index 2bf12adb0db..ce986c64500 100644
--- a/src/mongo/db/repl/replication_consistency_markers.h
+++ b/src/mongo/db/repl/replication_consistency_markers.h
@@ -203,6 +203,12 @@ public:
      */
     virtual void writeCheckpointTimestamp(OperationContext* opCtx, const Timestamp& timestamp) = 0;
     virtual Timestamp getCheckpointTimestamp(OperationContext* opCtx) = 0;
+
+    /**
+     * Create the set of collections required for steady-state replication to work. E.g: `minvalid`
+     * or `oplogTruncateAfterPoint`.
+     */
+    virtual Status createInternalCollections(OperationContext* opCtx) = 0;
 };
 
 }  // namespace repl
diff --git a/src/mongo/db/repl/replication_consistency_markers_impl.cpp b/src/mongo/db/repl/replication_consistency_markers_impl.cpp
index b35a67c78f2..490fd1234bf 100644
--- a/src/mongo/db/repl/replication_consistency_markers_impl.cpp
+++ b/src/mongo/db/repl/replication_consistency_markers_impl.cpp
@@ -116,18 +116,7 @@ void ReplicationConsistencyMarkersImpl::initializeMinValidDocument(OperationCont
     // the 'minValid' document, but we still want the initialization write to go into the next
     // checkpoint since a newly initialized 'minValid' document is always valid.
     upsert.timestamp = Timestamp();
-
-    Status status = _storageInterface->putSingleton(opCtx, _minValidNss, upsert);
-
-    // If the collection doesn't exist, create it and try again.
-    if (status == ErrorCodes::NamespaceNotFound) {
-        status = _storageInterface->createCollection(opCtx, _minValidNss, CollectionOptions());
-        fassertStatusOK(40509, status);
-
-        status = _storageInterface->putSingleton(opCtx, _minValidNss, upsert);
-    }
-
-    fassertStatusOK(40467, status);
+    fassertStatusOK(40467, _storageInterface->putSingleton(opCtx, _minValidNss, upsert));
 }
 
 bool ReplicationConsistencyMarkersImpl::getInitialSyncFlag(OperationContext* opCtx) const {
@@ -331,20 +320,10 @@ ReplicationConsistencyMarkersImpl::_getOplogTruncateAfterPointDocument(
 
 void ReplicationConsistencyMarkersImpl::_upsertOplogTruncateAfterPointDocument(
     OperationContext* opCtx, const BSONObj& updateSpec) {
-    auto status = _storageInterface->upsertById(
-        opCtx, _oplogTruncateAfterPointNss, kOplogTruncateAfterPointId["_id"], updateSpec);
-
-    // If the collection doesn't exist, creates it and tries again.
-    if (status == ErrorCodes::NamespaceNotFound) {
-        status = _storageInterface->createCollection(
-            opCtx, _oplogTruncateAfterPointNss, CollectionOptions());
-        fassertStatusOK(40511, status);
-
-        status = _storageInterface->upsertById(
-            opCtx, _oplogTruncateAfterPointNss, kOplogTruncateAfterPointId["_id"], updateSpec);
-    }
-
-    fassertStatusOK(40512, status);
+    fassertStatusOK(
+        40512,
+        _storageInterface->upsertById(
+            opCtx, _oplogTruncateAfterPointNss, kOplogTruncateAfterPointId["_id"], updateSpec));
 }
 
 void ReplicationConsistencyMarkersImpl::setOplogTruncateAfterPoint(OperationContext* opCtx,
@@ -449,6 +428,19 @@ Timestamp ReplicationConsistencyMarkersImpl::getCheckpointTimestamp(OperationCon
     return out;
 }
 
+Status ReplicationConsistencyMarkersImpl::createInternalCollections(OperationContext* opCtx) {
+    for (auto nss : std::vector<NamespaceString>(
+             {_oplogTruncateAfterPointNss, _minValidNss, _checkpointTimestampNss})) {
+        auto status = _storageInterface->createCollection(opCtx, nss, CollectionOptions());
+        if (!status.isOK() && status.code() != ErrorCodes::NamespaceExists) {
+            return {ErrorCodes::CannotCreateCollection,
+                    str::stream() << "Failed to create collection. Ns: " << nss.ns() << " Error: "
+                                  << status.toString()};
+        }
+    }
+
+    return Status::OK();
+}
 
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/replication_consistency_markers_impl.h b/src/mongo/db/repl/replication_consistency_markers_impl.h
index ea848a3a109..789d83eb3ab 100644
--- a/src/mongo/db/repl/replication_consistency_markers_impl.h
+++ b/src/mongo/db/repl/replication_consistency_markers_impl.h
@@ -83,6 +83,8 @@ public:
     void writeCheckpointTimestamp(OperationContext* opCtx, const Timestamp& timestamp);
     Timestamp getCheckpointTimestamp(OperationContext* opCtx);
 
+    Status createInternalCollections(OperationContext* opCtx);
+
 private:
     /**
      * Reads the MinValid document from disk.
@@ -91,8 +93,8 @@ private:
     boost::optional<MinValidDocument> _getMinValidDocument(OperationContext* opCtx) const;
 
     /**
-     * Updates the MinValid document according to the provided update spec. If the collection does
-     * not exist, it is created. If the document does not exist, it is upserted.
+     * Updates the MinValid document according to the provided update spec. The collection must
+     * exist, see `createInternalCollections`. If the document does not exist, it is upserted.
      *
      * This fasserts on failure.
      */
@@ -120,9 +122,8 @@ private:
         OperationContext* opCtx) const;
 
     /**
-     * Upserts the OplogTruncateAfterPoint document according to the provided update spec.
-     * If the collection does not exist, it is created. If the document does not exist,
-     * it is upserted.
+     * Upserts the OplogTruncateAfterPoint document according to the provided update spec. The
+     * collection must already exist. See `createInternalCollections`.
      *
      * This fasserts on failure.
      */
diff --git a/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp b/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp
index 729dad6becf..c45a4114ce7 100644
--- a/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp
+++ b/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp
@@ -172,6 +172,7 @@ TEST_F(ReplicationConsistencyMarkersTest, InitialSyncFlag) {
     ReplicationConsistencyMarkersImpl consistencyMarkers(
         getStorageInterface(), minValidNss, oplogTruncateAfterPointNss, checkpointTimestampNss);
     auto opCtx = getOperationContext();
+    ASSERT(consistencyMarkers.createInternalCollections(opCtx).isOK());
     consistencyMarkers.initializeMinValidDocument(opCtx);
 
     // Initial sync flag should be unset after initializing a new storage engine.
@@ -199,6 +200,7 @@ TEST_F(ReplicationConsistencyMarkersTest, GetMinValidAfterSettingInitialSyncFlag
     ReplicationConsistencyMarkersImpl consistencyMarkers(
         getStorageInterface(), minValidNss, oplogTruncateAfterPointNss, checkpointTimestampNss);
     auto opCtx = getOperationContext();
+    ASSERT(consistencyMarkers.createInternalCollections(opCtx).isOK());
     consistencyMarkers.initializeMinValidDocument(opCtx);
 
     // Initial sync flag should be unset after initializing a new storage engine.
@@ -221,6 +223,7 @@ TEST_F(ReplicationConsistencyMarkersTest, ReplicationConsistencyMarkers) {
     ReplicationConsistencyMarkersImpl consistencyMarkers(
         getStorageInterface(), minValidNss, oplogTruncateAfterPointNss, checkpointTimestampNss);
     auto opCtx = getOperationContext();
+    ASSERT(consistencyMarkers.createInternalCollections(opCtx).isOK());
     consistencyMarkers.initializeMinValidDocument(opCtx);
 
     // MinValid boundaries should all be null after initializing a new storage engine.
@@ -296,6 +299,7 @@ TEST_F(ReplicationConsistencyMarkersTest, SetMinValidOnPVChange) {
     ReplicationConsistencyMarkersImpl consistencyMarkers(
         getStorageInterface(), minValidNss, oplogTruncateAfterPointNss, checkpointTimestampNss);
     auto opCtx = getOperationContext();
+    ASSERT(consistencyMarkers.createInternalCollections(opCtx).isOK());
     consistencyMarkers.initializeMinValidDocument(opCtx);
 
     auto advanceAndCheckMinValidOpTime = [&](OpTime advanceTo, OpTime expected) {
diff --git a/src/mongo/db/repl/replication_consistency_markers_mock.cpp b/src/mongo/db/repl/replication_consistency_markers_mock.cpp
index 7510ab2679e..e0da40c094c 100644
--- a/src/mongo/db/repl/replication_consistency_markers_mock.cpp
+++ b/src/mongo/db/repl/replication_consistency_markers_mock.cpp
@@ -122,5 +122,9 @@ Timestamp ReplicationConsistencyMarkersMock::getCheckpointTimestamp(OperationCon
     return _checkpointTimestamp;
 }
 
+Status ReplicationConsistencyMarkersMock::createInternalCollections(OperationContext* opCtx) {
+    return Status::OK();
+}
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/replication_consistency_markers_mock.h b/src/mongo/db/repl/replication_consistency_markers_mock.h
index cc6a2009bc1..d74db3cd156 100644
--- a/src/mongo/db/repl/replication_consistency_markers_mock.h
+++ b/src/mongo/db/repl/replication_consistency_markers_mock.h
@@ -72,6 +72,8 @@ public:
     void writeCheckpointTimestamp(OperationContext* opCtx, const Timestamp& timestamp) override;
     Timestamp getCheckpointTimestamp(OperationContext* opCtx) override;
 
+    Status createInternalCollections(OperationContext* opCtx) override;
+
 private:
     mutable stdx::mutex _initialSyncFlagMutex;
     bool _initialSyncFlag = false;
diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
index 3c54e3a1931..899fd3ea5fe 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -449,6 +449,11 @@ void ReplicationCoordinatorImpl::appendConnectionStats(executor::ConnectionPoolS
 }
 
 bool ReplicationCoordinatorImpl::_startLoadLocalConfig(OperationContext* opCtx) {
+    // Create necessary replication collections to guarantee that if a checkpoint sees data after
+    // initial sync has completed, it also sees these collections.
+    fassertStatusOK(50708,
+                    _replicationProcess->getConsistencyMarkers()->createInternalCollections(opCtx));
+
     _replicationProcess->getConsistencyMarkers()->initializeMinValidDocument(opCtx);
 
     StatusWith<LastVote> lastVote = _externalState->loadLocalLastVoteDocument(opCtx);