SERVER-29274 Database::dropCollection() renames collection using provided drop optime if provided

2 files changed, 128 insertions, 28 deletions

diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 543810e8f18..2e48b0f5246 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -410,11 +410,18 @@ Status DatabaseImpl::dropCollection(OperationContext* opCtx,
 
 Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
                                                 const NamespaceString& fullns,
-                                                repl::OpTime dropOpTimeUnused) {
+                                                repl::OpTime dropOpTime) {
     invariant(opCtx->lockState()->isDbLockedForMode(name(), MODE_X));
 
     LOG(1) << "dropCollection: " << fullns;
 
+    // A valid 'dropOpTime' is not allowed when writes are replicated.
+    if (!dropOpTime.isNull() && opCtx->writesAreReplicated()) {
+        return Status(
+            ErrorCodes::BadValue,
+            "dropCollection() cannot accept a valid drop optime when writes are replicated.");
+    }
+
     Collection* collection = getCollection(opCtx, fullns);
 
     if (!collection) {
@@ -443,10 +450,12 @@ Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
     // Drop unreplicated collections immediately.
     // Replicated collections should also be dropped immediately if there is no active reaper to
     // remove the drop-pending collections.
+    // If 'dropOpTime' is provided and the reaper is available, we should proceed to rename the
+    // collection.
     auto isOplogDisabledForNamespace =
         repl::ReplicationCoordinator::get(opCtx)->isOplogDisabledFor(opCtx, fullns);
     auto reaper = repl::DropPendingCollectionReaper::get(opCtx);
-    if (isOplogDisabledForNamespace || !reaper) {
+    if ((dropOpTime.isNull() && isOplogDisabledForNamespace) || !reaper) {
         auto status = _finishDropCollection(opCtx, fullns, collection);
         if (!status.isOK()) {
             return status;
@@ -457,17 +466,32 @@ Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
 
     // Replicated collections will be renamed with a special drop-pending namespace and dropped when
     // the replica set optime reaches the drop optime.
-    auto dropOpTime =
-        getGlobalServiceContext()->getOpObserver()->onDropCollection(opCtx, fullns, uuid);
-
-    // Drop collection immediately if OpObserver did not write entry to oplog.
-    // After writing the oplog entry, all errors are fatal. See getNextOpTime() comments in
-    // oplog.cpp.
     if (dropOpTime.isNull()) {
-        log() << "dropCollection: " << fullns << " - no drop optime available for pending-drop. "
-              << "Dropping collection immediately.";
-        fassertStatusOK(40462, _finishDropCollection(opCtx, fullns, collection));
-        return Status::OK();
+        dropOpTime =
+            getGlobalServiceContext()->getOpObserver()->onDropCollection(opCtx, fullns, uuid);
+
+        // Drop collection immediately if OpObserver did not write entry to oplog.
+        // After writing the oplog entry, all errors are fatal. See getNextOpTime() comments in
+        // oplog.cpp.
+        if (dropOpTime.isNull()) {
+            log() << "dropCollection: " << fullns
+                  << " - no drop optime available for pending-drop. "
+                  << "Dropping collection immediately.";
+            fassertStatusOK(40462, _finishDropCollection(opCtx, fullns, collection));
+            return Status::OK();
+        }
+    } else {
+        // If we are provided with a valid 'dropOpTime', it means we are dropping this collection
+        // in the context of applying an oplog entry on a secondary.
+        // OpObserver::onDropCollection() should be returning a null OpTime because we should not be
+        // writing to the oplog.
+        auto opTime =
+            getGlobalServiceContext()->getOpObserver()->onDropCollection(opCtx, fullns, uuid);
+        if (!opTime.isNull()) {
+            severe() << "dropCollection: " << fullns
+                     << " - unexpected oplog entry written to the oplog with optime " << opTime;
+            fassertFailed(40468);
+        }
     }
 
     // Check if drop-pending namespace is too long for the index names in the collection.
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index def67507ee1..06d77d39b17 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -94,30 +94,35 @@ void DatabaseTest::setUp() {
 void DatabaseTest::tearDown() {
     _nss = {};
     _opCtx = {};
+
+    auto service = getServiceContext();
+    repl::DropPendingCollectionReaper::set(service, {});
+    repl::StorageInterface::set(service, {});
+
     ServiceContextMongoDTest::tearDown();
 }
 
 void _testDropCollection(OperationContext* opCtx,
                          const NamespaceString& nss,
-                         bool createCollectionBeforeDrop) {
-    writeConflictRetry(
-        opCtx, "testDropCollection", nss.ns(), [opCtx, nss, createCollectionBeforeDrop] {
-            AutoGetOrCreateDb autoDb(opCtx, nss.db(), MODE_X);
-            auto db = autoDb.getDb();
-            ASSERT_TRUE(db);
+                         bool createCollectionBeforeDrop,
+                         const repl::OpTime& dropOpTime = {}) {
+    writeConflictRetry(opCtx, "testDropCollection", nss.ns(), [=] {
+        AutoGetOrCreateDb autoDb(opCtx, nss.db(), MODE_X);
+        auto db = autoDb.getDb();
+        ASSERT_TRUE(db);
 
-            WriteUnitOfWork wuow(opCtx);
-            if (createCollectionBeforeDrop) {
-                ASSERT_TRUE(db->createCollection(opCtx, nss.ns()));
-            } else {
-                ASSERT_FALSE(db->getCollection(opCtx, nss));
-            }
+        WriteUnitOfWork wuow(opCtx);
+        if (createCollectionBeforeDrop) {
+            ASSERT_TRUE(db->createCollection(opCtx, nss.ns()));
+        } else {
+            ASSERT_FALSE(db->getCollection(opCtx, nss));
+        }
 
-            ASSERT_OK(db->dropCollection(opCtx, nss.ns()));
+        ASSERT_OK(db->dropCollection(opCtx, nss.ns(), dropOpTime));
 
-            ASSERT_FALSE(db->getCollection(opCtx, nss));
-            wuow.commit();
-        });
+        ASSERT_FALSE(db->getCollection(opCtx, nss));
+        wuow.commit();
+    });
 }
 
 TEST_F(DatabaseTest, DropCollectionReturnsOKIfCollectionDoesNotExist) {
@@ -188,6 +193,77 @@ TEST_F(
     ASSERT_EQUALS(dropOpTime, *reaperEarliestDropOpTime);
 }
 
+TEST_F(DatabaseTest, DropCollectionRejectsProvidedDropOpTimeIfWritesAreReplicated) {
+    ASSERT_TRUE(_opCtx->writesAreReplicated());
+    ASSERT_FALSE(
+        repl::ReplicationCoordinator::get(_opCtx.get())->isOplogDisabledFor(_opCtx.get(), _nss));
+
+    auto opCtx = _opCtx.get();
+    auto nss = _nss;
+    writeConflictRetry(opCtx, "testDropOpTimeWithReplicated", nss.ns(), [opCtx, nss] {
+        AutoGetOrCreateDb autoDb(opCtx, nss.db(), MODE_X);
+        auto db = autoDb.getDb();
+        ASSERT_TRUE(db);
+
+        WriteUnitOfWork wuow(opCtx);
+        ASSERT_TRUE(db->createCollection(opCtx, nss.ns()));
+
+        repl::OpTime dropOpTime(Timestamp(Seconds(100), 0), 1LL);
+        ASSERT_EQUALS(ErrorCodes::BadValue, db->dropCollection(opCtx, nss.ns(), dropOpTime));
+    });
+}
+
+TEST_F(DatabaseTest,
+       DropCollectionIgnoresProvidedDropOpTimeAndDropsCollectionImmediatelyIfReaperIsNotAvailable) {
+    repl::UnreplicatedWritesBlock uwb(_opCtx.get());
+    ASSERT_FALSE(_opCtx->writesAreReplicated());
+    ASSERT_TRUE(
+        repl::ReplicationCoordinator::get(_opCtx.get())->isOplogDisabledFor(_opCtx.get(), _nss));
+
+    repl::OpTime dropOpTime(Timestamp(Seconds(100), 0), 1LL);
+    _testDropCollection(_opCtx.get(), _nss, true, dropOpTime);
+
+    // Last optime in repl client is null because we did not write to the oplog.
+    ASSERT_EQUALS(repl::OpTime(), repl::ReplClientInfo::forClient(&cc()).getLastOp());
+
+    // If the drop-pending collection reaper is not available, the collection is not renamed to
+    // <db>.system.drop.*.
+    auto dpns = _nss.makeDropPendingNamespace(dropOpTime);
+    ASSERT_FALSE(mongo::AutoGetCollectionForRead(_opCtx.get(), dpns).getCollection());
+}
+
+TEST_F(
+    DatabaseTest,
+    DropCollectionRenamesCollectionToPendingDropNamespaceUsingProvidedDropOpTimeButDoesNotLogOperationIfReaperIsAvailable) {
+    repl::UnreplicatedWritesBlock uwb(_opCtx.get());
+    ASSERT_FALSE(_opCtx->writesAreReplicated());
+    ASSERT_TRUE(
+        repl::ReplicationCoordinator::get(_opCtx.get())->isOplogDisabledFor(_opCtx.get(), _nss));
+
+    auto service = getServiceContext();
+    repl::StorageInterface::set(service, stdx::make_unique<repl::StorageInterfaceMock>());
+    repl::DropPendingCollectionReaper::set(
+        service,
+        stdx::make_unique<repl::DropPendingCollectionReaper>(repl::StorageInterface::get(service)));
+
+    repl::OpTime dropOpTime(Timestamp(Seconds(100), 0), 1LL);
+    _testDropCollection(_opCtx.get(), _nss, true, dropOpTime);
+
+    // Last optime in repl client is null because we did not write to the oplog.
+    ASSERT_EQUALS(repl::OpTime(), repl::ReplClientInfo::forClient(&cc()).getLastOp());
+
+    // Replicated collection is renamed with a special drop-pending names in the <db>.system.drop.*
+    // namespace.
+    auto dpns = _nss.makeDropPendingNamespace(dropOpTime);
+    ASSERT_TRUE(mongo::AutoGetCollectionForRead(_opCtx.get(), dpns).getCollection());
+
+    // Reaper should have the drop optime of the collection.
+    auto reaperEarliestDropOpTime =
+        repl::DropPendingCollectionReaper::get(service)->getEarliestDropOpTime();
+    ASSERT_TRUE(reaperEarliestDropOpTime);
+    ASSERT_EQUALS(dropOpTime, *reaperEarliestDropOpTime);
+}
+
 void _testDropCollectionThrowsExceptionIfThereAreIndexesInProgress(OperationContext* opCtx,
                                                                    const NamespaceString& nss) {
     writeConflictRetry(opCtx, "testDropCollectionWithIndexesInProgress", nss.ns(), [opCtx, nss] {