SERVER-29273 rename dropped collection to special drop-pending name

The drop-pending collection will be removed by DropPendingCollectionReaper
when the replica set commit level reaches the drop optime.

3 files changed, 109 insertions, 4 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 9294840155c..271301d4d5f 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -101,6 +101,7 @@ env.CppUnitTest(
     ],
     LIBDEPS=[
         'database',
+        'index_create',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/op_observer_d',
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index f78d5270fc6..d0641b202a4 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -421,6 +421,15 @@ Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
 
     BackgroundOperation::assertNoBgOpInProgForNs(fullns);
 
+    // Make sure no indexes builds are in progress.
+    // Use massert() to be consistent with IndexCatalog::dropAllIndexes().
+    auto numIndexesInProgress = collection->getIndexCatalog()->numIndexesInProgress(opCtx);
+    massert(40461,
+            str::stream() << "cannot drop collection " << fullns.ns() << " when "
+                          << numIndexesInProgress
+                          << " index builds in progress.",
+            numIndexesInProgress == 0);
+
     audit::logDropCollection(&cc(), fullns.toString());
 
     collection->getCursorManager()->invalidateAll(opCtx, true, "collection dropped");
@@ -428,12 +437,49 @@ Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
     Top::get(opCtx->getClient()->getServiceContext()).collectionDropped(fullns.toString());
 
     auto uuid = collection->uuid();
-    auto status = _finishDropCollection(opCtx, fullns, collection);
+
+    // Drop unreplicated collections immediately.
+    if (repl::ReplicationCoordinator::get(opCtx)->isOplogDisabledFor(opCtx, fullns)) {
+        auto status = _finishDropCollection(opCtx, fullns, collection);
+        if (!status.isOK()) {
+            return status;
+        }
+        getGlobalServiceContext()->getOpObserver()->onDropCollection(opCtx, fullns, uuid);
+        return Status::OK();
+    }
+
+    // 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();
+    }
+
+    // Check if drop-pending namespace is too long for the index names in the collection.
+    auto dpns = fullns.makeDropPendingNamespace(dropOpTime);
+    auto status =
+        dpns.checkLengthForRename(collection->getIndexCatalog()->getLongestIndexNameLength(opCtx));
     if (!status.isOK()) {
-        return status;
+        log() << "dropCollection: " << fullns
+              << " - cannot proceed with collection rename for pending-drop: " << status
+              << ". Dropping collection immediately.";
+        fassertStatusOK(40463, _finishDropCollection(opCtx, fullns, collection));
+        return Status::OK();
     }
 
-    getGlobalServiceContext()->getOpObserver()->onDropCollection(opCtx, fullns, uuid);
+    // Rename collection using drop-pending namespace generated from drop optime.
+    const bool stayTemp = true;
+    log() << "dropCollection: " << fullns << " - renaming to drop-pending collection: " << dpns
+          << " with drop optime " << dropOpTime;
+    fassertStatusOK(40464, renameCollection(opCtx, fullns.ns(), dpns.ns(), stayTemp));
 
     return Status::OK();
 }
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index 2818bb11b98..1169867457f 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -28,9 +28,11 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/catalog/index_create.h"
 #include "mongo/db/client.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/op_observer_impl.h"
@@ -43,6 +45,7 @@
 #include "mongo/db/service_context_d_test_fixture.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/unittest/unittest.h"
+#include "mongo/util/scopeguard.h"
 
 namespace {
 
@@ -134,7 +137,8 @@ TEST_F(DatabaseTest, DropCollectionDropsCollectionButDoesNotLogOperationIfWrites
     ASSERT_EQUALS(repl::OpTime(), dropOpTime);
 }
 
-TEST_F(DatabaseTest, DropCollectionDropsCollectionAndLogsOperationIfWritesAreReplicated) {
+TEST_F(DatabaseTest,
+       DropCollectionRenamesCollectionToPendingDropNamespaceAndLogsOperationIfWritesAreReplicated) {
     ASSERT_TRUE(_opCtx->writesAreReplicated());
     ASSERT_FALSE(
         repl::ReplicationCoordinator::get(_opCtx.get())->isOplogDisabledFor(_opCtx.get(), _nss));
@@ -144,6 +148,60 @@ TEST_F(DatabaseTest, DropCollectionDropsCollectionAndLogsOperationIfWritesAreRep
     // Drop optime is non-null because an op was written to the oplog.
     auto dropOpTime = repl::ReplClientInfo::forClient(&cc()).getLastOp();
     ASSERT_GREATER_THAN(dropOpTime, repl::OpTime());
+
+    // 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());
+}
+
+void _testDropCollectionThrowsExceptionIfThereAreIndexesInProgress(OperationContext* opCtx,
+                                                                   const NamespaceString& nss) {
+    writeConflictRetry(opCtx, "testDropCollectionWithIndexesInProgress", nss.ns(), [opCtx, nss] {
+        AutoGetOrCreateDb autoDb(opCtx, nss.db(), MODE_X);
+        auto db = autoDb.getDb();
+        ASSERT_TRUE(db);
+
+        Collection* collection = nullptr;
+        {
+            WriteUnitOfWork wuow(opCtx);
+            ASSERT_TRUE(collection = db->createCollection(opCtx, nss.ns()));
+            wuow.commit();
+        }
+
+        MultiIndexBlock indexer(opCtx, collection);
+        ON_BLOCK_EXIT([&indexer, opCtx] {
+            WriteUnitOfWork wuow(opCtx);
+            indexer.commit();
+            wuow.commit();
+        });
+
+        auto indexCatalog = collection->getIndexCatalog();
+        ASSERT_EQUALS(indexCatalog->numIndexesInProgress(opCtx), 0);
+        auto indexInfoObj = BSON(
+            "v" << int(IndexDescriptor::kLatestIndexVersion) << "key" << BSON("a" << 1) << "name"
+                << "a_1"
+                << "ns"
+                << nss.ns());
+        ASSERT_OK(indexer.init(indexInfoObj).getStatus());
+        ASSERT_GREATER_THAN(indexCatalog->numIndexesInProgress(opCtx), 0);
+
+        WriteUnitOfWork wuow(opCtx);
+        ASSERT_THROWS_CODE(db->dropCollection(opCtx, nss.ns()), MsgAssertionException, 40461);
+    });
+}
+
+TEST_F(DatabaseTest,
+       DropCollectionThrowsExceptionIfThereAreIndexesInProgressAndWritesAreNotReplicated) {
+    repl::UnreplicatedWritesBlock uwb(_opCtx.get());
+    ASSERT_FALSE(_opCtx->writesAreReplicated());
+    _testDropCollectionThrowsExceptionIfThereAreIndexesInProgress(_opCtx.get(), _nss);
+}
+
+TEST_F(DatabaseTest,
+       DropCollectionThrowsExceptionIfThereAreIndexesInProgressAndWritesAreReplicated) {
+    ASSERT_TRUE(_opCtx->writesAreReplicated());
+    _testDropCollectionThrowsExceptionIfThereAreIndexesInProgress(_opCtx.get(), _nss);
 }
 
 }  // namespace