SERVER-49807 Move onCommit KVDropPendingIdentReaper handling out of the DurableCatlaog

17 files changed, 276 insertions, 122 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 92eb7d4af79..95446be6a41 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -904,6 +904,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/repl/timestamp_block',
         '$BUILD_DIR/mongo/db/s/sharding_api_d',
+        '$BUILD_DIR/mongo/db/storage/storage_util',
         '$BUILD_DIR/mongo/util/fail_point',
         "$BUILD_DIR/mongo/executor/task_executor_interface",
     ],
@@ -1050,6 +1051,7 @@ env.Library(
         'catalog/index_key_validate',
         'catalog/multi_index_block',
         'storage/storage_repair_observer',
+        'storage/storage_util',
         'repl/oplog',
     ],
 )
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 5ae71b7908d..9b593066c0d 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -355,6 +355,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/repl/repl_settings',
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
         '$BUILD_DIR/mongo/db/storage/storage_debug_util',
+        '$BUILD_DIR/mongo/db/storage/storage_util',
         '$BUILD_DIR/mongo/db/transaction',
         '$BUILD_DIR/mongo/db/vector_clock',
         'index_build_block',
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index f8347739529..db70d24876e 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -73,6 +73,7 @@
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/storage/storage_options.h"
+#include "mongo/db/storage/storage_util.h"
 #include "mongo/db/system_index.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
@@ -492,7 +493,10 @@ Status DatabaseImpl::_finishDropCollection(OperationContext* opCtx,
           "namespace"_attr = nss,
           "uuid"_attr = uuid);
 
-    auto status = DurableCatalog::get(opCtx)->dropCollection(opCtx, collection->getCatalogId());
+    auto status = catalog::dropCollection(opCtx,
+                                          collection->ns(),
+                                          collection->getCatalogId(),
+                                          collection->getRecordStore()->getIdent());
     if (!status.isOK())
         return status;
 
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 018796ceee5..eb13d37a434 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -72,6 +72,7 @@
 #include "mongo/db/storage/execution_context.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
+#include "mongo/db/storage/storage_util.h"
 #include "mongo/db/ttl_collection_cache.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/assert_util.h"
@@ -1083,21 +1084,8 @@ Status IndexCatalogImpl::dropIndexEntry(OperationContext* opCtx, IndexCatalogEnt
 
 void IndexCatalogImpl::deleteIndexFromDisk(OperationContext* opCtx, const string& indexName) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(_collection->ns(), MODE_X));
-
-    Status status =
-        DurableCatalog::get(opCtx)->removeIndex(opCtx, _collection->getCatalogId(), indexName);
-    if (status.code() == ErrorCodes::NamespaceNotFound) {
-        /*
-         * This is ok, as we may be partially through index creation.
-         */
-    } else if (!status.isOK()) {
-        LOGV2_WARNING(20364,
-                      "couldn't drop index {index} on collection: {namespace} because of {error}",
-                      "Couldn't drop index",
-                      "index"_attr = indexName,
-                      "namespace"_attr = _collection->ns(),
-                      "error"_attr = redact(status));
-    }
+    catalog::removeIndex(
+        opCtx, indexName, _collection->getCatalogId(), _collection->uuid(), _collection->ns());
 }
 
 void IndexCatalogImpl::setMultikeyPaths(OperationContext* const opCtx,
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index 34e06f06311..3f2bdd7f2b9 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -58,6 +58,7 @@
 #include "mongo/db/server_recovery.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/durable_catalog.h"
+#include "mongo/db/storage/storage_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/shard_key_pattern.h"
 #include "mongo/util/assert_util.h"
@@ -501,11 +502,11 @@ Status IndexBuildsCoordinator::_startIndexBuildForRecovery(OperationContext* opC
                     return s;
                 }
             } else {
-                Status status = DurableCatalog::get(opCtx)->removeIndex(
-                    opCtx, collection->getCatalogId(), indexNames[i]);
-                if (!status.isOK()) {
-                    return status;
-                }
+                catalog::removeIndex(opCtx,
+                                     indexNames[i],
+                                     collection->getCatalogId(),
+                                     collection->uuid(),
+                                     collection->ns());
             }
         }
 
