SERVER-65559 Get rid of legacy range deleter and RangeDeleterService feature flag

25 files changed, 39 insertions, 2034 deletions

diff --git a/src/mongo/db/s/collection_metadata_filtering_test.cpp b/src/mongo/db/s/collection_metadata_filtering_test.cpp
index aae08d987eb..57dfe2800c7 100644
--- a/src/mongo/db/s/collection_metadata_filtering_test.cpp
+++ b/src/mongo/db/s/collection_metadata_filtering_test.cpp
@@ -127,7 +127,7 @@ protected:
         }
 
         _manager = std::make_shared<MetadataManager>(
-            getServiceContext(), kNss, executor(), CollectionMetadata(cm, ShardId("0")));
+            getServiceContext(), kNss, CollectionMetadata(cm, ShardId("0")));
 
         return CollectionMetadata(std::move(cm), ShardId("0"));
     }
diff --git a/src/mongo/db/s/collection_sharding_runtime.cpp b/src/mongo/db/s/collection_sharding_runtime.cpp
index f5f5b01d6d9..3de77f88a9e 100644
--- a/src/mongo/db/s/collection_sharding_runtime.cpp
+++ b/src/mongo/db/s/collection_sharding_runtime.cpp
@@ -93,13 +93,9 @@ CollectionShardingRuntime::ScopedExclusiveCollectionShardingRuntime::
     ScopedExclusiveCollectionShardingRuntime(ScopedCollectionShardingState&& scopedCss)
     : _scopedCss(std::move(scopedCss)) {}
 
-CollectionShardingRuntime::CollectionShardingRuntime(
-    ServiceContext* service,
-    NamespaceString nss,
-    std::shared_ptr<executor::TaskExecutor> rangeDeleterExecutor)
+CollectionShardingRuntime::CollectionShardingRuntime(ServiceContext* service, NamespaceString nss)
     : _serviceContext(service),
       _nss(std::move(nss)),
-      _rangeDeleterExecutor(std::move(rangeDeleterExecutor)),
       _metadataType(_nss.isNamespaceAlwaysUnsharded() ? MetadataType::kUnsharded
                                                       : MetadataType::kUnknown) {}
 
@@ -277,8 +273,8 @@ void CollectionShardingRuntime::setFilteringMetadata(OperationContext* opCtx,
     _metadataType = MetadataType::kSharded;
 
     if (!_metadataManager || !newMetadata.uuidMatches(_metadataManager->getCollectionUuid())) {
-        _metadataManager = std::make_shared<MetadataManager>(
-            opCtx->getServiceContext(), _nss, _rangeDeleterExecutor, newMetadata);
+        _metadataManager =
+            std::make_shared<MetadataManager>(opCtx->getServiceContext(), _nss, newMetadata);
         ++_numMetadataManagerChanges;
     } else {
         _metadataManager->setFilteringMetadata(std::move(newMetadata));
@@ -319,20 +315,6 @@ void CollectionShardingRuntime::clearFilteringMetadataForDroppedCollection(
     _clearFilteringMetadata(opCtx, /* collIsDropped */ true);
 }
 
-SharedSemiFuture<void> CollectionShardingRuntime::cleanUpRange(ChunkRange const& range,
-                                                               CleanWhen when) const {
-    // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-    // flag is used to turn on new range deleter on startup.
-    if (!feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-        stdx::lock_guard lk(_metadataManagerLock);
-        invariant(_metadataType == MetadataType::kSharded);
-        return _metadataManager->cleanUpRange(range, when == kDelayed);
-    }
-
-    // This method must never be called if the range deleter service feature flag is enabled
-    MONGO_UNREACHABLE;
-}
-
 Status CollectionShardingRuntime::waitForClean(OperationContext* opCtx,
                                                const NamespaceString& nss,
                                                const UUID& collectionUuid,
@@ -355,17 +337,11 @@ Status CollectionShardingRuntime::waitForClean(OperationContext* opCtx,
                         "metadata reset"};
             }
 
-            // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The
-            // feature flag is used to turn on new range deleter on startup.
-            if (feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-                const auto rangeDeleterService = RangeDeleterService::get(opCtx);
-                rangeDeleterService->getRangeDeleterServiceInitializationFuture().get(opCtx);
+            const auto rangeDeleterService = RangeDeleterService::get(opCtx);
+            rangeDeleterService->getRangeDeleterServiceInitializationFuture().get(opCtx);
 
-                return rangeDeleterService->getOverlappingRangeDeletionsFuture(
-                    self->_metadataManager->getCollectionUuid(), orphanRange);
-            } else {
-                return self->_metadataManager->trackOrphanedDataCleanup(orphanRange);
-            }
+            return rangeDeleterService->getOverlappingRangeDeletionsFuture(
+                self->_metadataManager->getCollectionUuid(), orphanRange);
         }();
 
         if (!swOrphanCleanupFuture.isOK()) {
@@ -568,15 +544,6 @@ void CollectionShardingRuntime::appendShardVersion(BSONObjBuilder* builder) cons
     }
 }
 
-size_t CollectionShardingRuntime::numberOfRangesScheduledForDeletion() const {
-    stdx::lock_guard lk(_metadataManagerLock);
-    if (_metadataType == MetadataType::kSharded) {
-        return _metadataManager->numberOfRangesScheduledForDeletion();
-    }
-    return 0;
-}
-
-
 void CollectionShardingRuntime::setPlacementVersionRecoverRefreshFuture(
     SharedSemiFuture<void> future, CancellationSource cancellationSource) {
     invariant(!_placementVersionInRecoverOrRefresh);
diff --git a/src/mongo/db/s/collection_sharding_runtime.h b/src/mongo/db/s/collection_sharding_runtime.h
index 6da67980fa3..696b0e6497b 100644
--- a/src/mongo/db/s/collection_sharding_runtime.h
+++ b/src/mongo/db/s/collection_sharding_runtime.h
@@ -51,9 +51,7 @@ class CollectionShardingRuntime final : public CollectionShardingState,
     CollectionShardingRuntime& operator=(const CollectionShardingRuntime&) = delete;
 
 public:
-    CollectionShardingRuntime(ServiceContext* service,
-                              NamespaceString nss,
-                              std::shared_ptr<executor::TaskExecutor> rangeDeleterExecutor);
+    CollectionShardingRuntime(ServiceContext* service, NamespaceString nss);
 
     /**
      * Obtains the sharding runtime for the specified collection, along with a resource lock in
@@ -143,8 +141,6 @@ public:
 
     void appendShardVersion(BSONObjBuilder* builder) const override;
 
-    size_t numberOfRangesScheduledForDeletion() const override;
-
     boost::optional<ShardingIndexesCatalogCache> getIndexesInCritSec(OperationContext* opCtx) const;
 
     /**
@@ -219,18 +215,6 @@ public:
         OperationContext* opCtx, ShardingMigrationCriticalSection::Operation op) const;
 
     /**
-     * Schedules documents in `range` for cleanup after any running queries that may depend on them
-     * have terminated. Does not block. Fails if range overlaps any current local shard chunk.
-     * Passed kDelayed, an additional delay (configured via server parameter orphanCleanupDelaySecs)
-     * is added to permit (most) dependent queries on secondaries to complete, too.
-     *
-     * Returns a future that will be resolved when the deletion completes or fails. If that
-     * succeeds, waitForClean can be called to ensure no other deletions are pending for the range.
-     */
-    enum CleanWhen { kNow, kDelayed };
-    SharedSemiFuture<void> cleanUpRange(ChunkRange const& range, CleanWhen when) const;
-
-    /**
      * Waits for all ranges deletion tasks with UUID 'collectionUuid' overlapping range
      * 'orphanRange' to be processed, even if the collection does not exist in the storage catalog.
      * It will block until the minimum of the operation context's timeout deadline or 'deadline' is
@@ -353,9 +337,6 @@ private:
     // Namespace this state belongs to.
     const NamespaceString _nss;
 
-    // The executor used for deleting ranges of orphan chunks.
-    std::shared_ptr<executor::TaskExecutor> _rangeDeleterExecutor;
-
     // Tracks the migration critical section state for this collection.
     ShardingMigrationCriticalSection _critSec;
 
diff --git a/src/mongo/db/s/collection_sharding_runtime_test.cpp b/src/mongo/db/s/collection_sharding_runtime_test.cpp
index ea934294256..b432473d13d 100644
--- a/src/mongo/db/s/collection_sharding_runtime_test.cpp
+++ b/src/mongo/db/s/collection_sharding_runtime_test.cpp
@@ -97,7 +97,7 @@ protected:
 TEST_F(CollectionShardingRuntimeTest,
        GetCollectionDescriptionThrowsStaleConfigBeforeSetFilteringMetadataIsCalledAndNoOSSSet) {
     OperationContext* opCtx = operationContext();
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     ASSERT_FALSE(csr.getCollectionDescription(opCtx).isSharded());
     auto metadata = makeShardedMetadata(opCtx);
     ScopedSetShardRole scopedSetShardRole{
@@ -111,14 +111,14 @@ TEST_F(CollectionShardingRuntimeTest,
 TEST_F(
     CollectionShardingRuntimeTest,
     GetCollectionDescriptionReturnsUnshardedAfterSetFilteringMetadataIsCalledWithUnshardedMetadata) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     csr.setFilteringMetadata(operationContext(), CollectionMetadata());
     ASSERT_FALSE(csr.getCollectionDescription(operationContext()).isSharded());
 }
 
 TEST_F(CollectionShardingRuntimeTest,
        GetCollectionDescriptionReturnsShardedAfterSetFilteringMetadataIsCalledWithShardedMetadata) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadata(opCtx);
     csr.setFilteringMetadata(opCtx, metadata);
@@ -132,14 +132,14 @@ TEST_F(CollectionShardingRuntimeTest,
 
 TEST_F(CollectionShardingRuntimeTest,
        GetCurrentMetadataIfKnownReturnsNoneBeforeSetFilteringMetadataIsCalled) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     ASSERT_FALSE(csr.getCurrentMetadataIfKnown());
 }
 
 TEST_F(
     CollectionShardingRuntimeTest,
     GetCurrentMetadataIfKnownReturnsUnshardedAfterSetFilteringMetadataIsCalledWithUnshardedMetadata) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     csr.setFilteringMetadata(operationContext(), CollectionMetadata());
     const auto optCurrMetadata = csr.getCurrentMetadataIfKnown();
     ASSERT_TRUE(optCurrMetadata);
@@ -150,7 +150,7 @@ TEST_F(
 TEST_F(
     CollectionShardingRuntimeTest,
     GetCurrentMetadataIfKnownReturnsShardedAfterSetFilteringMetadataIsCalledWithShardedMetadata) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadata(opCtx);
     csr.setFilteringMetadata(opCtx, metadata);
@@ -162,7 +162,7 @@ TEST_F(
 
 TEST_F(CollectionShardingRuntimeTest,
        GetCurrentMetadataIfKnownReturnsNoneAfterClearFilteringMetadataIsCalled) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     OperationContext* opCtx = operationContext();
     csr.setFilteringMetadata(opCtx, makeShardedMetadata(opCtx));
     csr.clearFilteringMetadata(opCtx);
@@ -170,7 +170,7 @@ TEST_F(CollectionShardingRuntimeTest,
 }
 
 TEST_F(CollectionShardingRuntimeTest, SetFilteringMetadataWithSameUUIDKeepsSameMetadataManager) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     ASSERT_EQ(csr.getNumMetadataManagerChanges_forTest(), 0);
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadata(opCtx);
@@ -185,7 +185,7 @@ TEST_F(CollectionShardingRuntimeTest, SetFilteringMetadataWithSameUUIDKeepsSameM
 
 TEST_F(CollectionShardingRuntimeTest,
        SetFilteringMetadataWithDifferentUUIDReplacesPreviousMetadataManager) {
-    CollectionShardingRuntime csr(getServiceContext(), kTestNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), kTestNss);
     OperationContext* opCtx = operationContext();
     auto metadata = makeShardedMetadata(opCtx);
     csr.setFilteringMetadata(opCtx, metadata);
@@ -226,7 +226,7 @@ TEST_F(CollectionShardingRuntimeTest, ReturnUnshardedMetadataInServerlessMode) {
         boost::none                /* databaseVersion */
     };
 
-    CollectionShardingRuntime csr(getServiceContext(), testNss, executor());
+    CollectionShardingRuntime csr(getServiceContext(), testNss);
     auto collectionFilter = csr.getOwnershipFilter(
         opCtx, CollectionShardingRuntime::OrphanCleanupPolicy::kAllowOrphanCleanup, true);
     ASSERT_FALSE(collectionFilter.isSharded());
@@ -244,8 +244,8 @@ TEST_F(CollectionShardingRuntimeTest, ReturnUnshardedMetadataInServerlessMode) {
         boost::none                                           /* databaseVersion */
     };
 
-    CollectionShardingRuntime csrLogicalSession(
-        getServiceContext(), NamespaceString::kLogicalSessionsNamespace, executor());
+    CollectionShardingRuntime csrLogicalSession(getServiceContext(),
+                                                NamespaceString::kLogicalSessionsNamespace);
     ASSERT(csrLogicalSession.getCurrentMetadataIfKnown() == boost::none);
     ASSERT_THROWS_CODE(
         csrLogicalSession.getCollectionDescription(opCtx), DBException, ErrorCodes::StaleConfig);
diff --git a/src/mongo/db/s/collection_sharding_state.cpp b/src/mongo/db/s/collection_sharding_state.cpp
index eca22b162d8..eebfb2dc675 100644
--- a/src/mongo/db/s/collection_sharding_state.cpp
+++ b/src/mongo/db/s/collection_sharding_state.cpp
@@ -30,7 +30,6 @@
 #include "mongo/db/s/collection_sharding_state.h"
 
 #include "mongo/logv2/log.h"
-#include "mongo/s/sharding_feature_flags_gen.h"
 #include "mongo/util/string_map.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
