SERVER-29274 Database::dropCollection() adds drop-pending collection to reaper

If an instance of DropPendingCollectionReaper is not available, the collection
is dropped immediately.

3 files changed, 47 insertions, 2 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 5e37439c403..7e19fa604cb 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -176,6 +176,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/query/query',
+        '$BUILD_DIR/mongo/db/repl/drop_pending_collection_reaper',
         '$BUILD_DIR/mongo/db/repl/serveronly',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/service_context',
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 7abf504cbcb..f5cd2fd9180 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -53,6 +53,7 @@
 #include "mongo/db/introspect.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
+#include "mongo/db/repl/drop_pending_collection_reaper.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/server_options.h"
@@ -437,7 +438,12 @@ Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
     auto uuid = collection->uuid();
 
     // Drop unreplicated collections immediately.
-    if (repl::ReplicationCoordinator::get(opCtx)->isOplogDisabledFor(opCtx, fullns)) {
+    // Replicated collections should also be dropped immediately if there is no active reaper to
+    // remove the drop-pending collections.
+    auto isOplogDisabledForNamespace =
+        repl::ReplicationCoordinator::get(opCtx)->isOplogDisabledFor(opCtx, fullns);
+    auto reaper = repl::DropPendingCollectionReaper::get(opCtx);
+    if (isOplogDisabledForNamespace || !reaper) {
         auto status = _finishDropCollection(opCtx, fullns, collection);
         if (!status.isOK()) {
             return status;
@@ -479,6 +485,11 @@ Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
           << " with drop optime " << dropOpTime;
     fassertStatusOK(40464, renameCollection(opCtx, fullns.ns(), dpns.ns(), stayTemp));
 
+    // Register this drop-pending namespace with DropPendingCollectionReaper to remove when the
+    // committed optime reaches the drop optime.
+    invariant(reaper);
+    reaper->addDropPendingNamespace(dropOpTime, dpns);
+
     return Status::OK();
 }
 
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index 1169867457f..def67507ee1 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -37,11 +37,13 @@
 #include "mongo/db/op_observer.h"
 #include "mongo/db/op_observer_impl.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/repl/drop_pending_collection_reaper.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/repl/storage_interface_mock.h"
 #include "mongo/db/service_context_d_test_fixture.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/unittest/unittest.h"
@@ -138,7 +140,7 @@ TEST_F(DatabaseTest, DropCollectionDropsCollectionButDoesNotLogOperationIfWrites
 }
 
 TEST_F(DatabaseTest,
-       DropCollectionRenamesCollectionToPendingDropNamespaceAndLogsOperationIfWritesAreReplicated) {
+       DropCollectionDropsCollectionAndLogsOperationIfWritesAreReplicatedButReaperIsNotAvailable) {
     ASSERT_TRUE(_opCtx->writesAreReplicated());
     ASSERT_FALSE(
         repl::ReplicationCoordinator::get(_opCtx.get())->isOplogDisabledFor(_opCtx.get(), _nss));
@@ -149,10 +151,41 @@ TEST_F(DatabaseTest,
     auto dropOpTime = repl::ReplClientInfo::forClient(&cc()).getLastOp();
     ASSERT_GREATER_THAN(dropOpTime, repl::OpTime());
 
+    // 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,
+    DropCollectionRenamesCollectionToPendingDropNamespaceAndLogsOperationIfWritesAreReplicatedAndReaperIsAvailable) {
+    ASSERT_TRUE(_opCtx->writesAreReplicated());
+    ASSERT_FALSE(
+        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)));
+
+    _testDropCollection(_opCtx.get(), _nss, true);
+
+    // 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());
+
+    // 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,