diff --git a/src/mongo/db/repair.cpp b/src/mongo/db/repair.cpp
index 89f7e28e244..d69dbcfb691 100644
--- a/src/mongo/db/repair.cpp
+++ b/src/mongo/db/repair.cpp
@@ -60,6 +60,7 @@
 #include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_repair_observer.h"
+#include "mongo/db/storage/storage_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/scopeguard.h"
 
@@ -98,10 +99,8 @@ Status dropUnfinishedIndexes(OperationContext* opCtx, Collection* collection) {
                   "Dropping unfinished index after collection was modified by repair",
                   "index"_attr = indexName);
             WriteUnitOfWork wuow(opCtx);
-            auto status = durableCatalog->removeIndex(opCtx, collection->getCatalogId(), indexName);
-            if (!status.isOK()) {
-                return status;
-            }
+            catalog::removeIndex(
+                opCtx, indexName, collection->getCatalogId(), collection->uuid(), collection->ns());
             wuow.commit();
             StorageRepairObserver::get(opCtx->getServiceContext())
                 ->invalidatingModification(str::stream()
diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index fc8a18d2c20..84c84d3c071 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -487,7 +487,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/index_names',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/storage/bson_collection_catalog_entry',
-        '$BUILD_DIR/mongo/db/storage/kv/kv_drop_pending_ident_reaper',
         '$BUILD_DIR/mongo/db/storage/kv/kv_prefix',
         '$BUILD_DIR/mongo/db/catalog/collection_catalog',
     ],
@@ -497,6 +496,18 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_options_core',
     ],
 )
+
+env.Library(
+    target='storage_util',
+    source=[
+        'storage_util.cpp',
+    ],
+    LIBDEPS_PRIVATE=[
+        'durable_catalog_impl',
+        'kv/kv_drop_pending_ident_reaper',
+    ],
+)
+
 env.Library(
     target='two_phase_index_build_knobs_idl',
     source=[
@@ -526,6 +537,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/collection_catalog_helper',
         '$BUILD_DIR/mongo/db/resumable_index_builds_idl',
         '$BUILD_DIR/mongo/db/vector_clock',
+        'storage_util',
         'two_phase_index_build_knobs_idl',
     ],
 )
diff --git a/src/mongo/db/storage/durable_catalog.h b/src/mongo/db/storage/durable_catalog.h
index 2a498925e8a..e3b432b7fdc 100644
--- a/src/mongo/db/storage/durable_catalog.h
+++ b/src/mongo/db/storage/durable_catalog.h
@@ -143,6 +143,8 @@ public:
                                     bool stayTemp) = 0;
 
     /**
+     * Deletes the collection catalog entry identified by 'catalogId'.
+     *
      * Expects (invariants) that all of the index catalog entries have been removed already via
      * removeIndex.
      */
@@ -200,9 +202,11 @@ public:
                                  StringData validationLevel,
                                  StringData validationAction) = 0;
 