@@ -91,25 +90,6 @@ public:
         versionB.done();
     }
 
-    void appendInfoForServerStatus(BSONObjBuilder* builder) {
-        // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-        // flag is used to turn on new range deleter on startup.
-        if (!mongo::feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-            auto totalNumberOfRangesScheduledForDeletion = ([this] {
-                stdx::lock_guard lg(_mutex);
-                return std::accumulate(
-                    _collections.begin(),
-                    _collections.end(),
-                    0LL,
-                    [](long long total, const auto& coll) {
-                        return total + coll.second->css->numberOfRangesScheduledForDeletion();
-                    });
-            })();
-
-            builder->appendNumber("rangeDeleterTasks", totalNumberOfRangesScheduledForDeletion);
-        }
-    }
-
     std::vector<NamespaceString> getCollectionNames() {
         stdx::lock_guard lg(_mutex);
         std::vector<NamespaceString> result;
@@ -190,12 +170,6 @@ void CollectionShardingState::appendInfoForShardingStateCommand(OperationContext
     collectionsMap->appendInfoForShardingStateCommand(builder);
 }
 
-void CollectionShardingState::appendInfoForServerStatus(OperationContext* opCtx,
-                                                        BSONObjBuilder* builder) {
-    auto& collectionsMap = CollectionShardingStateMap::get(opCtx->getServiceContext());
-    collectionsMap->appendInfoForServerStatus(builder);
-}
-
 std::vector<NamespaceString> CollectionShardingState::getCollectionNames(OperationContext* opCtx) {
     auto& collectionsMap = CollectionShardingStateMap::get(opCtx->getServiceContext());
     return collectionsMap->getCollectionNames();
diff --git a/src/mongo/db/s/collection_sharding_state.h b/src/mongo/db/s/collection_sharding_state.h
index 27a8da745b1..45cca1b50e4 100644
--- a/src/mongo/db/s/collection_sharding_state.h
+++ b/src/mongo/db/s/collection_sharding_state.h
@@ -116,11 +116,6 @@ public:
     static void appendInfoForShardingStateCommand(OperationContext* opCtx, BSONObjBuilder* builder);
 
     /**
-     * Attaches info for server status.
-     */
-    static void appendInfoForServerStatus(OperationContext* opCtx, BSONObjBuilder* builder);
-
-    /**
      * Returns the namespace to which this CSS corresponds.
      */
     virtual const NamespaceString& nss() const = 0;
@@ -207,11 +202,6 @@ public:
      * Appends information about the shard version of the collection.
      */
     virtual void appendShardVersion(BSONObjBuilder* builder) const = 0;
-
-    /**
-     * Returns the number of ranges scheduled for deletion on the collection.
-     */
-    virtual size_t numberOfRangesScheduledForDeletion() const = 0;
 };
 
 /**
diff --git a/src/mongo/db/s/collection_sharding_state_factory_shard.cpp b/src/mongo/db/s/collection_sharding_state_factory_shard.cpp
index 0dd89232c3d..7acf184457a 100644
--- a/src/mongo/db/s/collection_sharding_state_factory_shard.cpp
+++ b/src/mongo/db/s/collection_sharding_state_factory_shard.cpp
@@ -27,67 +27,19 @@
  *    it in the license file.
  */
 
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/collection_sharding_state_factory_shard.h"
-#include "mongo/db/service_context.h"
-#include "mongo/executor/network_interface_factory.h"
-#include "mongo/executor/network_interface_thread_pool.h"
-#include "mongo/executor/thread_pool_task_executor.h"
-#include "mongo/s/sharding_feature_flags_gen.h"
+#include "mongo/db/s/collection_sharding_runtime.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
 
-
 namespace mongo {
 CollectionShardingStateFactoryShard::CollectionShardingStateFactoryShard(
     ServiceContext* serviceContext)
     : CollectionShardingStateFactory(serviceContext) {}
 
-CollectionShardingStateFactoryShard::~CollectionShardingStateFactoryShard() {
-    join();
-}
-
-void CollectionShardingStateFactoryShard::join() {
-    if (_rangeDeletionExecutor) {
-        _rangeDeletionExecutor->shutdown();
-        _rangeDeletionExecutor->join();
-    }
-}
-
 std::unique_ptr<CollectionShardingState> CollectionShardingStateFactoryShard::make(
     const NamespaceString& nss) {
-    return std::make_unique<CollectionShardingRuntime>(
-        _serviceContext, nss, _getRangeDeletionExecutor());
-}
-
-std::shared_ptr<executor::TaskExecutor>
-CollectionShardingStateFactoryShard::_getRangeDeletionExecutor() {
-    // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-    // flag is used to turn on new range deleter on startup.
-    if (feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-        return nullptr;
-    }
-
-    stdx::lock_guard<Latch> lg(_mutex);
-    if (!_rangeDeletionExecutor) {
-        const std::string kExecName("CollectionRangeDeleter-TaskExecutor");
-
-        // CAUTION: The safety of range deletion depends on using a task executor that schedules
-        // work on a single thread.
-        auto net = executor::makeNetworkInterface(kExecName);
-        auto pool = std::make_unique<executor::NetworkInterfaceThreadPool>(net.get());
-        auto taskExecutor =
-            std::make_shared<executor::ThreadPoolTaskExecutor>(std::move(pool), std::move(net));
-        taskExecutor->startup();
-
-        _rangeDeletionExecutor = std::move(taskExecutor);
-    }
-
-    return _rangeDeletionExecutor;
+    return std::make_unique<CollectionShardingRuntime>(_serviceContext, nss);
 }
 
-
 }  // namespace mongo
diff --git a/src/mongo/db/s/collection_sharding_state_factory_shard.h b/src/mongo/db/s/collection_sharding_state_factory_shard.h
index a3a0c04e82a..4060de88ee0 100644
--- a/src/mongo/db/s/collection_sharding_state_factory_shard.h
+++ b/src/mongo/db/s/collection_sharding_state_factory_shard.h
@@ -30,7 +30,6 @@
 #pragma once
 
 #include "mongo/db/s/collection_sharding_state.h"
-#include "mongo/executor/task_executor.h"
 
 namespace mongo {
 
@@ -38,20 +37,9 @@ class CollectionShardingStateFactoryShard final : public CollectionShardingState
 public:
     CollectionShardingStateFactoryShard(ServiceContext* serviceContext);
 
-    ~CollectionShardingStateFactoryShard();
-
-    void join() override;
+    void join() override{};
 
     std::unique_ptr<CollectionShardingState> make(const NamespaceString& nss) override;
-
-private:
-    std::shared_ptr<executor::TaskExecutor> _getRangeDeletionExecutor();
-
-    // Serializes the instantiation of the task executor
-    Mutex _mutex = MONGO_MAKE_LATCH("CollectionShardingStateFactoryShard::_mutex");
-
-    // Required to be a shared_ptr since it is used as an executor for ExecutorFutures.
-    std::shared_ptr<executor::TaskExecutor> _rangeDeletionExecutor = {nullptr};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp b/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp
index 48fed34a989..f9a858ae655 100644
--- a/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp
+++ b/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp
@@ -92,10 +92,6 @@ public:
 
     void appendShardVersion(BSONObjBuilder* builder) const override {}
 
-    size_t numberOfRangesScheduledForDeletion() const override {
-        return 0;
-    }
-
 private:
     const NamespaceString& _nss;
 };
diff --git a/src/mongo/db/s/metadata_manager.cpp b/src/mongo/db/s/metadata_manager.cpp
index ba3fc1e25bd..7859124afd4 100644
--- a/src/mongo/db/s/metadata_manager.cpp
+++ b/src/mongo/db/s/metadata_manager.cpp
@@ -32,20 +32,14 @@
 #include "mongo/base/string_data.h"
 #include "mongo/bson/util/builder.h"
 #include "mongo/db/s/migration_util.h"
-#include "mongo/db/s/range_deleter_service.h"
-#include "mongo/db/s/range_deletion_util.h"
 #include "mongo/db/s/sharding_runtime_d_params_gen.h"
 #include "mongo/logv2/log.h"
-#include "mongo/s/sharding_feature_flags_gen.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
 
 namespace mongo {
 namespace {
 
-using TaskExecutor = executor::TaskExecutor;
-using CallbackArgs = TaskExecutor::CallbackArgs;
-
 /**
  * Returns whether the given metadata object has a chunk owned by this shard that overlaps the
  * input range.
@@ -107,12 +101,10 @@ private:
 
 MetadataManager::MetadataManager(ServiceContext* serviceContext,
                                  NamespaceString nss,
-                                 std::shared_ptr<TaskExecutor> executor,
                                  CollectionMetadata initialMetadata)
     : _serviceContext(serviceContext),
       _nss(std::move(nss)),
-      _collectionUuid(initialMetadata.getChunkManager()->getUUID()),
-      _executor(std::move(executor)) {
+      _collectionUuid(initialMetadata.getChunkManager()->getUUID()) {
     _metadata.emplace_back(std::make_shared<CollectionMetadataTracker>(std::move(initialMetadata)));
 }
 
@@ -230,106 +222,6 @@ void MetadataManager::_retireExpiredMetadata(WithLock) {
     }
 }
 
-void MetadataManager::append(BSONObjBuilder* builder) const {
-    stdx::lock_guard<Latch> lg(_managerLock);
-
-    BSONArrayBuilder arr(builder->subarrayStart("rangesToClean"));
-    for (auto const& [range, _] : _rangesScheduledForDeletion) {
-        BSONObjBuilder obj;
-        range.append(&obj);
-        arr.append(obj.done());
-    }
-
-    invariant(!_metadata.empty());
-
-    BSONArrayBuilder amrArr(builder->subarrayStart("activeMetadataRanges"));
-    for (const auto& entry : _metadata.back()->metadata->getChunks()) {
-        BSONObjBuilder obj;
-        ChunkRange r = ChunkRange(entry.first, entry.second);
-        r.append(&obj);
-        amrArr.append(obj.done());
-    }
-    amrArr.done();
-}
-
-SharedSemiFuture<void> MetadataManager::cleanUpRange(ChunkRange const& range,
-                                                     bool shouldDelayBeforeDeletion) {
-    stdx::lock_guard<Latch> lg(_managerLock);
-    invariant(!_metadata.empty());
-
-    auto* const activeMetadata = _metadata.back().get();
-    auto* const overlapMetadata = _findNewestOverlappingMetadata(lg, range);
-
-    if (overlapMetadata == activeMetadata) {
-        return Status{ErrorCodes::RangeOverlapConflict,
-                      str::stream() << "Requested deletion range overlaps a live shard chunk"};
-    }
-
-    auto delayForActiveQueriesOnSecondariesToComplete =
-        shouldDelayBeforeDeletion ? Seconds(orphanCleanupDelaySecs.load()) : Seconds(0);
-
-    if (overlapMetadata) {
-        LOGV2_OPTIONS(21989,
-                      {logv2::LogComponent::kShardingMigration},
-                      "Deletion of {namespace} range {range} will be scheduled after all possibly "
-                      "dependent queries finish",
-                      "Deletion of the collection's specified range will be scheduled after all "
-                      "possibly dependent queries finish",
-                      logAttrs(_nss),
-                      "range"_attr = redact(range.toString()));
-        ++overlapMetadata->numContingentRangeDeletionTasks;
-        // Schedule the range for deletion once the overlapping metadata object is destroyed
-        // (meaning no more queries can be using the range) and obtain a future which will be
-        // signaled when deletion is complete.
-        return _submitRangeForDeletion(lg,
-                                       overlapMetadata->onDestructionPromise.getFuture().semi(),
-                                       range,
-                                       delayForActiveQueriesOnSecondariesToComplete);
-    } else {
-        // No running queries can depend on this range, so queue it for deletion immediately.
-        LOGV2_OPTIONS(21990,
-                      {logv2::LogComponent::kShardingMigration},
-                      "Scheduling deletion of {namespace} range {range}",
-                      "Scheduling deletion of the collection's specified range",
-                      logAttrs(_nss),
-                      "range"_attr = redact(range.toString()));
-
-        return _submitRangeForDeletion(
-            lg, SemiFuture<void>::makeReady(), range, delayForActiveQueriesOnSecondariesToComplete);
-    }
-}
-
-size_t MetadataManager::numberOfRangesToCleanStillInUse() const {
-    stdx::lock_guard<Latch> lg(_managerLock);
-    size_t count = 0;
-    for (auto& tracker : _metadata) {
-        count += tracker->numContingentRangeDeletionTasks;
-    }
-    return count;
-}
-
-size_t MetadataManager::numberOfRangesToClean() const {
-    auto rangesToCleanInUse = numberOfRangesToCleanStillInUse();
-    stdx::lock_guard<Latch> lg(_managerLock);
-    return _rangesScheduledForDeletion.size() - rangesToCleanInUse;
-}
-
-size_t MetadataManager::numberOfRangesScheduledForDeletion() const {
-    stdx::lock_guard<Latch> lg(_managerLock);
-    return _rangesScheduledForDeletion.size();
-}
-
-SharedSemiFuture<void> MetadataManager::trackOrphanedDataCleanup(ChunkRange const& range) const {
-    stdx::lock_guard<Latch> lg(_managerLock);
-    for (const auto& [orphanRange, deletionComplete] : _rangesScheduledForDeletion) {
-        if (orphanRange.overlapWith(range)) {
-            return deletionComplete;
-        }
-    }
-
-    return SemiFuture<void>::makeReady().share();
-}
-
 SharedSemiFuture<void> MetadataManager::getOngoingQueriesCompletionFuture(ChunkRange const& range) {
     stdx::lock_guard<Latch> lg(_managerLock);
 
@@ -354,47 +246,4 @@ auto MetadataManager::_findNewestOverlappingMetadata(WithLock, ChunkRange const&
     return nullptr;
 }
 
-bool MetadataManager::_overlapsInUseChunk(WithLock lk, ChunkRange const& range) {
-    auto* cm = _findNewestOverlappingMetadata(lk, range);
-    return (cm != nullptr);
-}
-
-SharedSemiFuture<void> MetadataManager::_submitRangeForDeletion(
-    const WithLock&,
-    SemiFuture<void> waitForActiveQueriesToComplete,
-    const ChunkRange& range,
-    Seconds delayForActiveQueriesOnSecondariesToComplete) {
-    auto cleanupComplete = [&]() {
-        const auto collUUID = _metadata.back()->metadata->getChunkManager()->getUUID();
-
-        // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-        // flag is used to turn on new range deleter on startup.
-        if (feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-            return RangeDeleterService::get(_serviceContext)
-                ->getOverlappingRangeDeletionsFuture(collUUID, range);
-        }
-
-        return removeDocumentsInRange(_executor,
-                                      std::move(waitForActiveQueriesToComplete),
-                                      _nss,
-                                      collUUID,
-                                      _metadata.back()->metadata->getKeyPattern().getOwned(),
-                                      range,
-                                      delayForActiveQueriesOnSecondariesToComplete);
-    }();
-
-    _rangesScheduledForDeletion.emplace_front(range, cleanupComplete);
-    // Attach a continuation so that once the range has been deleted, we will remove the deletion
-    // from the _rangesScheduledForDeletion.  std::list iterators are never invalidated, which
-    // allows us to save the iterator pointing to the newly added element for use later when
-    // deleting it.
-    cleanupComplete.thenRunOn(_executor).getAsync(
-        [self = shared_from_this(), it = _rangesScheduledForDeletion.begin()](Status s) {
-            stdx::lock_guard<Latch> lg(self->_managerLock);
-            self->_rangesScheduledForDeletion.erase(it);
-        });
-
-    return cleanupComplete;
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/s/metadata_manager.h b/src/mongo/db/s/metadata_manager.h
index d2477c4fe06..99edec31d56 100644
--- a/src/mongo/db/s/metadata_manager.h
+++ b/src/mongo/db/s/metadata_manager.h
@@ -34,7 +34,6 @@
 #include "mongo/db/logical_time.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/s/scoped_collection_metadata.h"
-#include "mongo/executor/task_executor.h"
 #include "mongo/s/catalog/type_chunk.h"
 #include "mongo/util/concurrency/with_lock.h"
 
@@ -49,7 +48,6 @@ class MetadataManager : public std::enable_shared_from_this<MetadataManager> {
 public:
     MetadataManager(ServiceContext* serviceContext,
                     NamespaceString nss,
-                    std::shared_ptr<executor::TaskExecutor> executor,
                     CollectionMetadata initialMetadata);
     ~MetadataManager() = default;
 
@@ -101,50 +99,6 @@ public:
     void setFilteringMetadata(CollectionMetadata newMetadata);
 
     /**
-     * Appends information on all the chunk ranges in rangesToClean to builder.
-     */
-    void append(BSONObjBuilder* builder) const;
-
-    /**
-     * Schedules documents in `range` for cleanup after any running queries that may depend on them
-     * have terminated. Does not block. Fails if the range overlaps any current local shard chunk.
-     *
-     * If shouldDelayBeforeDeletion is false, deletion is scheduled immediately after the last
-     * dependent query completes; otherwise, deletion is postponed until after
-     * orphanCleanupDelaySecs after the last dependent query completes.
-     *
-     * Returns a future that will be fulfilled when the range deletion completes or fails.
-     */
-    SharedSemiFuture<void> cleanUpRange(ChunkRange const& range, bool shouldDelayBeforeDeletion);
-
-    /**
-     * Returns the number of ranges scheduled to be cleaned, exclusive of such ranges that might
-     * still be in use by running queries.  Outside of test drivers, the actual number may vary
-     * after it returns, so this is really only useful for unit tests.
-     */
-    size_t numberOfRangesToClean() const;
-
-    /**
-     * Returns the number of ranges scheduled to be cleaned once all queries that could depend on
-     * them have terminated. The actual number may vary after it returns, so this is really only
-     * useful for unit tests.
-     */
-    size_t numberOfRangesToCleanStillInUse() const;
-
-    /**
-     * Returns the number of ranges scheduled for deletion, regardless of whether they may still be
-     * in use by running queries.
-     */
-    size_t numberOfRangesScheduledForDeletion() const;
-
-    /**
-     * Reports whether any range still scheduled for deletion overlaps the argument range. If so,
-     * returns a future that will be resolved when the newest overlapping range's deletion (possibly
-     * the one of interest) completes or fails.
-     */
-    SharedSemiFuture<void> trackOrphanedDataCleanup(ChunkRange const& orphans) const;
-
-    /**
      * Returns a future marked as ready when all the ongoing queries retaining the range complete
      */
     SharedSemiFuture<void> getOngoingQueriesCompletionFuture(ChunkRange const& range);
@@ -172,12 +126,6 @@ private:
         boost::optional<CollectionMetadata> metadata;
 
         /**
-         * Number of range deletion tasks waiting on this CollectionMetadataTracker to be destroyed
-         * before deleting documents.
-         */
-        uint32_t numContingentRangeDeletionTasks{0};
-
-        /**
          * Promise that will be signaled when this object is destroyed.
          *
          * In the case where this CollectionMetadataTracker may refer to orphaned documents for one
@@ -208,24 +156,6 @@ private:
      */
     CollectionMetadataTracker* _findNewestOverlappingMetadata(WithLock, ChunkRange const& range);
 
-    /**
-     * Returns true if the specified range overlaps any chunk that is currently in use by a running
-     * query.
-     */
-
-    bool _overlapsInUseChunk(WithLock, ChunkRange const& range);
-
-    /**
-     * Schedule a task to delete the given range of documents once waitForActiveQueriesToComplete
-     * has been signaled, and store the resulting future for the task in
-     * _rangesScheduledForDeletion.
-     */
-    SharedSemiFuture<void> _submitRangeForDeletion(
-        const WithLock&,
-        SemiFuture<void> waitForActiveQueriesToComplete,
-        const ChunkRange& range,
-        Seconds delayForActiveQueriesOnSecondariesToComplete);
-
     // ServiceContext from which to obtain instances of global support objects
     ServiceContext* const _serviceContext;
 
@@ -235,9 +165,6 @@ private:
     // The UUID for the collection tracked by this manager object.
     const UUID _collectionUuid;
 
-    // The background task that deletes documents from orphaned chunk ranges.
-    std::shared_ptr<executor::TaskExecutor> const _executor;
-
     // Mutex to protect the state below
     mutable Mutex _managerLock = MONGO_MAKE_LATCH("MetadataManager::_managerLock");
 
@@ -246,9 +173,6 @@ private:
     // the most recent metadata and is what is returned to new queries. The rest are previously
     // active collection metadata instances still in use by active server operations or cursors.
     std::list<std::shared_ptr<CollectionMetadataTracker>> _metadata;
-
-    // Ranges being deleted, or scheduled to be deleted, by a background task.
-    std::list<std::pair<ChunkRange, SharedSemiFuture<void>>> _rangesScheduledForDeletion;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/metadata_manager_test.cpp b/src/mongo/db/s/metadata_manager_test.cpp
index e8c2d3154ac..a8411252ded 100644
--- a/src/mongo/db/s/metadata_manager_test.cpp
+++ b/src/mongo/db/s/metadata_manager_test.cpp
@@ -66,8 +66,8 @@ class MetadataManagerTest : public ShardServerTestFixture {
 protected:
     void setUp() override {
         ShardServerTestFixture::setUp();
-        _manager = std::make_shared<MetadataManager>(
-            getServiceContext(), kNss, executor(), makeEmptyMetadata());
+        _manager =
+            std::make_shared<MetadataManager>(getServiceContext(), kNss, makeEmptyMetadata());
         orphanCleanupDelaySecs.store(1);
     }
 
@@ -196,101 +196,6 @@ private:
     const int _defaultOrphanCleanupDelaySecs = orphanCleanupDelaySecs.load();
 };
 
-// The 'pending' field must not be set in order for a range deletion task to succeed, but the
-// ShardServerOpObserver will submit the task for deletion upon seeing an insert without the
-// 'pending' field. The tests call removeDocumentsFromRange directly, so we want to avoid having
-// the op observer also submit the task. The ShardServerOpObserver will ignore replacement
-//  updates on the range deletions namespace though, so we can get around the issue by inserting
-// the task with the 'pending' field set, and then remove the field using a replacement update
-// after.
-RangeDeletionTask insertRangeDeletionTask(OperationContext* opCtx,
-                                          const NamespaceString& nss,
-                                          const UUID& uuid,
-                                          const ChunkRange& range,
-                                          int64_t numOrphans) {
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-    auto migrationId = UUID::gen();
-    RangeDeletionTask t(migrationId, nss, uuid, ShardId("donor"), range, CleanWhenEnum::kDelayed);
-    t.setPending(true);
-    t.setNumOrphanDocs(numOrphans);
-    const auto currentTime = VectorClock::get(opCtx)->getTime();
-    t.setTimestamp(currentTime.clusterTime().asTimestamp());
-    store.add(opCtx, t);
-
-    auto query = BSON(RangeDeletionTask::kIdFieldName << migrationId);
-    t.setPending(boost::none);
-    auto update = t.toBSON();
-    store.update(opCtx, query, update);
-
-    return t;
-}
-
-TEST_F(MetadataManagerTest, TrackOrphanedDataCleanupBlocksOnScheduledRangeDeletions) {
-    RAIIServerParameterControllerForTest enableFeatureFlag{"featureFlagRangeDeleterService", false};
-    ChunkRange cr1(BSON("key" << 0), BSON("key" << 10));
-    const auto task =
-        insertRangeDeletionTask(operationContext(), kNss, _manager->getCollectionUuid(), cr1, 0);
-
-    // Enable fail point to suspendRangeDeletion.
-    globalFailPointRegistry().find("suspendRangeDeletion")->setMode(FailPoint::alwaysOn);
-
-    auto notifn1 = _manager->cleanUpRange(cr1, false /*delayBeforeDeleting*/);
-    ASSERT_FALSE(notifn1.isReady());
-    ASSERT_EQ(_manager->numberOfRangesToClean(), 1UL);
-
-    auto future = _manager->trackOrphanedDataCleanup(cr1);
-    ASSERT_FALSE(notifn1.isReady());
-    ASSERT_FALSE(future.isReady());
-
-    globalFailPointRegistry().find("suspendRangeDeletion")->setMode(FailPoint::off);
-}
-
-TEST_F(MetadataManagerTest, CleanupNotificationsAreSignaledWhenMetadataManagerIsDestroyed) {
-    RAIIServerParameterControllerForTest enableFeatureFlag{"featureFlagRangeDeleterService", false};
-    const ChunkRange rangeToClean(BSON("key" << 20), BSON("key" << 30));
-    const auto task = insertRangeDeletionTask(
-        operationContext(), kNss, _manager->getCollectionUuid(), rangeToClean, 0);
-
-
-    _manager->setFilteringMetadata(cloneMetadataPlusChunk(
-        _manager->getActiveMetadata(boost::none)->get(), {BSON("key" << 0), BSON("key" << 20)}));
-
-    _manager->setFilteringMetadata(
-        cloneMetadataPlusChunk(_manager->getActiveMetadata(boost::none)->get(), rangeToClean));
-
-    // Optional so that it can be reset.
-    boost::optional<ScopedCollectionDescription> cursorOnMovedMetadata{
-        _manager->getActiveMetadata(boost::none)};
-
-    _manager->setFilteringMetadata(
-        cloneMetadataMinusChunk(_manager->getActiveMetadata(boost::none)->get(), rangeToClean));
-
-    auto notif = _manager->cleanUpRange(rangeToClean, false /*delayBeforeDeleting*/);
-    ASSERT(!notif.isReady());
-
-    auto future = _manager->trackOrphanedDataCleanup(rangeToClean);
-    ASSERT(!future.isReady());
-
-    // Reset the original shared_ptr. The cursorOnMovedMetadata will still contain its own copy of
-    // the shared_ptr though, so the destructor of ~MetadataManager won't yet be called.
-    _manager.reset();
-    ASSERT(!notif.isReady());
-    ASSERT(!future.isReady());
-
-    // Destroys the ScopedCollectionDescription object and causes the destructor of MetadataManager
-    // to run, which should trigger all deletion notifications.
-    cursorOnMovedMetadata.reset();
-
-    // Advance time to simulate orphanCleanupDelaySecs passing.
-    {
-        executor::NetworkInterfaceMock::InNetworkGuard guard(network());
-        network()->advanceTime(network()->now() + Seconds{5});
-    }
-
-    notif.wait();
-    future.wait();
-}
-
 TEST_F(MetadataManagerTest, RefreshAfterSuccessfulMigrationSinglePending) {
     ChunkRange cr1(BSON("key" << 0), BSON("key" << 10));
 
@@ -306,7 +211,6 @@ TEST_F(MetadataManagerTest, RefreshAfterSuccessfulMigrationMultiplePending) {
     {
         _manager->setFilteringMetadata(
             cloneMetadataPlusChunk(_manager->getActiveMetadata(boost::none)->get(), cr1));
-        ASSERT_EQ(_manager->numberOfRangesToClean(), 0UL);
         ASSERT_EQ(_manager->getActiveMetadata(boost::none)->get().getChunks().size(), 1UL);
     }
 
@@ -338,25 +242,6 @@ TEST_F(MetadataManagerTest, BeginReceiveWithOverlappingRange) {
     ChunkRange crOverlap(BSON("key" << 5), BSON("key" << 35));
 }
 
-// Tests membership functions for _rangesToClean
-TEST_F(MetadataManagerTest, RangesToCleanMembership) {
-    RAIIServerParameterControllerForTest enableFeatureFlag{"featureFlagRangeDeleterService", false};
-    ChunkRange cr(BSON("key" << 0), BSON("key" << 10));
-    const auto task =
-        insertRangeDeletionTask(operationContext(), kNss, _manager->getCollectionUuid(), cr, 0);
-
-    ASSERT_EQ(0UL, _manager->numberOfRangesToClean());
-
-    // Enable fail point to suspendRangeDeletion.
-    globalFailPointRegistry().find("suspendRangeDeletion")->setMode(FailPoint::alwaysOn);
-
-    auto notifn = _manager->cleanUpRange(cr, false /*delayBeforeDeleting*/);
-    ASSERT(!notifn.isReady());
-    ASSERT_EQ(1UL, _manager->numberOfRangesToClean());
-
-    globalFailPointRegistry().find("suspendRangeDeletion")->setMode(FailPoint::off);
-}
-
 TEST_F(MetadataManagerTest, ClearUnneededChunkManagerObjectsLastSnapshotInList) {
     ChunkRange cr1(BSON("key" << 0), BSON("key" << 10));
     ChunkRange cr2(BSON("key" << 30), BSON("key" << 40));
@@ -365,7 +250,6 @@ TEST_F(MetadataManagerTest, ClearUnneededChunkManagerObjectsLastSnapshotInList)
     {
         _manager->setFilteringMetadata(cloneMetadataPlusChunk(scm1->get(), cr1));
         ASSERT_EQ(_manager->numberOfMetadataSnapshots(), 1UL);
-        ASSERT_EQ(_manager->numberOfRangesToClean(), 0UL);
 
         auto scm2 = _manager->getActiveMetadata(boost::none);
         ASSERT_EQ(scm2->get().getChunks().size(), 1UL);
@@ -392,7 +276,6 @@ TEST_F(MetadataManagerTest, ClearUnneededChunkManagerObjectSnapshotInMiddleOfLis
     auto scm = _manager->getActiveMetadata(boost::none);
     _manager->setFilteringMetadata(cloneMetadataPlusChunk(scm->get(), cr1));
     ASSERT_EQ(_manager->numberOfMetadataSnapshots(), 1UL);
-    ASSERT_EQ(_manager->numberOfRangesToClean(), 0UL);
 
     auto scm2 = _manager->getActiveMetadata(boost::none);
     ASSERT_EQ(scm2->get().getChunks().size(), 1UL);
diff --git a/src/mongo/db/s/migration_coordinator.cpp b/src/mongo/db/s/migration_coordinator.cpp
index a44be90514a..e3bbe8388d7 100644
--- a/src/mongo/db/s/migration_coordinator.cpp
+++ b/src/mongo/db/s/migration_coordinator.cpp
@@ -39,7 +39,6 @@
 #include "mongo/db/vector_clock_mutable.h"
 #include "mongo/logv2/log.h"
 #include "mongo/platform/atomic_word.h"
-#include "mongo/s/sharding_feature_flags_gen.h"
 #include "mongo/util/fail_point.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kShardingMigration
@@ -259,26 +258,6 @@ SharedSemiFuture<void> MigrationCoordinator::_commitMigrationOnDonorAndRecipient
         deletionTask.setKeyPattern(*_shardKeyPattern);
     }
 
-    // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-    // flag is used to turn on new range deleter on startup.
-    if (!feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-        LOGV2_DEBUG(23897,
-                    2,
-                    "Marking range deletion task on donor as ready for processing",
-                    "migrationId"_attr = _migrationInfo.getId());
-        migrationutil::markAsReadyRangeDeletionTaskLocally(
-            opCtx, _migrationInfo.getCollectionUuid(), _migrationInfo.getRange());
-
-        // At this point the decision cannot be changed and will be recovered in the event of a
-        // failover, so it is safe to schedule the deletion task after updating the persisted state.
-        LOGV2_DEBUG(23898,
-                    2,
-                    "Scheduling range deletion task on donor",
-                    "migrationId"_attr = _migrationInfo.getId());
-
-        return migrationutil::submitRangeDeletionTask(opCtx, deletionTask).share();
-    }
-
 
     auto waitForActiveQueriesToComplete = [&]() {
         AutoGetCollection autoColl(opCtx, deletionTask.getNss(), MODE_IS);
diff --git a/src/mongo/db/s/migration_util.cpp b/src/mongo/db/s/migration_util.cpp
index db69fb0a243..eb169c07f6a 100644
--- a/src/mongo/db/s/migration_util.cpp
+++ b/src/mongo/db/s/migration_util.cpp
@@ -439,194 +439,6 @@ bool deletionTaskUuidMatchesFilteringMetadataUuid(
         optCollDescr->uuidMatches(deletionTask.getCollectionUuid());
 }
 
-ExecutorFuture<void> cleanUpRange(ServiceContext* serviceContext,
-                                  const std::shared_ptr<executor::ThreadPoolTaskExecutor>& executor,
-                                  const RangeDeletionTask& deletionTask) {
-    return AsyncTry([=]() mutable {
-               ThreadClient tc(kRangeDeletionThreadName, serviceContext);
-               auto uniqueOpCtx = tc->makeOperationContext();
-               auto opCtx = uniqueOpCtx.get();
-               opCtx->setAlwaysInterruptAtStepDownOrUp_UNSAFE();
-
-               const auto dbName = deletionTask.getNss().dbName();
-               const auto collectionUuid = deletionTask.getCollectionUuid();
-
-               while (true) {
-                   boost::optional<NamespaceString> optNss;
-                   try {
-                       // Holding the locks while enqueueing the task protects against possible
-                       // concurrent cleanups of the filtering metadata, that be serialized
-                       AutoGetCollection autoColl(
-                           opCtx, NamespaceStringOrUUID{dbName, collectionUuid}, MODE_IS);
-                       optNss.emplace(autoColl.getNss());
-                       auto scopedCsr =
-                           CollectionShardingRuntime::assertCollectionLockedAndAcquireShared(
-                               opCtx, *optNss);
-                       auto optCollDescr = scopedCsr->getCurrentMetadataIfKnown();
-
-                       if (optCollDescr) {
-                           uassert(ErrorCodes::
-                                       RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist,
-                                   str::stream()
-                                       << "Filtering metadata for " << optNss->toStringForErrorMsg()
-                                       << (optCollDescr->isSharded()
-                                               ? " has UUID that does not match UUID of "
-                                                 "the deletion task"
-                                               : " is unsharded"),
-                                   deletionTaskUuidMatchesFilteringMetadataUuid(
-                                       opCtx, optCollDescr, deletionTask));
-
-                           LOGV2(6955500,
-                                 "Submitting range deletion task",
-                                 "deletionTask"_attr = redact(deletionTask.toBSON()));
-
-                           const auto whenToClean =
-                               deletionTask.getWhenToClean() == CleanWhenEnum::kNow
-                               ? CollectionShardingRuntime::kNow
-                               : CollectionShardingRuntime::kDelayed;
-
-                           return scopedCsr->cleanUpRange(deletionTask.getRange(), whenToClean);
-                       }
-                   } catch (ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
-                       uasserted(
-                           ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist,
-                           str::stream() << "Collection has been dropped since enqueuing this "
-                                            "range deletion task: "
-                                         << deletionTask.toBSON());
-                   }
-
-
-                   refreshFilteringMetadataUntilSuccess(opCtx, *optNss);
-               }
-           })
-        .until([](Status status) mutable {
-            // Resubmit the range for deletion on a RangeOverlapConflict error.
-            return status != ErrorCodes::RangeOverlapConflict;
-        })
-        .withBackoffBetweenIterations(kExponentialBackoff)
-        .on(executor, CancellationToken::uncancelable());
-}
-
-ExecutorFuture<void> submitRangeDeletionTask(OperationContext* opCtx,
-                                             const RangeDeletionTask& deletionTask) {
-    const auto serviceContext = opCtx->getServiceContext();
-    auto executor = getMigrationUtilExecutor(serviceContext);
-    return ExecutorFuture<void>(executor)
-        .then([=] {
-            ThreadClient tc(kRangeDeletionThreadName, serviceContext);
-
-            uassert(
-                ErrorCodes::ResumableRangeDeleterDisabled,
-                str::stream()
-                    << "Not submitting range deletion task " << redact(deletionTask.toBSON())
-                    << " because the disableResumableRangeDeleter server parameter is set to true",
-                !disableResumableRangeDeleter.load());
-
-            return AsyncTry([=]() {
-                       return cleanUpRange(serviceContext, executor, deletionTask)
-                           .onError<ErrorCodes::KeyPatternShorterThanBound>([=](Status status) {
-                               ThreadClient tc(kRangeDeletionThreadName, serviceContext);
-                               auto uniqueOpCtx = tc->makeOperationContext();
-                               uniqueOpCtx->setAlwaysInterruptAtStepDownOrUp_UNSAFE();
-
-                               LOGV2(55557,
-                                     "cleanUpRange failed due to keyPattern shorter than range "
-                                     "deletion bounds. Refreshing collection metadata to retry.",
-                                     logAttrs(deletionTask.getNss()),
-                                     "status"_attr = redact(status));
-
-                               onCollectionPlacementVersionMismatch(
-                                   uniqueOpCtx.get(), deletionTask.getNss(), boost::none);
-
-                               return status;
-                           });
-                   })
-                .until(
-                    [](Status status) { return status != ErrorCodes::KeyPatternShorterThanBound; })
-                .on(executor, CancellationToken::uncancelable());
-        })
-        .onError([=](const Status status) {
-            ThreadClient tc(kRangeDeletionThreadName, serviceContext);
-            auto uniqueOpCtx = tc->makeOperationContext();
-            auto opCtx = uniqueOpCtx.get();
-
-            LOGV2(22027,
-                  "Failed to submit range deletion task",
-                  "deletionTask"_attr = redact(deletionTask.toBSON()),
-                  "error"_attr = redact(status),
-                  "migrationId"_attr = deletionTask.getId());
-
-            if (status == ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist) {
-                deleteRangeDeletionTaskLocally(opCtx,
-                                               deletionTask.getCollectionUuid(),
-                                               deletionTask.getRange(),
-                                               ShardingCatalogClient::kLocalWriteConcern);
-            }
-
-            // Note, we use onError and make it return its input status, because ExecutorFuture does
-            // not support tapError.
-            return status;
-        });
-}
-
-void submitPendingDeletions(OperationContext* opCtx) {
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    auto query = BSON("pending" << BSON("$exists" << false));
-
-    store.forEach(opCtx, query, [&opCtx](const RangeDeletionTask& deletionTask) {
-        migrationutil::submitRangeDeletionTask(opCtx, deletionTask).getAsync([](auto) {});
-        return true;
-    });
-}
-
-void resubmitRangeDeletionsOnStepUp(ServiceContext* serviceContext) {
-    LOGV2(22028, "Starting pending deletion submission thread.");
-
-    ExecutorFuture<void>(getMigrationUtilExecutor(serviceContext))
-        .then([serviceContext] {
-            ThreadClient tc("ResubmitRangeDeletions", serviceContext);
-
-            auto opCtx = tc->makeOperationContext();
-            opCtx->setAlwaysInterruptAtStepDownOrUp_UNSAFE();
-
-            DBDirectClient client(opCtx.get());
-            FindCommandRequest findCommand(NamespaceString::kRangeDeletionNamespace);
-            findCommand.setFilter(BSON(RangeDeletionTask::kProcessingFieldName << true));
-            auto cursor = client.find(std::move(findCommand));
-
-            auto retFuture = ExecutorFuture<void>(getMigrationUtilExecutor(serviceContext));
-
-            int rangeDeletionsMarkedAsProcessing = 0;
-            while (cursor->more()) {
-                retFuture = migrationutil::submitRangeDeletionTask(
-                    opCtx.get(),
-                    RangeDeletionTask::parse(IDLParserContext("rangeDeletionRecovery"),
-                                             cursor->next()));
-                rangeDeletionsMarkedAsProcessing++;
-            }
-
-            if (rangeDeletionsMarkedAsProcessing > 1) {
-                LOGV2_WARNING(
-                    6695800,
-                    "Rescheduling several range deletions marked as processing. Orphans count "
-                    "may be off while they are not drained",
-                    "numRangeDeletionsMarkedAsProcessing"_attr = rangeDeletionsMarkedAsProcessing);
-            }
-
-            return retFuture;
-        })
-        .then([serviceContext] {
-            ThreadClient tc("ResubmitRangeDeletions", serviceContext);
-
-            auto opCtx = tc->makeOperationContext();
-            opCtx->setAlwaysInterruptAtStepDownOrUp_UNSAFE();
-
-            submitPendingDeletions(opCtx.get());
-        })
-        .getAsync([](auto) {});
-}
-
 void persistMigrationCoordinatorLocally(OperationContext* opCtx,
                                         const MigrationCoordinatorDocument& migrationDoc) {
     PersistentTaskStore<MigrationCoordinatorDocument> store(
diff --git a/src/mongo/db/s/migration_util.h b/src/mongo/db/s/migration_util.h
index 2c1872659f0..bd88ac829a7 100644
--- a/src/mongo/db/s/migration_util.h
+++ b/src/mongo/db/s/migration_util.h
@@ -123,31 +123,6 @@ size_t checkForConflictingDeletions(OperationContext* opCtx,
                                     const UUID& uuid);
 
 /**
- * Asynchronously attempts to submit the RangeDeletionTask for processing.
- *
- * Note that if the current filtering metadata's UUID does not match the task's UUID, the filtering
- * metadata will be refreshed once. If the UUID's still don't match, the task will be deleted from
- * disk. If the UUID's do match, the range will be submitted for deletion.
- *
- * If the range is submitted for deletion, the returned future is set when the range deletion
- * completes. If the range is not submitted for deletion, the returned future is set with an error
- * explaining why.
- */
-ExecutorFuture<void> submitRangeDeletionTask(OperationContext* oppCtx,
-                                             const RangeDeletionTask& deletionTask);
-
-/**
- * Queries the rangeDeletions collection for ranges that are ready to be deleted and submits them to
- * the range deleter.
- */
-void submitPendingDeletions(OperationContext* opCtx);
-
-/**
- * Asynchronously calls submitPendingDeletions using the fixed executor pool.
- */
-void resubmitRangeDeletionsOnStepUp(ServiceContext* serviceContext);
-
-/**
  * Writes the migration coordinator document to config.migrationCoordinators and waits for majority
  * write concern.
  */
diff --git a/src/mongo/db/s/migration_util_test.cpp b/src/mongo/db/s/migration_util_test.cpp
index 029a7ff39fc..463f674a2e9 100644
--- a/src/mongo/db/s/migration_util_test.cpp
+++ b/src/mongo/db/s/migration_util_test.cpp
@@ -366,373 +366,5 @@ TEST_F(MigrationUtilsTest, TestUpdateNumberOfOrphans) {
     ASSERT_EQ(store.count(opCtx, rangeDeletionDoc.toBSON().removeField("timestamp")), 1);
 }
 
-/**
- * Fixture that uses a mocked CatalogCacheLoader and CatalogClient to allow metadata refreshes
- * without using the mock network.
- */
-class SubmitRangeDeletionTaskTest : public CollectionShardingRuntimeWithRangeDeleterTest {
-public:
-    const HostAndPort kConfigHostAndPort{"dummy", 123};
-    const ShardKeyPattern kShardKeyPattern = ShardKeyPattern(BSON("_id" << 1));
-    const UUID kDefaultUUID = UUID::gen();
-    const OID kEpoch = OID::gen();
-    const Timestamp kDefaultTimestamp = Timestamp(2, 0);
-    const DatabaseType kDefaultDatabaseType = DatabaseType(
-        kTestNss.db().toString(), ShardId("0"), DatabaseVersion(kDefaultUUID, kDefaultTimestamp));
-    const std::vector<ShardType> kShardList = {ShardType("0", "Host0:12345"),
-                                               ShardType("1", "Host1:12345")};
-
-    void setUp() override {
-        // Don't call ShardServerTestFixture::setUp so we can install a mock catalog cache loader.
-        ShardingMongodTestFixture::setUp();
-
-        replicationCoordinator()->alwaysAllowWrites(true);
-        serverGlobalParams.clusterRole = ClusterRole::ShardServer;
-
-        _clusterId = OID::gen();
-        ShardingState::get(getServiceContext())->setInitialized(_myShardName, _clusterId);
-
-        auto mockLoader = std::make_unique<CatalogCacheLoaderMock>();
-        _mockCatalogCacheLoader = mockLoader.get();
-        CatalogCacheLoader::set(getServiceContext(), std::move(mockLoader));
-
-        uassertStatusOK(
-            initializeGlobalShardingStateForMongodForTest(ConnectionString(kConfigHostAndPort)));
-
-        configTargeterMock()->setFindHostReturnValue(kConfigHostAndPort);
-
-        WaitForMajorityService::get(getServiceContext()).startup(getServiceContext());
-
-        // Set up 2 default shards.
-        for (const auto& shard : kShardList) {
-            std::unique_ptr<RemoteCommandTargeterMock> targeter(
-                std::make_unique<RemoteCommandTargeterMock>());
-            HostAndPort host(shard.getHost());
-            targeter->setConnectionStringReturnValue(ConnectionString(host));
-            targeter->setFindHostReturnValue(host);
-            targeterFactory()->addTargeterToReturn(ConnectionString(host), std::move(targeter));
-        }
-    }
-
-    void tearDown() override {
-        WaitForMajorityService::get(getServiceContext()).shutDown();
-
-        ShardServerTestFixture::tearDown();
-    }
-
-    // Mock for the ShardingCatalogClient used to satisfy loading all shards for the ShardRegistry
-    // and loading all collections when a database is loaded for the first time by the CatalogCache.
-    class StaticCatalogClient final : public ShardingCatalogClientMock {
-    public:
-        StaticCatalogClient(std::vector<ShardType> shards) : _shards(std::move(shards)) {}
-
-        StatusWith<repl::OpTimeWith<std::vector<ShardType>>> getAllShards(
-            OperationContext* opCtx, repl::ReadConcernLevel readConcern) override {
-            return repl::OpTimeWith<std::vector<ShardType>>(_shards);
-        }
-
-        std::vector<CollectionType> getCollections(OperationContext* opCtx,
-                                                   StringData dbName,
-                                                   repl::ReadConcernLevel readConcernLevel,
-                                                   const BSONObj& sort) override {
-            return _colls;
-        }
-
-        std::pair<CollectionType, std::vector<IndexCatalogType>>
-        getCollectionAndShardingIndexCatalogEntries(
-            OperationContext* opCtx,
-            const NamespaceString& nss,
-            const repl::ReadConcernArgs& readConcern) override {
-            if (!_coll) {
-                uasserted(ErrorCodes::NamespaceNotFound, "dummy errmsg");
-            }
-            return std::make_pair(*_coll, std::vector<IndexCatalogType>());
-        }
-
-        void setCollections(std::vector<CollectionType> colls) {
-            _colls = std::move(colls);
-        }
-
-        void setCollection(boost::optional<CollectionType> coll) {
-            _coll = coll;
-        }
-
-    private:
-        const std::vector<ShardType> _shards;
-        std::vector<CollectionType> _colls;
-        boost::optional<CollectionType> _coll;
-    };
-
-    UUID createCollectionAndGetUUID(const NamespaceString& nss) {
-        {
-            OperationShardingState::ScopedAllowImplicitCollectionCreate_UNSAFE
-                unsafeCreateCollection(operationContext());
-            uassertStatusOK(
-                createCollection(operationContext(), nss.dbName(), BSON("create" << nss.coll())));
-        }
-
-        AutoGetCollection autoColl(operationContext(), nss, MODE_IX);
-        return autoColl.getCollection()->uuid();
-    }
-
-    std::unique_ptr<ShardingCatalogClient> makeShardingCatalogClient() override {
-        auto mockCatalogClient = std::make_unique<StaticCatalogClient>(kShardList);
-        // Stash a pointer to the mock so its return values can be set.
-        _mockCatalogClient = mockCatalogClient.get();
-        return mockCatalogClient;
-    }
-
-    CollectionType makeCollectionType(UUID uuid, OID epoch, Timestamp timestamp) {
-        CollectionType coll(
-            kTestNss, epoch, timestamp, Date_t::now(), uuid, kShardKeyPattern.getKeyPattern());
-        coll.setUnique(true);
-        return coll;
-    }
-
-    std::vector<ChunkType> makeChangedChunks(ChunkVersion startingVersion) {
-        const auto uuid = UUID::gen();
-        ChunkType chunk1(uuid,
-                         {kShardKeyPattern.getKeyPattern().globalMin(), BSON("_id" << -100)},
-                         startingVersion,
-                         {"0"});
-        chunk1.setName(OID::gen());
-        startingVersion.incMinor();
-
-        ChunkType chunk2(uuid, {BSON("_id" << -100), BSON("_id" << 0)}, startingVersion, {"1"});
-        chunk2.setName(OID::gen());
-        startingVersion.incMinor();
-
-        ChunkType chunk3(uuid, {BSON("_id" << 0), BSON("_id" << 100)}, startingVersion, {"0"});
-        chunk3.setName(OID::gen());
-        startingVersion.incMinor();
-
-        ChunkType chunk4(uuid,
-                         {BSON("_id" << 100), kShardKeyPattern.getKeyPattern().globalMax()},
-                         startingVersion,
-                         {"1"});
-        chunk4.setName(OID::gen());
-        startingVersion.incMinor();
-
-        return std::vector<ChunkType>{chunk1, chunk2, chunk3, chunk4};
-    }
-
-    CatalogCacheLoaderMock* _mockCatalogCacheLoader;
-    StaticCatalogClient* _mockCatalogClient;
-
-    RAIIServerParameterControllerForTest enableFeatureFlag{"featureFlagRangeDeleterService", false};
-};
-
-TEST_F(SubmitRangeDeletionTaskTest,
-       FailsAndDeletesTaskIfFilteringMetadataIsUnknownEvenAfterRefresh) {
-    auto opCtx = operationContext();
-    auto deletionTask = createDeletionTask(opCtx, kTestNss, kDefaultUUID, 0, 10, _myShardName);
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    store.add(opCtx, deletionTask);
-    ASSERT_EQ(store.count(opCtx), 1);
-    migrationutil::markAsReadyRangeDeletionTaskLocally(
-        opCtx, deletionTask.getCollectionUuid(), deletionTask.getRange());
-
-    // Make the refresh triggered by submitting the task return an empty result when loading the
-    // database.
-    _mockCatalogCacheLoader->setDatabaseRefreshReturnValue(
-        Status(ErrorCodes::NamespaceNotFound, "dummy errmsg"));
-
-    auto cleanupCompleteFuture = migrationutil::submitRangeDeletionTask(opCtx, deletionTask);
-
-    // The task should not have been submitted, and the task's entry should have been removed from
-    // the persistent store.
-    ASSERT_THROWS_CODE(cleanupCompleteFuture.get(opCtx),
-                       AssertionException,
-                       ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist);
-    ASSERT_EQ(store.count(opCtx), 0);
-}
-
-TEST_F(SubmitRangeDeletionTaskTest, FailsAndDeletesTaskIfNamespaceIsUnshardedEvenAfterRefresh) {
-    auto opCtx = operationContext();
-
-    auto deletionTask = createDeletionTask(opCtx, kTestNss, kDefaultUUID, 0, 10, _myShardName);
-
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    store.add(opCtx, deletionTask);
-    ASSERT_EQ(store.count(opCtx), 1);
-    migrationutil::markAsReadyRangeDeletionTaskLocally(
-        opCtx, deletionTask.getCollectionUuid(), deletionTask.getRange());
-
-    // Make the refresh triggered by submitting the task return an empty result when loading the
-    // collection so it is considered unsharded.
-    _mockCatalogCacheLoader->setDatabaseRefreshReturnValue(kDefaultDatabaseType);
-    _mockCatalogCacheLoader->setCollectionRefreshReturnValue(
-        Status(ErrorCodes::NamespaceNotFound, "dummy errmsg"));
-
-    auto cleanupCompleteFuture = migrationutil::submitRangeDeletionTask(opCtx, deletionTask);
-
-    // The task should not have been submitted, and the task's entry should have been removed from
-    // the persistent store.
-    ASSERT_THROWS_CODE(cleanupCompleteFuture.get(opCtx),
-                       AssertionException,
-                       ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist);
-    ASSERT_EQ(store.count(opCtx), 0);
-}
-
-TEST_F(SubmitRangeDeletionTaskTest,
-       FailsAndDeletesTaskIfNamespaceIsUnshardedBeforeAndAfterRefresh) {
-    auto opCtx = operationContext();
-
-    auto deletionTask = createDeletionTask(opCtx, kTestNss, kDefaultUUID, 0, 10, _myShardName);
-
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    store.add(opCtx, deletionTask);
-    ASSERT_EQ(store.count(opCtx), 1);
-    migrationutil::markAsReadyRangeDeletionTaskLocally(
-        opCtx, deletionTask.getCollectionUuid(), deletionTask.getRange());
-
-    // Mock an empty result for the task's collection and force a refresh so the node believes the
-    // collection is unsharded.
-    _mockCatalogCacheLoader->setDatabaseRefreshReturnValue(kDefaultDatabaseType);
-    _mockCatalogCacheLoader->setCollectionRefreshReturnValue(
-        Status(ErrorCodes::NamespaceNotFound, "dummy errmsg"));
-    forceShardFilteringMetadataRefresh(opCtx, kTestNss);
-
-    auto cleanupCompleteFuture = migrationutil::submitRangeDeletionTask(opCtx, deletionTask);
-
-    // The task should not have been submitted, and the task's entry should have been removed from
-    // the persistent store.
-    ASSERT_THROWS_CODE(cleanupCompleteFuture.get(opCtx),
-                       AssertionException,
-                       ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist);
-    ASSERT_EQ(store.count(opCtx), 0);
-}
-
-TEST_F(SubmitRangeDeletionTaskTest, SucceedsIfFilteringMetadataUUIDMatchesTaskUUID) {
-    auto opCtx = operationContext();
-
-    auto collectionUUID = createCollectionAndGetUUID(kTestNss);
-    auto deletionTask = createDeletionTask(opCtx, kTestNss, collectionUUID, 0, 10, _myShardName);
-
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    store.add(opCtx, deletionTask);
-    ASSERT_EQ(store.count(opCtx), 1);
-    migrationutil::markAsReadyRangeDeletionTaskLocally(
-        opCtx, deletionTask.getCollectionUuid(), deletionTask.getRange());
-
-    // Force a metadata refresh with the task's UUID before the task is submitted.
-    auto coll = makeCollectionType(collectionUUID, kEpoch, kDefaultTimestamp);
-    _mockCatalogCacheLoader->setDatabaseRefreshReturnValue(kDefaultDatabaseType);
-    _mockCatalogCacheLoader->setCollectionRefreshReturnValue(coll);
-    _mockCatalogCacheLoader->setChunkRefreshReturnValue(
-        makeChangedChunks(ChunkVersion({kEpoch, kDefaultTimestamp}, {1, 0})));
-    _mockCatalogClient->setCollections({coll});
-    _mockCatalogClient->setCollection(coll);
-    forceShardFilteringMetadataRefresh(opCtx, kTestNss);
-
-    // The task should have been submitted successfully.
-    auto cleanupCompleteFuture = migrationutil::submitRangeDeletionTask(opCtx, deletionTask);
-    cleanupCompleteFuture.get(opCtx);
-}
-
-TEST_F(
-    SubmitRangeDeletionTaskTest,
-    SucceedsIfFilteringMetadataInitiallyUnknownButFilteringMetadataUUIDMatchesTaskUUIDAfterRefresh) {
-    auto opCtx = operationContext();
-
-    auto collectionUUID = createCollectionAndGetUUID(kTestNss);
-    auto deletionTask = createDeletionTask(opCtx, kTestNss, collectionUUID, 0, 10, _myShardName);
-
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    store.add(opCtx, deletionTask);
-    ASSERT_EQ(store.count(opCtx), 1);
-    migrationutil::markAsReadyRangeDeletionTaskLocally(
-        opCtx, deletionTask.getCollectionUuid(), deletionTask.getRange());
-
-    // Make the refresh triggered by submitting the task return a UUID that matches the task's UUID.
-    auto coll = makeCollectionType(collectionUUID, kEpoch, kDefaultTimestamp);
-    _mockCatalogCacheLoader->setDatabaseRefreshReturnValue(kDefaultDatabaseType);
-    _mockCatalogCacheLoader->setCollectionRefreshReturnValue(coll);
-    _mockCatalogCacheLoader->setChunkRefreshReturnValue(
-        makeChangedChunks(ChunkVersion({kEpoch, kDefaultTimestamp}, {1, 0})));
-    _mockCatalogClient->setCollections({coll});
-
-    auto metadata = makeShardedMetadata(opCtx, collectionUUID);
-    csr()->setFilteringMetadata(opCtx, metadata);
-
-    // The task should have been submitted successfully.
-    auto cleanupCompleteFuture = migrationutil::submitRangeDeletionTask(opCtx, deletionTask);
-    cleanupCompleteFuture.get(opCtx);
-}
-
-TEST_F(SubmitRangeDeletionTaskTest,
-       SucceedsIfTaskNamespaceInitiallyUnshardedButUUIDMatchesAfterRefresh) {
-    auto opCtx = operationContext();
-
-    // Force a metadata refresh with no collection entry so the node believes the namespace is
-    // unsharded when the task is submitted.
-    _mockCatalogCacheLoader->setDatabaseRefreshReturnValue(kDefaultDatabaseType);
-    _mockCatalogCacheLoader->setCollectionRefreshReturnValue(
-        Status(ErrorCodes::NamespaceNotFound, "dummy errmsg"));
-    forceShardFilteringMetadataRefresh(opCtx, kTestNss);
-
-    auto collectionUUID = createCollectionAndGetUUID(kTestNss);
-    auto deletionTask = createDeletionTask(opCtx, kTestNss, collectionUUID, 0, 10, _myShardName);
-
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    store.add(opCtx, deletionTask);
-    ASSERT_EQ(store.count(opCtx), 1);
-    migrationutil::markAsReadyRangeDeletionTaskLocally(
-        opCtx, deletionTask.getCollectionUuid(), deletionTask.getRange());
-
-    // Make the refresh triggered by submitting the task return a UUID that matches the task's UUID.
-    auto matchingColl = makeCollectionType(collectionUUID, kEpoch, kDefaultTimestamp);
-    _mockCatalogCacheLoader->setCollectionRefreshReturnValue(matchingColl);
-    _mockCatalogCacheLoader->setChunkRefreshReturnValue(
-        makeChangedChunks(ChunkVersion({kEpoch, kDefaultTimestamp}, {10, 0})));
-    _mockCatalogClient->setCollections({matchingColl});
-    _mockCatalogClient->setCollection({matchingColl});
-
-    auto metadata = makeShardedMetadata(opCtx, collectionUUID);
-    csr()->setFilteringMetadata(opCtx, metadata);
-
-    // The task should have been submitted successfully.
-    auto cleanupCompleteFuture = migrationutil::submitRangeDeletionTask(opCtx, deletionTask);
-    cleanupCompleteFuture.get(opCtx);
-}
-
-TEST_F(SubmitRangeDeletionTaskTest,
-       FailsAndDeletesTaskIfFilteringMetadataUUIDDifferentFromTaskUUIDEvenAfterRefresh) {
-    auto opCtx = operationContext();
-
-    auto deletionTask = createDeletionTask(opCtx, kTestNss, kDefaultUUID, 0, 10, _myShardName);
-
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-
-    store.add(opCtx, deletionTask);
-    ASSERT_EQ(store.count(opCtx), 1);
-    migrationutil::markAsReadyRangeDeletionTaskLocally(
-        opCtx, deletionTask.getCollectionUuid(), deletionTask.getRange());
-
-    // Make the refresh triggered by submitting the task return an arbitrary UUID.
-    const auto otherEpoch = OID::gen();
-    const auto otherTimestamp = Timestamp(3, 0);
-    auto otherColl = makeCollectionType(UUID::gen(), otherEpoch, otherTimestamp);
-    _mockCatalogCacheLoader->setDatabaseRefreshReturnValue(kDefaultDatabaseType);
-    _mockCatalogCacheLoader->setCollectionRefreshReturnValue(otherColl);
-    _mockCatalogCacheLoader->setChunkRefreshReturnValue(
-        makeChangedChunks(ChunkVersion({otherEpoch, otherTimestamp}, {1, 0})));
-    _mockCatalogClient->setCollections({otherColl});
-
-    // The task should not have been submitted, and the task's entry should have been removed from
-    // the persistent store.
-    auto cleanupCompleteFuture = migrationutil::submitRangeDeletionTask(opCtx, deletionTask);
-    ASSERT_THROWS_CODE(cleanupCompleteFuture.get(opCtx),
-                       AssertionException,
-                       ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist);
-    ASSERT_EQ(store.count(opCtx), 0);
-}
-
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/s/range_deleter_service.cpp b/src/mongo/db/s/range_deleter_service.cpp
index 94539b3127b..2c8af8ec2b8 100644
--- a/src/mongo/db/s/range_deleter_service.cpp
+++ b/src/mongo/db/s/range_deleter_service.cpp
@@ -40,7 +40,6 @@
 #include "mongo/db/s/range_deletion_util.h"
 #include "mongo/db/s/shard_filtering_metadata_refresh.h"
 #include "mongo/logv2/log.h"
-#include "mongo/s/sharding_feature_flags_gen.h"
 #include "mongo/util/future_util.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kShardingRangeDeleter
@@ -278,10 +277,7 @@ void RangeDeleterService::ReadyRangeDeletionsProcessor::_runRangeDeletions() {
 }
 
 void RangeDeleterService::onStartup(OperationContext* opCtx) {
-    // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-    // flag is used to turn on new range deleter on startup.
-    if (disableResumableRangeDeleter.load() ||
-        !feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
+    if (disableResumableRangeDeleter.load()) {
         return;
     }
 
@@ -291,12 +287,6 @@ void RangeDeleterService::onStartup(OperationContext* opCtx) {
 }
 
 void RangeDeleterService::onStepUpComplete(OperationContext* opCtx, long long term) {
-    // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-    // flag is used to turn on new range deleter on startup.
-    if (!feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-        return;
-    }
-
     if (disableResumableRangeDeleter.load()) {
         LOGV2_INFO(
             6872508,
diff --git a/src/mongo/db/s/range_deleter_service_test.cpp b/src/mongo/db/s/range_deleter_service_test.cpp
index ab5513ecebb..d2f8b692f29 100644
--- a/src/mongo/db/s/range_deleter_service_test.cpp
+++ b/src/mongo/db/s/range_deleter_service_test.cpp
@@ -588,7 +588,7 @@ TEST_F(RangeDeleterServiceTest, TotalNumOfRegisteredTasks) {
 }
 
 TEST_F(RangeDeleterServiceTest, RegisterTaskWithDisableResumableRangeDeleterFlagEnabled) {
-    RAIIServerParameterControllerForTest enableFeatureFlag{"disableResumableRangeDeleter", true};
+    RAIIServerParameterControllerForTest disableRangeDeleter{"disableResumableRangeDeleter", true};
 
     auto rds = RangeDeleterService::get(opCtx);
     auto taskWithOngoingQueries = rangeDeletionTask0ForCollA;
@@ -615,7 +615,7 @@ TEST_F(RangeDeleterServiceTest,
         uuidCollA, taskWithOngoingQueries->getTask().getRange());
     ASSERT(!overlappingRangeFuture.isReady());
 
-    RAIIServerParameterControllerForTest enableFeatureFlag{"disableResumableRangeDeleter", true};
+    RAIIServerParameterControllerForTest disableRangeDeleter{"disableResumableRangeDeleter", true};
     auto overlappingRangeFutureWhenDisabled = rds->getOverlappingRangeDeletionsFuture(
         uuidCollA, taskWithOngoingQueries->getTask().getRange());
     ASSERT(overlappingRangeFutureWhenDisabled.isReady());
diff --git a/src/mongo/db/s/range_deleter_service_test.h b/src/mongo/db/s/range_deleter_service_test.h
index cc44ac03225..3053ae8314c 100644
--- a/src/mongo/db/s/range_deleter_service_test.h
+++ b/src/mongo/db/s/range_deleter_service_test.h
@@ -78,7 +78,6 @@ private:
     void _setFilteringMetadataByUUID(OperationContext* opCtx, const UUID& uuid);
 
     // Scoped objects
-    RAIIServerParameterControllerForTest enableFeatureFlag{"featureFlagRangeDeleterService", true};
     unittest::MinimumLoggedSeverityGuard _severityGuard{logv2::LogComponent::kShardingRangeDeleter,
                                                         logv2::LogSeverity::Debug(2)};
 };
diff --git a/src/mongo/db/s/range_deletion_util.cpp b/src/mongo/db/s/range_deletion_util.cpp
index b5fccf133e6..42f18b201c9 100644
--- a/src/mongo/db/s/range_deletion_util.cpp
+++ b/src/mongo/db/s/range_deletion_util.cpp
@@ -47,7 +47,6 @@
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/query/query_planner.h"
 #include "mongo/db/repl/repl_client_info.h"
-#include "mongo/db/repl/wait_for_majority_service.h"
 #include "mongo/db/s/balancer_stats_registry.h"
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/s/sharding_runtime_d_params_gen.h"
@@ -56,11 +55,7 @@
 #include "mongo/db/shard_role.h"
 #include "mongo/db/storage/remove_saver.h"
 #include "mongo/db/write_concern.h"
-#include "mongo/executor/task_executor.h"
 #include "mongo/logv2/log.h"
-#include "mongo/s/catalog/sharding_catalog_client.h"
-#include "mongo/util/cancellation.h"
-#include "mongo/util/future_util.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kShardingRangeDeleter
 
@@ -248,53 +243,6 @@ void markRangeDeletionTaskAsProcessing(OperationContext* opCtx,
     }
 }
 
-/**
- * Delete the range in a sequence of batches until there are no more documents to delete or deletion
- * returns an error.
- */
-ExecutorFuture<void> deleteRangeInBatchesWithExecutor(
-    const std::shared_ptr<executor::TaskExecutor>& executor,
-    const NamespaceString& nss,
-    const UUID& collectionUuid,
-    const BSONObj& keyPattern,
-    const ChunkRange& range) {
-    return ExecutorFuture<void>(executor).then([=] {
-        return withTemporaryOperationContext(
-            [=](OperationContext* opCtx) {
-                return deleteRangeInBatches(opCtx, nss.dbName(), collectionUuid, keyPattern, range);
-            },
-            nss.dbName(),
-            collectionUuid);
-    });
-}
-
-ExecutorFuture<void> waitForDeletionsToMajorityReplicate(
-    const std::shared_ptr<executor::TaskExecutor>& executor,
-    const NamespaceString& nss,
-    const UUID& collectionUuid,
-    const ChunkRange& range) {
-    return withTemporaryOperationContext(
-        [=](OperationContext* opCtx) {
-            repl::ReplClientInfo::forClient(opCtx->getClient()).setLastOpToSystemLastOpTime(opCtx);
-            auto clientOpTime = repl::ReplClientInfo::forClient(opCtx->getClient()).getLastOp();
-
-            LOGV2_DEBUG(5346202,
-                        1,
-                        "Waiting for majority replication of local deletions",
-                        logAttrs(nss),
-                        "collectionUUID"_attr = collectionUuid,
-                        "range"_attr = redact(range.toString()),
-                        "clientOpTime"_attr = clientOpTime);
-
-            // Asynchronously wait for majority write concern.
-            return WaitForMajorityService::get(opCtx->getServiceContext())
-                .waitUntilMajority(clientOpTime, CancellationToken::uncancelable())
-                .thenRunOn(executor);
-        },
-        nss.dbName(),
-        collectionUuid);
-}
-
 std::vector<RangeDeletionTask> getPersistentRangeDeletionTasks(OperationContext* opCtx,
                                                                const NamespaceString& nss) {
     std::vector<RangeDeletionTask> tasks;
@@ -463,119 +411,6 @@ void deleteRangeDeletionTasksForRename(OperationContext* opCtx,
                                         BSON(RangeDeletionTask::kNssFieldName << toNss.ns()));
 }
 
-SharedSemiFuture<void> removeDocumentsInRange(
-    const std::shared_ptr<executor::TaskExecutor>& executor,
-    SemiFuture<void> waitForActiveQueriesToComplete,
-    const NamespaceString& nss,
-    const UUID& collectionUuid,
-    const BSONObj& keyPattern,
-    const ChunkRange& range,
-    Seconds delayForActiveQueriesOnSecondariesToComplete) {
-    return std::move(waitForActiveQueriesToComplete)
-        .thenRunOn(executor)
-        .onError([&](Status s) {
-            // The code does not expect the input future to have an error set on it, so we
-            // invariant here to prevent future misuse (no pun intended).
-            invariant(s.isOK());
-        })
-        .then([=]() mutable {
-            // Wait for possibly ongoing queries on secondaries to complete.
-            return sleepUntil(executor,
-                              executor->now() + delayForActiveQueriesOnSecondariesToComplete);
-        })
-        .then([=]() mutable {
-            LOGV2_DEBUG(23772,
-                        1,
-                        "Beginning deletion of documents",
-                        logAttrs(nss),
-                        "range"_attr = redact(range.toString()));
-
-            return deleteRangeInBatchesWithExecutor(
-                       executor, nss, collectionUuid, keyPattern, range)
-                .onCompletion([=](Status s) {
-                    if (!s.isOK() &&
-                        s.code() !=
-                            ErrorCodes::RangeDeletionAbandonedBecauseTaskDocumentDoesNotExist) {
-                        // Propagate any errors to the onCompletion() handler below.
-                        return ExecutorFuture<void>(executor, s);
-                    }
-
-                    // We wait for majority write concern even if the range deletion task document
-                    // doesn't exist to guarantee the deletion (which must have happened earlier) is
-                    // visible to the caller at non-local read concerns.
-                    return waitForDeletionsToMajorityReplicate(executor, nss, collectionUuid, range)
-                        .then([=] {
-                            LOGV2_DEBUG(5346201,
-                                        1,
-                                        "Finished waiting for majority for deleted batch",
-                                        logAttrs(nss),
-                                        "range"_attr = redact(range.toString()));
-                            // Propagate any errors to the onCompletion() handler below.
-                            return s;
-                        });
-                });
-        })
-        .onCompletion([=](Status s) {
-            if (s.isOK()) {
-                LOGV2_DEBUG(23773,
-                            1,
-                            "Completed deletion of documents in {namespace} range {range}",
-                            "Completed deletion of documents",
-                            logAttrs(nss),
-                            "range"_attr = redact(range.toString()));
-            } else {
-                LOGV2(23774,
-                      "Failed to delete documents in {namespace} range {range} due to {error}",
-                      "Failed to delete documents",
-                      logAttrs(nss),
-                      "range"_attr = redact(range.toString()),
-                      "error"_attr = redact(s));
-            }
-
-            if (s.code() == ErrorCodes::RangeDeletionAbandonedBecauseTaskDocumentDoesNotExist) {
-                return Status::OK();
-            }
-
-            if (!s.isOK() &&
-                s.code() !=
-                    ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist) {
-                // Propagate any errors to callers waiting on the result.
-                return s;
-            }
-
-            try {
-                withTemporaryOperationContext(
-                    [&](OperationContext* opCtx) {
-                        removePersistentRangeDeletionTask(opCtx, collectionUuid, range);
-                    },
-                    nss.dbName(),
-                    collectionUuid);
-            } catch (const DBException& e) {
-                LOGV2_ERROR(23770,
-                            "Failed to delete range deletion task for range {range} in collection "
-                            "{namespace} due to {error}",
-                            "Failed to delete range deletion task",
-                            "range"_attr = range,
-                            logAttrs(nss),
-                            "error"_attr = e.what());
-
-                return e.toStatus();
-            }
-
-            LOGV2_DEBUG(23775,
-                        1,
-                        "Completed removal of persistent range deletion task for {namespace} "
-                        "range {range}",
-                        "Completed removal of persistent range deletion task",
-                        logAttrs(nss),
-                        "range"_attr = redact(range.toString()));
-
-            // Propagate any errors to callers waiting on the result.
-            return s;
-        })
-        .semi()
-        .share();
-}
 
 void persistUpdatedNumOrphans(OperationContext* opCtx,
                               const UUID& collectionUuid,
diff --git a/src/mongo/db/s/range_deletion_util.h b/src/mongo/db/s/range_deletion_util.h
index 870284248b6..566972461f9 100644
--- a/src/mongo/db/s/range_deletion_util.h
+++ b/src/mongo/db/s/range_deletion_util.h
@@ -43,30 +43,6 @@ namespace mongo {
 constexpr auto kRangeDeletionThreadName = "range-deleter"_sd;
 
 /**
- * DO NOT USE - only necessary for the legacy range deleter
- *
- * Deletes a range of orphaned documents for the given namespace and collection UUID. Returns a
- * future which will be resolved when the range has finished being deleted. The resulting future
- * will contain an error in cases where the range could not be deleted successfully.
- *
- * The overall algorithm is as follows:
- * 1. Wait for the all active queries which could be using the range to resolve by waiting
- *    for the waitForActiveQueriesToComplete future to resolve.
- * 2. Waits for delayForActiveQueriesOnSecondariesToComplete seconds before deleting any documents,
- *    to give queries running on secondaries a chance to finish.
- * 3. Delete documents in a series of batches with up to numDocsToRemovePerBatch documents per
- *    batch, with a delay of delayBetweenBatches milliseconds in between batches.
- */
-SharedSemiFuture<void> removeDocumentsInRange(
-    const std::shared_ptr<executor::TaskExecutor>& executor,
-    SemiFuture<void> waitForActiveQueriesToComplete,
-    const NamespaceString& nss,
-    const UUID& collectionUuid,
-    const BSONObj& keyPattern,
-    const ChunkRange& range,
-    Seconds delayForActiveQueriesOnSecondariesToComplete);
-
-/**
  * Delete the range in a sequence of batches until there are no more documents to delete or deletion
  * returns an error.
  */
diff --git a/src/mongo/db/s/range_deletion_util_test.cpp b/src/mongo/db/s/range_deletion_util_test.cpp
index c04869cb31e..0d6bbb3b41e 100644
--- a/src/mongo/db/s/range_deletion_util_test.cpp
+++ b/src/mongo/db/s/range_deletion_util_test.cpp
@@ -211,647 +211,6 @@ RangeDeletionTask insertRangeDeletionTask(OperationContext* opCtx,
     return insertRangeDeletionTask(opCtx, kNss, uuid, range, numOrphans);
 }
 
-TEST_F(RangeDeleterTest,
-       RemoveDocumentsInRangeRemovesAllDocumentsInRangeWhenAllDocumentsFitInSingleBatch) {
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    setFilteringMetadataWithUUID(uuid());
-    auto task = insertRangeDeletionTask(_opCtx, uuid(), range, 1);
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-    cleanupComplete.get();
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-}
-
-TEST_F(RangeDeleterTest,
-       RemoveDocumentsInRangeRemovesAllDocumentsInRangeWhenSeveralBatchesAreRequired) {
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    // More documents than the batch size.
-    const auto numDocsToInsert = 3;
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    // Insert documents in range.
-    setFilteringMetadataWithUUID(uuid());
-    auto task = insertRangeDeletionTask(_opCtx, uuid(), range, numDocsToInsert);
-    DBDirectClient dbclient(_opCtx);
-    for (auto i = 0; i < numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-    cleanupComplete.get();
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-}
-
-TEST_F(RangeDeleterTest,
-       RemoveDocumentsInRangeDoesNotRemoveDocumentsWithKeysLowerThanMinKeyOfRange) {
-    const auto numDocsToInsert = 3;
-
-    const auto minKey = 0;
-    const auto range = ChunkRange(BSON(kShardKey << minKey), BSON(kShardKey << 10));
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    setFilteringMetadataWithUUID(uuid());
-    auto task = insertRangeDeletionTask(_opCtx, uuid(), range, 0);
-    DBDirectClient dbclient(_opCtx);
-    // All documents below the range.
-    for (auto i = minKey - numDocsToInsert; i < minKey; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-    cleanupComplete.get();
-    // No documents should have been deleted.
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), numDocsToInsert);
-}
-
-TEST_F(RangeDeleterTest,
-       RemoveDocumentsInRangeDoesNotRemoveDocumentsWithKeysGreaterThanOrEqualToMaxKeyOfRange) {
-    const auto numDocsToInsert = 3;
-
-    const auto maxKey = 10;
-    const auto range = ChunkRange(BSON(kShardKey << 0), BSON(kShardKey << maxKey));
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    setFilteringMetadataWithUUID(uuid());
-    auto task = insertRangeDeletionTask(_opCtx, uuid(), range, 0);
-    DBDirectClient dbclient(_opCtx);
-    // All documents greater than or equal to the range.
-    for (auto i = maxKey; i < maxKey + numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-    cleanupComplete.get();
-    // No documents should have been deleted.
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), numDocsToInsert);
-}
-
-TEST_F(RangeDeleterTest,
-       RemoveDocumentsInRangeDoesNotRemoveDocumentsForCollectionWithSameNamespaceAndDifferentUUID) {
-    const auto numDocsToInsert = 3;
-    const auto range = ChunkRange(BSON(kShardKey << 0), BSON(kShardKey << 10));
-
-    setFilteringMetadataWithUUID(uuid());
-    auto task = insertRangeDeletionTask(_opCtx, uuid(), range, numDocsToInsert);
-    const auto collUuidWrongTaks = UUID::gen();
-    auto wrongTask = insertRangeDeletionTask(_opCtx, collUuidWrongTaks, range, numDocsToInsert);
-    DBDirectClient dbclient(_opCtx);
-    for (auto i = 0; i < numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               // Use a different UUID from the collection UUID.
-                               collUuidWrongTaks,
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-
-    ASSERT_THROWS_CODE(cleanupComplete.get(),
-                       DBException,
-                       ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist);
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), numDocsToInsert);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeThrowsErrorWhenCollectionDoesNotExist) {
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    const auto notExistingNss =
-        NamespaceString::createNamespaceString_forTest("someFake.namespace");
-    const auto notExistingCollectionUUID = UUID::gen();
-    const auto range = ChunkRange(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto task =
-        insertRangeDeletionTask(_opCtx, notExistingNss, notExistingCollectionUUID, range, 0);
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               notExistingNss,
-                               notExistingCollectionUUID,
-                               kShardKeyPattern,
-                               ChunkRange(BSON(kShardKey << 0), BSON(kShardKey << 10)),
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-
-    ASSERT_THROWS_CODE(cleanupComplete.get(),
-                       DBException,
-                       ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeLeavesDocumentsWhenTaskDocumentDoesNotExist) {
-    auto replCoord = checked_cast<repl::ReplicationCoordinatorMock*>(
-        repl::ReplicationCoordinator::get(getServiceContext()));
-
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    // We intentionally skip inserting a range deletion task document to simulate it already having
-    // been deleted.
-
-    // We should wait for replication after attempting to delete the document in the range even when
-    // the task document doesn't exist.
-    const auto expectedNumTimesWaitedForReplication = 1;
-    int numTimesWaitedForReplication = 0;
-
-    // Override special handler for waiting for replication to count the number of times we wait for
-    // replication.
-    replCoord->setAwaitReplicationReturnValueFunction(
-        [&](OperationContext* opCtx, const repl::OpTime& opTime) {
-            ++numTimesWaitedForReplication;
-            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-        });
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    cleanupComplete.get();
-
-    // Document should not have been deleted.
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 1);
-    ASSERT_EQ(numTimesWaitedForReplication, expectedNumTimesWaitedForReplication);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeWaitsForReplicationAfterDeletingSingleBatch) {
-    auto replCoord = checked_cast<repl::ReplicationCoordinatorMock*>(
-        repl::ReplicationCoordinator::get(getServiceContext()));
-
-    const auto numDocsToInsert = 3;
-    const auto numDocsToRemovePerBatch = 10;
-    rangeDeleterBatchSize.store(numDocsToRemovePerBatch);
-    const auto numBatches = ceil((double)numDocsToInsert / numDocsToRemovePerBatch);
-    ASSERT_EQ(numBatches, 1);
-    // We should wait twice: Once after deleting documents in the range, and once after deleting the
-    // range deletion task.
-    const auto expectedNumTimesWaitedForReplication = 2;
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    for (auto i = 0; i < numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    // Insert range deletion task for this collection and range.
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    int numTimesWaitedForReplication = 0;
-    // Override special handler for waiting for replication to count the number of times we wait for
-    // replication.
-    replCoord->setAwaitReplicationReturnValueFunction(
-        [&](OperationContext* opCtx, const repl::OpTime& opTime) {
-            ++numTimesWaitedForReplication;
-            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-        });
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-    cleanupComplete.get();
-
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-    ASSERT_EQ(numTimesWaitedForReplication, expectedNumTimesWaitedForReplication);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeWaitsForReplicationOnlyOnceAfterSeveralBatches) {
-    auto replCoord = checked_cast<repl::ReplicationCoordinatorMock*>(
-        repl::ReplicationCoordinator::get(getServiceContext()));
-
-    const auto numDocsToInsert = 3;
-    const auto numDocsToRemovePerBatch = 1;
-    rangeDeleterBatchSize.store(numDocsToRemovePerBatch);
-    const auto numBatches = ceil((double)numDocsToInsert / numDocsToRemovePerBatch);
-    ASSERT_GTE(numBatches, 1);
-
-    // We should wait twice: Once after deleting documents in the range, and once after deleting the
-    // range deletion task.
-    const auto expectedNumTimesWaitedForReplication = 2;
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    for (auto i = 0; i < numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    // Insert range deletion task for this collection and range.
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    int numTimesWaitedForReplication = 0;
-
-    // Set special handler for waiting for replication.
-    replCoord->setAwaitReplicationReturnValueFunction(
-        [&](OperationContext* opCtx, const repl::OpTime& opTime) {
-            ++numTimesWaitedForReplication;
-            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-        });
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    cleanupComplete.get();
-
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-    ASSERT_EQ(numTimesWaitedForReplication, expectedNumTimesWaitedForReplication);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeDoesNotWaitForReplicationIfErrorDuringDeletion) {
-    auto replCoord = checked_cast<repl::ReplicationCoordinatorMock*>(
-        repl::ReplicationCoordinator::get(getServiceContext()));
-
-    const auto numDocsToInsert = 3;
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    for (auto i = 0; i < numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    // Insert range deletion task for this collection and range.
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    int numTimesWaitedForReplication = 0;
-    // Override special handler for waiting for replication to count the number of times we wait for
-    // replication.
-    replCoord->setAwaitReplicationReturnValueFunction(
-        [&](OperationContext* opCtx, const repl::OpTime& opTime) {
-            ++numTimesWaitedForReplication;
-            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-        });
-
-    // Pretend we stepped down.
-    replCoord->setCanAcceptNonLocalWrites(false);
-    std::ignore = replCoord->setFollowerMode(repl::MemberState::RS_SECONDARY);
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-    ASSERT_THROWS_CODE(cleanupComplete.get(), DBException, ErrorCodes::PrimarySteppedDown);
-    ASSERT_EQ(numTimesWaitedForReplication, 0);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeRetriesOnWriteConflictException) {
-    // Enable fail point to throw WriteConflictException.
-    globalFailPointRegistry()
-        .find("throwWriteConflictExceptionInDeleteRange")
-        ->setMode(FailPoint::nTimes, 3 /* Throw a few times before disabling. */);
-
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    // Insert range deletion task for this collection and range.
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    cleanupComplete.get();
-
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeRetriesOnUnexpectedError) {
-    // Enable fail point to throw InternalError.
-    globalFailPointRegistry()
-        .find("throwInternalErrorInDeleteRange")
-        ->setMode(FailPoint::nTimes, 3 /* Throw a few times before disabling. */);
-
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    // Insert range deletion task for this collection and range.
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    cleanupComplete.get();
-
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-}
-
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeRespectsDelayInBetweenBatches) {
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    // More documents than the batch size.
-    const auto numDocsToInsert = 3;
-    const auto numDocsToRemovePerBatch = 1;
-    rangeDeleterBatchSize.store(numDocsToRemovePerBatch);
-
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    // Insert documents in range.
-    setFilteringMetadataWithUUID(uuid());
-    auto task = insertRangeDeletionTask(_opCtx, uuid(), range, numDocsToInsert);
-    DBDirectClient dbclient(_opCtx);
-    for (auto i = 0; i < numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    // The deletion of a document in unit tests with ephemeral storage engine is usually
-    // extremely fast (less than 5ms), so setting the delay to 1 second ensures the test
-    // is relevant: it is very improbable for a deletion to last so much, even on slow
-    // machines.
-    const auto delayBetweenBatchesMS = 1000 /* 1 second */;
-    rangeDeleterBatchDelayMS.store(delayBetweenBatchesMS);
-
-    auto beforeRangeDeletion = Date_t::now();
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    cleanupComplete.get();
-    auto afterRangeDeletion = Date_t::now();
-    auto rangeDeletionTimeMS =
-        afterRangeDeletion.toMillisSinceEpoch() - beforeRangeDeletion.toMillisSinceEpoch();
-    ASSERT(rangeDeletionTimeMS >= delayBetweenBatchesMS * numDocsToInsert);
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeRespectsOrphanCleanupDelay) {
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    // More documents than the batch size.
-    const auto numDocsToInsert = 3;
-    const auto orphanCleanupDelay = Seconds(10);
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    // Insert documents in range.
-    setFilteringMetadataWithUUID(uuid());
-    auto task = insertRangeDeletionTask(_opCtx, uuid(), range, numDocsToInsert);
-    DBDirectClient dbclient(_opCtx);
-    for (auto i = 0; i < numDocsToInsert; ++i) {
-        dbclient.insert(kNss, BSON(kShardKey << i));
-    }
-
-    auto cleanupComplete = removeDocumentsInRange(executor(),
-                                                  std::move(queriesComplete),
-                                                  kNss,
-                                                  uuid(),
-                                                  kShardKeyPattern,
-                                                  range,
-                                                  orphanCleanupDelay);
-
-    // A best-effort check that cleanup has not completed without advancing the clock.
-    sleepsecs(1);
-    ASSERT_FALSE(cleanupComplete.isReady());
-
-    // Advance the time past the delay until cleanup is complete. This cannot be made exact because
-    // there's no way to tell when the sleep operation gets hit exactly, so instead we incrementally
-    // advance time until it's ready.
-    while (!cleanupComplete.isReady()) {
-        executor::NetworkInterfaceMock::InNetworkGuard guard(network());
-        network()->advanceTime(network()->now() + orphanCleanupDelay);
-    }
-
-    cleanupComplete.get();
-
-    ASSERT_EQUALS(dbclient.count(kNss, BSONObj()), 0);
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeRemovesRangeDeletionTaskOnSuccess) {
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    // Insert range deletion task for this collection and range.
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    cleanupComplete.get();
-    // Document should have been deleted.
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-    ASSERT_EQUALS(countDocsInConfigRangeDeletions(store, _opCtx), 0);
-}
-
-TEST_F(RangeDeleterTest,
-       RemoveDocumentsInRangeRemovesRangeDeletionTaskOnCollectionDroppedErrorWhenStillPrimary) {
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    // Insert range deletion task for this collection and range.
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    dbclient.dropCollection(kNss);
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    ASSERT_THROWS_CODE(cleanupComplete.get(),
-                       DBException,
-                       ErrorCodes::RangeDeletionAbandonedBecauseCollectionWithUUIDDoesNotExist);
-
-    // Document should have been deleted.
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-    ASSERT_EQUALS(countDocsInConfigRangeDeletions(store, _opCtx), 0);
-}
-
-TEST_F(RangeDeleterTest,
-       RemoveDocumentsInRangeDoesNotRemoveRangeDeletionTaskOnErrorWhenNotStillPrimary) {
-    const ChunkRange range(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    auto queriesComplete = SemiFuture<void>::makeReady();
-
-    setFilteringMetadataWithUUID(uuid());
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    // Insert range deletion task for this collection and range.
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    // Pretend we stepped down.
-    auto replCoord = checked_cast<repl::ReplicationCoordinatorMock*>(
-        repl::ReplicationCoordinator::get(getServiceContext()));
-    replCoord->setCanAcceptNonLocalWrites(false);
-    std::ignore = replCoord->setFollowerMode(repl::MemberState::RS_SECONDARY);
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-    ASSERT_THROWS_CODE(cleanupComplete.get(), DBException, ErrorCodes::PrimarySteppedDown);
-
-    // Pretend we stepped back up so we can read the task store.
-    replCoord->setCanAcceptNonLocalWrites(true);
-    std::ignore = replCoord->setFollowerMode(repl::MemberState::RS_PRIMARY);
-
-    // Document should not have been deleted.
-    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
-    ASSERT_EQUALS(countDocsInConfigRangeDeletions(store, _opCtx), 1);
-}
-
-// The input future should never have an error.
-DEATH_TEST_F(RangeDeleterTest, RemoveDocumentsInRangeCrashesIfInputFutureHasError, "invariant") {
-    const ChunkRange range = ChunkRange(BSON(kShardKey << 0), BSON(kShardKey << 10));
-    DBDirectClient dbclient(_opCtx);
-    dbclient.insert(kNss, BSON(kShardKey << 5));
-
-    // Insert range deletion task for this collection and range.
-    auto t = insertRangeDeletionTask(_opCtx, uuid(), range);
-
-    auto queriesCompletePf = makePromiseFuture<void>();
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move((queriesCompletePf.future)).semi(),
-                               kNss,
-                               uuid(),
-                               kShardKeyPattern,
-                               range,
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete */);
-
-
-    // Should cause an invariant failure.
-    queriesCompletePf.promise.setError(Status(ErrorCodes::InternalError, "Some unexpected error"));
-    cleanupComplete.get();
-}
-
-TEST_F(RangeDeleterTest, RemoveDocumentsInRangeDoesNotCrashWhenShardKeyIndexDoesNotExist) {
-    auto queriesComplete = SemiFuture<void>::makeReady();
-    const std::string kNoShardKeyIndexMsg("Unable to find shard key index for");
-    auto logCountBefore = countTextFormatLogLinesContaining(kNoShardKeyIndexMsg);
-
-    auto cleanupComplete =
-        removeDocumentsInRange(executor(),
-                               std::move(queriesComplete),
-                               kNss,
-                               uuid(),
-                               BSON("x" << 1) /* shard key pattern */,
-                               ChunkRange(BSON("x" << 0), BSON("x" << 10)),
-                               Seconds(0) /* delayForActiveQueriesOnSecondariesToComplete*/);
-
-    // Range deleter will keep on retrying when it encounters non-stepdown errors. Make it run
-    // a few iterations and then create the index to make it exit the retry loop.
-    while (countTextFormatLogLinesContaining(kNoShardKeyIndexMsg) < logCountBefore) {
-        sleepmicros(100);
-    }
-
-    DBDirectClient client(_opCtx);
-    client.createIndex(kNss, BSON("x" << 1));
-
-    cleanupComplete.get();
-}
-
 /**
  *  Tests that the rename range deletion flow:
  *  - Renames range deletions from source to target collection
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service_external_state_test.cpp b/src/mongo/db/s/resharding/resharding_recipient_service_external_state_test.cpp
index cf01d4db953..db9e0d2da51 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service_external_state_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service_external_state_test.cpp
@@ -263,7 +263,7 @@ public:
         RecipientStateMachineExternalStateImpl externalState;
         externalState.ensureTempReshardingCollectionExistsWithIndexes(
             operationContext(), kMetadata, kDefaultFetchTimestamp);
-        CollectionShardingRuntime csr(getServiceContext(), kOrigNss, executor());
+        CollectionShardingRuntime csr(getServiceContext(), kOrigNss);
         ASSERT(csr.getCurrentMetadataIfKnown() == boost::none);
     }
 };
diff --git a/src/mongo/db/s/shard_server_op_observer.cpp b/src/mongo/db/s/shard_server_op_observer.cpp
index d324610decc..cc4dce227d6 100644
--- a/src/mongo/db/s/shard_server_op_observer.cpp
+++ b/src/mongo/db/s/shard_server_op_observer.cpp
@@ -101,31 +101,6 @@ private:
 };
 
 /**
- * Used to submit a range deletion task once it is certain that the update/insert to
- * config.rangeDeletions is committed.
- */
-class SubmitRangeDeletionHandler final : public RecoveryUnit::Change {
-public:
-    SubmitRangeDeletionHandler(OperationContext* opCtx, RangeDeletionTask task)
-        : _opCtx(opCtx), _task(std::move(task)) {}
-
-    void commit(OperationContext* opCtx, boost::optional<Timestamp>) override {
-        // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The feature
-        // flag is used to turn on new range deleter on startup.
-        if (!feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-            migrationutil::submitRangeDeletionTask(_opCtx, _task).getAsync([](auto) {});
-        }
-    }
-
-    void rollback(OperationContext* opCtx) override {}
-
-private:
-    OperationContext* _opCtx;
-    RangeDeletionTask _task;
-};
-
-
-/**
  * Invalidates the in-memory routing table cache when a collection is dropped, so the next caller
  * with routing information will provoke a routing table refresh and see the drop.
  *
@@ -220,11 +195,6 @@ void ShardServerOpObserver::onInserts(OperationContext* opCtx,
             auto deletionTask =
                 RangeDeletionTask::parse(IDLParserContext("ShardServerOpObserver"), insertedDoc);
 
-            if (!deletionTask.getPending()) {
-                opCtx->recoveryUnit()->registerChange(
-                    std::make_unique<SubmitRangeDeletionHandler>(opCtx, deletionTask));
-            }
-
             const auto numOrphanDocs = deletionTask.getNumOrphanDocs();
             BalancerStatsRegistry::get(opCtx)->onRangeDeletionTaskInsertion(
                 deletionTask.getCollectionUuid(), numOrphanDocs);
@@ -372,26 +342,6 @@ void ShardServerOpObserver::onUpdate(OperationContext* opCtx,
         }
     }
 
-    if (needsSpecialHandling && args.coll->ns() == NamespaceString::kRangeDeletionNamespace) {
-        if (!isStandaloneOrPrimary(opCtx))
-            return;
-
-        const auto pendingFieldRemovedStatus =
-            update_oplog_entry::isFieldRemovedByUpdate(args.updateArgs->update, "pending");
-
-        if (pendingFieldRemovedStatus == update_oplog_entry::FieldRemovedStatus::kFieldRemoved) {
-            auto deletionTask = RangeDeletionTask::parse(IDLParserContext("ShardServerOpObserver"),
-                                                         args.updateArgs->updatedDoc);
-
-            if (deletionTask.getDonorShardId() != ShardingState::get(opCtx)->shardId()) {
-                // Range deletion tasks for moved away chunks are scheduled through the
-                // MigrationCoordinator, so only schedule a task for received chunks.
-                opCtx->recoveryUnit()->registerChange(
-                    std::make_unique<SubmitRangeDeletionHandler>(opCtx, deletionTask));
-            }
-        }
-    }
-
     if (args.coll->ns() == NamespaceString::kCollectionCriticalSectionsNamespace &&
         !sharding_recovery_util::inRecoveryMode(opCtx)) {
         const auto collCSDoc = CollectionCriticalSectionDocument::parse(
diff --git a/src/mongo/db/s/sharding_server_status.cpp b/src/mongo/db/s/sharding_server_status.cpp
index 51ade66b553..343a9e3061a 100644
--- a/src/mongo/db/s/sharding_server_status.cpp
+++ b/src/mongo/db/s/sharding_server_status.cpp
@@ -128,20 +128,14 @@ public:
 
             ShardingStatistics::get(opCtx).report(&result);
             catalogCache->report(&result);
-            // (Ignore FCV check): This feature doesn't have any upgrade/downgrade concerns. The
-            // feature flag is used to turn on new range deleter on startup.
-            if (mongo::feature_flags::gRangeDeleterService.isEnabledAndIgnoreFCVUnsafe()) {
-                auto nRangeDeletions = [&]() {
-                    try {
-                        return RangeDeleterService::get(opCtx)->totalNumOfRegisteredTasks();
-                    } catch (const ExceptionFor<ErrorCodes::NotYetInitialized>&) {
-                        return 0LL;
-                    }
-                }();
-                result.appendNumber("rangeDeleterTasks", nRangeDeletions);
-            }
-
-            CollectionShardingState::appendInfoForServerStatus(opCtx, &result);
+            auto nRangeDeletions = [&]() {
+                try {
+                    return RangeDeleterService::get(opCtx)->totalNumOfRegisteredTasks();
+                } catch (const ExceptionFor<ErrorCodes::NotYetInitialized>&) {
+                    return 0LL;
+                }
+            }();
+            result.appendNumber("rangeDeleterTasks", nRangeDeletions);
         }
 
         // To calculate the number of sharded collection we simply get the number of records from