-    virtual Status removeIndex(OperationContext* opCtx,
-                               RecordId catalogId,
-                               StringData indexName) = 0;
+    /**
+     * Removes the 'indexName' index catalog entry from the collection catalog entry identified by
+     * 'catalogId'.
+     */
+    virtual void removeIndex(OperationContext* opCtx, RecordId catalogId, StringData indexName) = 0;
 
     /**
      * Updates the persisted catalog entry for 'ns' with the new index and creates the index on
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index 5d05ea365d8..f208a417439 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -239,53 +239,6 @@ public:
     const std::string _ident;
 };
 
-class DurableCatalogImpl::RemoveIndexChange : public RecoveryUnit::Change {
-public:
-    RemoveIndexChange(OperationContext* opCtx,
-                      StorageEngineInterface* engine,
-                      OptionalCollectionUUID uuid,
-                      const NamespaceString& indexNss,
-                      StringData indexName,
-                      StringData ident)
-        : _opCtx(opCtx),
-          _recoveryUnit(opCtx->recoveryUnit()),
-          _engine(engine),
-          _uuid(uuid),
-          _indexNss(indexNss),
-          _indexName(indexName),
-          _ident(ident.toString()) {}
-
-    virtual void rollback() {}
-    virtual void commit(boost::optional<Timestamp> commitTimestamp) {
-        // Intentionally ignoring failure here. Since we've removed the metadata pointing to the
-        // index, we should never see it again anyway.
-        if (_engine->getStorageEngine()->supportsPendingDrops() && commitTimestamp) {
-            LOGV2(22206,
-                  "Deferring table drop for index '{index}' on collection "
-                  "'{namespace}{uuid}. Ident: '{ident}', commit timestamp: '{commitTimestamp}'",
-                  "Deferring table drop for index",
-                  "index"_attr = _indexName,
-                  logAttrs(_indexNss),
-                  "uuid"_attr = _uuid,
-                  "ident"_attr = _ident,
-                  "commitTimestamp"_attr = commitTimestamp);
-            _engine->addDropPendingIdent(*commitTimestamp, _indexNss, _ident);
-        } else {
-            auto kvEngine = _engine->getEngine();
-            MONGO_COMPILER_VARIABLE_UNUSED auto status =
-                kvEngine->dropIdent(_opCtx, _recoveryUnit, _ident);
-        }
-    }
-
-    OperationContext* const _opCtx;
-    RecoveryUnit* const _recoveryUnit;
-    StorageEngineInterface* _engine;
-    OptionalCollectionUUID _uuid;
-    const NamespaceString _indexNss;
-    const std::string _indexName;
-    const std::string _ident;
-};
-
 bool DurableCatalogImpl::FeatureTracker::isFeatureDocument(BSONObj obj) {
     BSONElement firstElem = obj.firstElement();
     if (firstElem.fieldNameStringData() == kIsFeatureDocumentFieldName) {
@@ -931,29 +884,6 @@ Status DurableCatalogImpl::dropCollection(OperationContext* opCtx, RecordId cata
         return status;
     }
 
-    auto ru = opCtx->recoveryUnit();
-    // This will notify the storageEngine to drop the collection only on WUOW::commit().
-    opCtx->recoveryUnit()->onCommit(
-        [opCtx, ru, catalog = this, entry](boost::optional<Timestamp> commitTimestamp) {
-            StorageEngineInterface* engine = catalog->_engine;
-            auto storageEngine = engine->getStorageEngine();
-            if (storageEngine->supportsPendingDrops() && commitTimestamp) {
-                LOGV2(22214,
-                      "Deferring table drop for collection '{namespace}'. Ident: {ident}, "
-                      "commit timestamp: {commitTimestamp}",
-                      "Deferring table drop for collection",
-                      logAttrs(entry.nss),
-                      "ident"_attr = entry.ident,
-                      "commitTimestamp"_attr = commitTimestamp);
-                engine->addDropPendingIdent(*commitTimestamp, entry.nss, entry.ident);
-            } else {
-                // Intentionally ignoring failure here. Since we've removed the metadata pointing to
-                // the collection, we should never see it again anyway.
-                auto kvEngine = engine->getEngine();
-                kvEngine->dropIdent(opCtx, ru, entry.ident).ignore();
-            }
-        });
-
     return Status::OK();
 }
 
@@ -1022,23 +952,18 @@ void DurableCatalogImpl::updateValidator(OperationContext* opCtx,
     putMetaData(opCtx, catalogId, md);
 }
 
-Status DurableCatalogImpl::removeIndex(OperationContext* opCtx,
-                                       RecordId catalogId,
-                                       StringData indexName) {
+void DurableCatalogImpl::removeIndex(OperationContext* opCtx,
+                                     RecordId catalogId,
+                                     StringData indexName) {
     BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, catalogId);
 
     if (md.findIndexOffset(indexName) < 0)
-        return Status::OK();  // never had the index so nothing to do.
+        return;  // never had the index so nothing to do.
 
     const string ident = getIndexIdent(opCtx, catalogId, indexName);
 
     md.eraseIndex(indexName);
     putMetaData(opCtx, catalogId, md);
-
-    // Lazily remove to isolate underlying engine from rollback.
-    opCtx->recoveryUnit()->registerChange(std::make_unique<RemoveIndexChange>(
-        opCtx, _engine, md.options.uuid, NamespaceString(md.ns), indexName, ident));
-    return Status::OK();
 }
 
 Status DurableCatalogImpl::prepareForIndexBuild(OperationContext* opCtx,
diff --git a/src/mongo/db/storage/durable_catalog_impl.h b/src/mongo/db/storage/durable_catalog_impl.h
index 12d030fd13a..b24816d333d 100644
--- a/src/mongo/db/storage/durable_catalog_impl.h
+++ b/src/mongo/db/storage/durable_catalog_impl.h
@@ -148,7 +148,7 @@ public:
                          StringData validationLevel,
                          StringData validationAction);
 
-    Status removeIndex(OperationContext* opCtx, RecordId catalogId, StringData indexName);
+    void removeIndex(OperationContext* opCtx, RecordId catalogId, StringData indexName);
 
     Status prepareForIndexBuild(OperationContext* opCtx,
                                 RecordId catalogId,
@@ -206,7 +206,6 @@ private:
     class AddIdentChange;
     class RemoveIdentChange;
     class AddIndexChange;
-    class RemoveIndexChange;
 
     friend class StorageEngineImpl;
     friend class DurableCatalogImplTest;
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
index 7e945346125..4a08cdfc1a0 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
@@ -85,7 +85,7 @@ std::unique_ptr<mongo::RecordStore> KVEngine::makeTemporaryRecordStore(Operation
 };
 
 
-std::unique_ptr<mongo::RecordStore> KVEngine::getRecordStore(OperationContext* opCtx,
+std::unique_ptr<mongo::RecordStore> KVEngine::getRecordStore(OperationContext* unused,
                                                              StringData ns,
                                                              StringData ident,
                                                              const CollectionOptions& options) {
@@ -143,7 +143,9 @@ std::unique_ptr<mongo::SortedDataInterface> KVEngine::getSortedDataInterface(
         return std::make_unique<SortedDataInterfaceStandard>(opCtx, ident, desc);
 }
 
-Status KVEngine::dropIdent(OperationContext* opCtx, mongo::RecoveryUnit* ru, StringData ident) {
+Status KVEngine::dropIdent(OperationContext* unusedOpCtx,
+                           mongo::RecoveryUnit* ru,
+                           StringData ident) {
     Status dropStatus = Status::OK();
     stdx::unique_lock lock(_identsLock);
     if (_idents.count(ident.toString()) > 0) {
@@ -153,7 +155,7 @@ Status KVEngine::dropIdent(OperationContext* opCtx, mongo::RecoveryUnit* ru, Str
         lock.unlock();
         if (isRecordStore) {  // ident is RecordStore.
             CollectionOptions s;
-            auto rs = getRecordStore(/*unused*/ opCtx, ""_sd, ident, s);
+            auto rs = getRecordStore(/*unused*/ unusedOpCtx, ""_sd, ident, s);
             dropStatus =
                 checked_cast<RecordStore*>(rs.get())->truncateWithoutUpdatingCount(ru).getStatus();
         } else {  // ident is SortedDataInterface.
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index 085336a23c6..ccbe3ff5c1f 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -442,6 +442,14 @@ public:
     virtual void clearDropPendingState() = 0;
 
     /**
+     * Adds 'ident' to a list of indexes/collections whose data will be dropped when the
+     * 'dropTimestamp' is sufficiently old to ensure no future data accesses.
+     */
+    virtual void addDropPendingIdent(const Timestamp& dropTimestamp,
+                                     const NamespaceString& nss,
+                                     StringData ident) = 0;
+
+    /**
      * Recovers the storage engine state to the last stable timestamp. "Stable" in this case
      * refers to a timestamp that is guaranteed to never be rolled back. The stable timestamp
      * used should be one provided by StorageEngine::setStableTimestamp().
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 002f20ca43b..3fc33776c13 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -46,6 +46,7 @@
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/temporary_kv_record_store.h"
 #include "mongo/db/storage/storage_repair_observer.h"
+#include "mongo/db/storage/storage_util.h"
 #include "mongo/db/storage/two_phase_index_build_knobs_gen.h"
 #include "mongo/logv2/log.h"
 #include "mongo/stdx/unordered_map.h"
@@ -732,14 +733,15 @@ Status StorageEngineImpl::_dropCollectionsNoTimestamp(OperationContext* opCtx,
             coll->getIndexCatalog()->getIndexIterator(opCtx, true /* includeUnfinishedIndexes */);
         while (ii->more()) {
             const IndexCatalogEntry* ice = ii->next();
-            Status status = durableCatalog->removeIndex(
-                opCtx, coll->getCatalogId(), ice->descriptor()->indexName());
-            if (!status.isOK() && firstError.isOK()) {
-                firstError = status;
-            }
+            catalog::removeIndex(opCtx,
+                                 ice->descriptor()->indexName(),
+                                 coll->getCatalogId(),
+                                 coll->uuid(),
+                                 coll->ns());
         }
 
-        Status result = durableCatalog->dropCollection(opCtx, coll->getCatalogId());
+        Status result = catalog::dropCollection(
+            opCtx, coll->ns(), coll->getCatalogId(), coll->getRecordStore()->getIdent());
         if (!result.isOK() && firstError.isOK()) {
             firstError = result;
         }
diff --git a/src/mongo/db/storage/storage_engine_interface.h b/src/mongo/db/storage/storage_engine_interface.h
index 77703c8aa6f..23f2b3f2058 100644
--- a/src/mongo/db/storage/storage_engine_interface.h
+++ b/src/mongo/db/storage/storage_engine_interface.h
@@ -41,9 +41,6 @@ public:
     virtual ~StorageEngineInterface() = default;
     virtual StorageEngine* getStorageEngine() = 0;
     virtual KVEngine* getEngine() = 0;
-    virtual void addDropPendingIdent(const Timestamp& dropTimestamp,
-                                     const NamespaceString& nss,
-                                     StringData ident) = 0;
     virtual DurableCatalog* getCatalog() = 0;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/storage/storage_engine_mock.h b/src/mongo/db/storage/storage_engine_mock.h
index 1b303b35985..7592448041a 100644
--- a/src/mongo/db/storage/storage_engine_mock.h
+++ b/src/mongo/db/storage/storage_engine_mock.h
@@ -162,6 +162,9 @@ public:
     std::set<std::string> getDropPendingIdents() const final {
         return {};
     }
+    void addDropPendingIdent(const Timestamp& dropTimestamp,
+                             const NamespaceString& nss,
+                             StringData ident) final {}
     Status currentFilesCompatible(OperationContext* opCtx) const final {
         return Status::OK();
     }
diff --git a/src/mongo/db/storage/storage_util.cpp b/src/mongo/db/storage/storage_util.cpp
new file mode 100644
index 00000000000..094f0cdafa0
--- /dev/null
+++ b/src/mongo/db/storage/storage_util.cpp
@@ -0,0 +1,138 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
+
+#include "mongo/platform/basic.h"
+
+#include <string>
+
+#include "mongo/db/storage/storage_util.h"
+
+#include "mongo/db/storage/durable_catalog.h"
+#include "mongo/db/storage/kv/kv_engine.h"
+#include "mongo/db/storage/storage_engine.h"
+#include "mongo/logv2/log.h"
+
+namespace mongo {
+namespace catalog {
+
+void removeIndex(OperationContext* opCtx,
+                 StringData indexName,
+                 RecordId collectionCatalogId,
+                 UUID collectionUUID,
+                 const NamespaceString& nss) {
+    const std::string indexIdent =
+        DurableCatalog::get(opCtx)->getIndexIdent(opCtx, collectionCatalogId, indexName);
+
+    // Run the first phase of drop to remove the catalog entry.
+    DurableCatalog::get(opCtx)->removeIndex(opCtx, collectionCatalogId, indexName);
+
+    // The OperationContext may not be valid when the RecoveryUnit executes the onCommit handlers.
+    // Therefore, anything that would normally be fetched from the opCtx must be passed in
+    // separately to the onCommit handler below.
+    //
+    // Index creation (and deletion) are allowed in multi-document transactions that use the same
+    // RecoveryUnit throughout but not the same OperationContext.
+    auto recoveryUnit = opCtx->recoveryUnit();
+    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
+
+    // Schedule the second phase of drop to delete the data when it is no longer in use, if the
+    // first phase is successuflly committed.
+    opCtx->recoveryUnit()->onCommit([opCtx,
+                                     recoveryUnit,
+                                     storageEngine,
+                                     collectionUUID,
+                                     nss,
+                                     indexNameStr = indexName.toString(),
+                                     indexIdent](boost::optional<Timestamp> commitTimestamp) {
+        if (storageEngine->supportsPendingDrops() && commitTimestamp) {
+            LOGV2(22206,
+                  "Deferring table drop for index '{index}' on collection "
+                  "'{namespace}{uuid}. Ident: '{ident}', commit timestamp: '{commitTimestamp}'",
+                  "Deferring table drop for index",
+                  "index"_attr = indexNameStr,
+                  logAttrs(nss),
+                  "uuid"_attr = collectionUUID,
+                  "ident"_attr = indexIdent,
+                  "commitTimestamp"_attr = commitTimestamp);
+            storageEngine->addDropPendingIdent(*commitTimestamp, nss, indexIdent);
+        } else {
+            auto kvEngine = storageEngine->getEngine();
+            kvEngine->dropIdent(opCtx, recoveryUnit, indexIdent).ignore();
+        }
+    });
+}
+
+Status dropCollection(OperationContext* opCtx,
+                      const NamespaceString& nss,
+                      RecordId collectionCatalogId,
+                      StringData ident) {
+    // Run the first phase of drop to remove the catalog entry.
+    Status status = DurableCatalog::get(opCtx)->dropCollection(opCtx, collectionCatalogId);
+    if (!status.isOK()) {
+        return status;
+    }
+
+    // The OperationContext may not be valid when the RecoveryUnit executes the onCommit handlers.
+    // Therefore, anything that would normally be fetched from the opCtx must be passed in
+    // separately to the onCommit handler below.
+    //
+    // Create (and drop) collection are allowed in multi-document transactions that use the same
+    // RecoveryUnit throughout but not the same OperationContext.
+    auto recoveryUnit = opCtx->recoveryUnit();
+    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
+
+    // Schedule the second phase of drop to delete the data when it is no longer in use, if the
+    // first phase is successuflly committed.
+    opCtx->recoveryUnit()->onCommit(
+        [opCtx, recoveryUnit, storageEngine, nss, identStr = ident.toString()](
+            boost::optional<Timestamp> commitTimestamp) {
+            if (storageEngine->supportsPendingDrops() && commitTimestamp) {
+                LOGV2(22214,
+                      "Deferring table drop for collection '{namespace}'. Ident: {ident}, "
+                      "commit timestamp: {commitTimestamp}",
+                      "Deferring table drop for collection",
+                      logAttrs(nss),
+                      "ident"_attr = identStr,
+                      "commitTimestamp"_attr = commitTimestamp);
+                storageEngine->addDropPendingIdent(*commitTimestamp, nss, identStr);
+            } else {
+                // Intentionally ignoring failure here. Since we've removed the metadata pointing to
+                // the collection, we should never see it again anyway.
+                auto kvEngine = storageEngine->getEngine();
+                kvEngine->dropIdent(opCtx, recoveryUnit, identStr).ignore();
+            }
+        });
+
+    return Status::OK();
+}
+
+}  // namespace catalog
+}  // namespace mongo
diff --git a/src/mongo/db/storage/storage_util.h b/src/mongo/db/storage/storage_util.h
new file mode 100644
index 00000000000..880efda27cf
--- /dev/null
+++ b/src/mongo/db/storage/storage_util.h
@@ -0,0 +1,69 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/record_id.h"
+#include "mongo/util/uuid.h"
+
+namespace mongo {
+
+class OperationContext;
+class NamespaceString;
+
+namespace catalog {
+
+/**
+ * Performs two-phase index drop.
+ *
+ * Passthrough to DurableCatalog::removeIndex to execute the first phase of drop by removing the
+ * index catalog entry, then registers an onCommit hook to schedule the second phase of drop to
+ * delete the index data.
+ */
+void removeIndex(OperationContext* opCtx,
+                 StringData indexName,
+                 RecordId collectionCatalogId,
+                 UUID collectionUUID,
+                 const NamespaceString& nss);
+
+/**
+ * Performs two-phase collection drop.
+ *
+ * Passthrough to DurableCatalog::dropCollection to execute the first phase of drop by removing the
+ * collection entry, then registers and onCommit hook to schedule the second phase of drop to delete
+ * the collection data.
+ */
+Status dropCollection(OperationContext* opCtx,
+                      const NamespaceString& nss,
+                      RecordId collectionCatalogId,
+                      StringData ident);
+
+
+}  // namespace catalog
+}  // namespace mongo