Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

47 files changed, 413 insertions, 982 deletions

diff --git a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp
index 2695ed842c0..75cb7be7049 100644
--- a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp
+++ b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp
@@ -101,7 +101,8 @@ StageConstraints DocumentSourceReshardingAddResumeId::constraints(
                             ChangeStreamRequirement::kDenylist);
 }
 
-Value DocumentSourceReshardingAddResumeId::serialize(const SerializationOptions& opts) const {
+Value DocumentSourceReshardingAddResumeId::serialize(
+    boost::optional<ExplainOptions::Verbosity> explain) const {
     return Value(Document{{kStageName, Value(Document{})}});
 }
 
diff --git a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h
index 4fb27980c68..31cbd97c694 100644
--- a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h
+++ b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h
@@ -53,7 +53,7 @@ public:
 
     DocumentSource::GetModPathsReturn getModifiedPaths() const final;
 
-    Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override;
+    Value serialize(boost::optional<ExplainOptions::Verbosity> explain) const;
 
     StageConstraints constraints(Pipeline::SplitState pipeState) const final;
 
diff --git a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp
index 8260cf3e4cc..9d412147b7f 100644
--- a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp
+++ b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp
@@ -112,7 +112,7 @@ StageConstraints DocumentSourceReshardingIterateTransaction::constraints(
 }
 
 Value DocumentSourceReshardingIterateTransaction::serialize(
-    const SerializationOptions& opts) const {
+    boost::optional<ExplainOptions::Verbosity> explain) const {
     return Value(
         Document{{kStageName,
                   Value(Document{{kIncludeCommitTransactionTimestampFieldName,
@@ -134,7 +134,7 @@ DepsTracker::State DocumentSourceReshardingIterateTransaction::getDependencies(
 
 DocumentSource::GetModPathsReturn DocumentSourceReshardingIterateTransaction::getModifiedPaths()
     const {
-    return {DocumentSource::GetModPathsReturn::Type::kAllPaths, OrderedPathSet{}, {}};
+    return {DocumentSource::GetModPathsReturn::Type::kAllPaths, std::set<std::string>{}, {}};
 }
 
 DocumentSource::GetNextResult DocumentSourceReshardingIterateTransaction::doGetNext() {
diff --git a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h
index 8c792116e6c..9589cb64a08 100644
--- a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h
+++ b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h
@@ -66,7 +66,7 @@ public:
 
     DocumentSource::GetModPathsReturn getModifiedPaths() const final;
 
-    Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override;
+    Value serialize(boost::optional<ExplainOptions::Verbosity> explain) const;
 
     StageConstraints constraints(Pipeline::SplitState pipeState) const final;
 
diff --git a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp
index 157876391d3..f8a186e3bcd 100644
--- a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp
+++ b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp
@@ -90,11 +90,12 @@ StageConstraints DocumentSourceReshardingOwnershipMatch::constraints(
                             ChangeStreamRequirement::kDenylist);
 }
 
-Value DocumentSourceReshardingOwnershipMatch::serialize(const SerializationOptions& opts) const {
+Value DocumentSourceReshardingOwnershipMatch::serialize(
+    boost::optional<ExplainOptions::Verbosity> explain) const {
     return Value{Document{{kStageName,
                            DocumentSourceReshardingOwnershipMatchSpec(
                                _recipientShardId, _reshardingKey.getKeyPattern())
-                               .toBSON(opts)}}};
+                               .toBSON()}}};
 }
 
 DepsTracker::State DocumentSourceReshardingOwnershipMatch::getDependencies(
@@ -108,7 +109,7 @@ DepsTracker::State DocumentSourceReshardingOwnershipMatch::getDependencies(
 
 DocumentSource::GetModPathsReturn DocumentSourceReshardingOwnershipMatch::getModifiedPaths() const {
     // This stage does not modify or rename any paths.
-    return {DocumentSource::GetModPathsReturn::Type::kFiniteSet, OrderedPathSet{}, {}};
+    return {DocumentSource::GetModPathsReturn::Type::kFiniteSet, std::set<std::string>{}, {}};
 }
 
 DocumentSource::GetNextResult DocumentSourceReshardingOwnershipMatch::doGetNext() {
diff --git a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h
index b7da07a5a57..7a6db2bc125 100644
--- a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h
+++ b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h
@@ -58,7 +58,7 @@ public:
 
     DocumentSource::GetModPathsReturn getModifiedPaths() const final;
 
-    Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override;
+    Value serialize(boost::optional<ExplainOptions::Verbosity> explain) const final;
 
     StageConstraints constraints(Pipeline::SplitState pipeState) const final;
 
diff --git a/src/mongo/db/s/resharding/recipient_document.idl b/src/mongo/db/s/resharding/recipient_document.idl
index 1eda4620b8c..e3128de0db5 100644
--- a/src/mongo/db/s/resharding/recipient_document.idl
+++ b/src/mongo/db/s/resharding/recipient_document.idl
@@ -82,11 +82,6 @@ structs:
             startConfigTxnCloneTime:
                 type:  date
                 optional: true
-            approxBytesToCopy:
-                type: long
-                description: >-
-                    Approximate number of bytes to copy during cloning
-                optional: true
             metrics:
                 type: ReshardingRecipientMetrics
                 description: "Metrics related to this recipient."
diff --git a/src/mongo/db/s/resharding/resharding_agg_test.cpp b/src/mongo/db/s/resharding/resharding_agg_test.cpp
index 99ad259c7d0..f2c5345e907 100644
--- a/src/mongo/db/s/resharding/resharding_agg_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_agg_test.cpp
@@ -239,7 +239,6 @@ repl::DurableOplogEntry makeApplyOpsOplog(std::vector<BSONObj> operations,
             {},
             UUID::gen(),
             false /* fromMigrate */,
-            boost::none,  // checkExistenceForDiffInsert
             0 /* version */,
             applyOpsBuilder.obj(), /* o */
             boost::none,           /* o2 */
@@ -1452,7 +1451,6 @@ TEST_F(ReshardingAggWithStorageTest, RetryableFindAndModifyWithImageLookup) {
                                   kCrudNs,
                                   kCrudUUID,
                                   false /* fromMigrate */,
-                                  boost::none,  // checkExistenceForDiffInsert
                                   0 /* version */,
                                   BSON("$set" << BSON("y" << 1)), /* o1 */
                                   BSON("_id" << 2),               /* o2 */
diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
index e51d2e2ed1f..9132438053a 100644
--- a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
+++ b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
@@ -29,6 +29,8 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kResharding
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/s/resharding/resharding_collection_cloner.h"
 
 #include <utility>
@@ -37,6 +39,7 @@
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/client.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/exec/document_value/document.h"
 #include "mongo/db/logical_session_id_helpers.h"
@@ -274,7 +277,7 @@ bool ReshardingCollectionCloner::doOneBatch(OperationContext* opCtx, Pipeline& p
 
     // ReshardingOpObserver depends on the collection metadata being known when processing writes to
     // the temporary resharding collection. We attach shard version IGNORED to the insert operations
-    // and retry once on a StaleConfig error to allow the collection metadata information to be
+    // and retry once on a StaleConfig exception to allow the collection metadata information to be
     // recovered.
     ScopedSetShardRole scopedSetShardRole(opCtx,
                                           _outputNss,
diff --git a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp
index 5700f0326ae..374dcd6538f 100644
--- a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp
+++ b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp
@@ -164,21 +164,13 @@ CoordinatorCommitMonitor::queryRemainingOperationTimeForRecipients() const {
         uassertStatusOKWithContext(status, errorContext);
 
         const auto remainingTime = extractOperationRemainingTime(shardResponse.data);
-
-        // If any recipient omits the "remainingMillis" field of the response then
-        // we cannot conclude that it is safe to begin the critical section.
-        // It is possible that the recipient just had a failover and
-        // was not able to restore its metrics before it replied to the
-        // _shardsvrReshardingOperationTime command.
-        if (!remainingTime) {
-            maxRemainingTime = Milliseconds::max();
-            continue;
-        }
-        if (remainingTime.value() < minRemainingTime) {
-            minRemainingTime = remainingTime.value();
+        // A recipient shard does not report the remaining operation time when there is no data
+        // to copy and no oplog entry to apply.
+        if (remainingTime && remainingTime.get() < minRemainingTime) {
+            minRemainingTime = remainingTime.get();
         }
-        if (remainingTime.value() > maxRemainingTime) {
-            maxRemainingTime = remainingTime.value();
+        if (remainingTime && remainingTime.get() > maxRemainingTime) {
+            maxRemainingTime = remainingTime.get();
         }
     }
 
@@ -211,20 +203,12 @@ ExecutorFuture<void> CoordinatorCommitMonitor::_makeFuture() const {
                           "Encountered an error while querying recipients, will retry shortly",
                           "error"_attr = status);
 
-            // On error we definitely cannot begin the critical section.  Therefore,
-            // return Milliseconds::max for remainingTimes.max (remainingTimes.max is used
-            // for determining whether the critical section should begin).
-            return RemainingOperationTimes{Milliseconds(-1), Milliseconds::max()};
+            return RemainingOperationTimes{Milliseconds(0), Milliseconds::max()};
         })
         .then([this, anchor = shared_from_this()](RemainingOperationTimes remainingTimes) {
             auto metrics = ReshardingMetrics::get(cc().getServiceContext());
-            // If remainingTimes.max (or remainingTimes.min) is Milliseconds::max, then use -1 so
-            // that the scale of the y-axis is still useful when looking at FTDC metrics.
-            auto clampIfMax = [](Milliseconds t) {
-                return t != Milliseconds::max() ? t : Milliseconds(-1);
-            };
-            metrics->setMinRemainingOperationTime(clampIfMax(remainingTimes.min));
-            metrics->setMaxRemainingOperationTime(clampIfMax(remainingTimes.max));
+            metrics->setMinRemainingOperationTime(remainingTimes.min);
+            metrics->setMaxRemainingOperationTime(remainingTimes.max);
 
             // Check if all recipient shards are within the commit threshold.
             if (remainingTimes.max <= _threshold)
diff --git a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp
index 2fe3075f1fc..e7f90cc41fa 100644
--- a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp
@@ -92,8 +92,6 @@ protected:
     void tearDown() override;
 
     void mockCommandForRecipients(Milliseconds remainingOperationTime);
-    void mockOmitRemainingMillisForRecipients();
-    void mockOmitRemainingMillisForOneRecipient();
     void mockRemaingOperationTimesCommandForRecipients(
         CoordinatorCommitMonitor::RemainingOperationTimes remainingOperationTimes);
 
@@ -182,31 +180,6 @@ void CoordinatorCommitMonitorTest::mockCommandForRecipients(Milliseconds remaini
         _recipientShards.begin(), _recipientShards.end(), [&](const ShardId&) { onCommand(func); });
 }
 
-void CoordinatorCommitMonitorTest::mockOmitRemainingMillisForRecipients() {
-    // Omit remainingMillis from all shard responses.
-    std::for_each(_recipientShards.begin(), _recipientShards.end(), [this](const ShardId&) {
-        onCommand([](const executor::RemoteCommandRequest& request) -> StatusWith<BSONObj> {
-            // Return an empty BSON object.
-            return BSONObj();
-        });
-    });
-}
-
-void CoordinatorCommitMonitorTest::mockOmitRemainingMillisForOneRecipient() {
-    // Omit remainingMillis from a single recipient.
-    for (const auto& shard : _recipientShards) {
-        onCommand([&](const executor::RemoteCommandRequest&) -> StatusWith<BSONObj> {
-            if (shard == _recipientShards.front()) {
-                // Return an empty BSON object.
-                return BSONObj();
-            }
-            auto threshold = Milliseconds(gRemainingReshardingOperationTimeThresholdMillis.load());
-            return BSON("remainingMillis"
-                        << durationCount<Milliseconds>(threshold - Milliseconds(1)));
-        });
-    }
-}
-
 void CoordinatorCommitMonitorTest::mockRemaingOperationTimesCommandForRecipients(
     CoordinatorCommitMonitor::RemainingOperationTimes remainingOperationTimes) {
     bool useMin = true;
@@ -291,20 +264,6 @@ TEST_F(CoordinatorCommitMonitorTest, RetriesWhenEncountersErrorsWhileQueryingRec
     future.get();
 }
 
-TEST_F(CoordinatorCommitMonitorTest, BlocksWhenRemainingMillisIsOmitted) {
-    auto future = getCommitMonitor()->waitUntilRecipientsAreWithinCommitThreshold();
-
-    mockOmitRemainingMillisForRecipients();
-    ASSERT(!future.isReady());
-
-    // If even a single shard omits remainingMillis, we cannot begin the critical section.
-    mockOmitRemainingMillisForOneRecipient();
-    ASSERT(!future.isReady());
-
-    respondWithReadyToCommit();
-    future.get();
-}
-
 }  // namespace
 }  // namespace resharding
 }  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service.cpp b/src/mongo/db/s/resharding/resharding_coordinator_service.cpp
index ea3451f1c4a..85c14b6478f 100644
--- a/src/mongo/db/s/resharding/resharding_coordinator_service.cpp
+++ b/src/mongo/db/s/resharding/resharding_coordinator_service.cpp
@@ -37,7 +37,6 @@
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/json.h"
 #include "mongo/db/auth/authorization_session_impl.h"
-#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/logical_session_cache.h"
 #include "mongo/db/ops/write_ops.h"
@@ -95,7 +94,6 @@ MONGO_FAIL_POINT_DEFINE(reshardingPauseCoordinatorBeforeStartingErrorFlow);
 MONGO_FAIL_POINT_DEFINE(reshardingPauseCoordinatorBeforePersistingStateTransition);
 MONGO_FAIL_POINT_DEFINE(pauseBeforeTellDonorToRefresh);
 MONGO_FAIL_POINT_DEFINE(pauseBeforeInsertCoordinatorDoc);
-MONGO_FAIL_POINT_DEFINE(pauseBeforeCTHolderInitialization);
 
 const std::string kReshardingCoordinatorActiveIndexName = "ReshardingCoordinatorActiveIndex";
 const Backoff kExponentialBackoff(Seconds(1), Milliseconds::max());
@@ -111,14 +109,14 @@ Date_t getCurrentTime() {
     return svcCtx->getFastClockSource()->now();
 }
 
-void assertNumDocsMatchedEqualsExpected(const BatchedCommandRequest& request,
-                                        const BSONObj& response,
-                                        int expected) {
-    auto numDocsMatched = response.getIntField("n");
+void assertNumDocsModifiedMatchesExpected(const BatchedCommandRequest& request,
+                                          const BSONObj& response,
+                                          int expected) {
+    auto numDocsModified = response.getIntField("n");
     uassert(5030401,
             str::stream() << "Expected to match " << expected << " docs, but only matched "
-                          << numDocsMatched << " for write request " << request.toString(),
-            expected == numDocsMatched);
+                          << numDocsModified << " for write request " << request.toString(),
+            expected == numDocsModified);
 }
 
 void appendShardEntriesToSetBuilder(const ReshardingCoordinatorDocument& coordinatorDoc,
@@ -217,15 +215,15 @@ void writeToCoordinatorStateNss(OperationContext* opCtx,
         }
     }());
 
-    auto expectedNumMatched = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert)
+    auto expectedNumModified = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert)
         ? boost::none
         : boost::make_optional(1);
 
     auto res = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn(
         opCtx, NamespaceString::kConfigReshardingOperationsNamespace, request, txnNumber);
 
-    if (expectedNumMatched) {
-        assertNumDocsMatchedEqualsExpected(request, res, *expectedNumMatched);
+    if (expectedNumModified) {
+        assertNumDocsModifiedMatchesExpected(request, res, *expectedNumModified);
     }
 }
 
@@ -387,7 +385,7 @@ void updateConfigCollectionsForOriginalNss(OperationContext* opCtx,
     auto res = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn(
         opCtx, CollectionType::ConfigNS, request, txnNumber);
 
-    assertNumDocsMatchedEqualsExpected(request, res, 1 /* expected */);
+    assertNumDocsModifiedMatchesExpected(request, res, 1 /* expected */);
 }
 
 void writeToConfigCollectionsForTempNss(OperationContext* opCtx,
@@ -478,15 +476,15 @@ void writeToConfigCollectionsForTempNss(OperationContext* opCtx,
         }
     }());
 
-    auto expectedNumMatched = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert)
+    auto expectedNumModified = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert)
         ? boost::none
         : boost::make_optional(1);
 
     auto res = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn(
         opCtx, CollectionType::ConfigNS, request, txnNumber);
 
-    if (expectedNumMatched) {
-        assertNumDocsMatchedEqualsExpected(request, res, *expectedNumMatched);
+    if (expectedNumModified) {
+        assertNumDocsModifiedMatchesExpected(request, res, *expectedNumModified);
     }
 }
 
@@ -507,33 +505,43 @@ void insertChunkAndTagDocsForTempNss(OperationContext* opCtx,
     ShardingCatalogManager::get(opCtx)->insertConfigDocuments(opCtx, TagsType::ConfigNS, newZones);
 }
 
-void removeTagsDocs(OperationContext* opCtx, const BSONObj& tagsQuery, TxnNumber txnNumber) {
-    // Remove tag documents with the specified tagsQuery.
-    const auto tagDeleteOperationHint = BSON(TagsType::ns() << 1 << TagsType::min() << 1);
-    ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn(
-        opCtx,
-        TagsType::ConfigNS,
-        BatchedCommandRequest::buildDeleteOp(TagsType::ConfigNS,
-                                             tagsQuery,              // query
-                                             true,                   // multi
-                                             tagDeleteOperationHint  // hint
-                                             ),
-        txnNumber);
-}
-
 // Requires that there be no session information on the opCtx.
 void removeChunkAndTagsDocs(OperationContext* opCtx,
                             const BSONObj& tagsQuery,
                             const UUID& collUUID) {
-    // Remove all chunk documents and specified tag documents.
-    resharding::removeChunkDocs(opCtx, collUUID);
-
+    // Remove all chunk documents for the original nss. We do not know how many chunk docs
+    // currently exist, so cannot pass a value for expectedNumModified
+    const auto chunksQuery = BSON(ChunkType::collectionUUID() << collUUID);
     const auto tagDeleteOperationHint = BSON(TagsType::ns() << 1 << TagsType::min() << 1);
+
     const auto catalogClient = Grid::get(opCtx)->catalogClient();
+
+    uassertStatusOK(catalogClient->removeConfigDocuments(
+        opCtx, ChunkType::ConfigNS, chunksQuery, kMajorityWriteConcern));
     uassertStatusOK(catalogClient->removeConfigDocuments(
         opCtx, TagsType::ConfigNS, tagsQuery, kMajorityWriteConcern, tagDeleteOperationHint));
 }
 
+void updateChunkAndTagsDocsForTempNss(OperationContext* opCtx,
+                                      const ReshardingCoordinatorDocument& coordinatorDoc,
+                                      OID newCollectionEpoch,
+                                      TxnNumber txnNumber) {
+    auto hint = BSON("ns" << 1 << "min" << 1);
+    auto tagsRequest = BatchedCommandRequest::buildUpdateOp(
+        TagsType::ConfigNS,
+        BSON(TagsType::ns(coordinatorDoc.getTempReshardingNss().ns())),    // query
+        BSON("$set" << BSON("ns" << coordinatorDoc.getSourceNss().ns())),  // update
+        false,                                                             // upsert
+        true,                                                              // multi
+        hint                                                               // hint
+    );
+
+    // Update the 'ns' field to be the original collection namespace for all tags documents that
+    // currently have 'ns' as the temporary collection namespace
+    auto tagsRes = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn(
+        opCtx, TagsType::ConfigNS, tagsRequest, txnNumber);
+}
+
 /**
  * Executes metadata changes in a transaction without bumping the collection version.
  */
@@ -584,14 +592,28 @@ CollectionType createTempReshardingCollectionType(
     return collType;
 }
 
-void removeChunkDocs(OperationContext* opCtx, const UUID& collUUID) {
-    // Remove all chunk documents for the specified collUUID. We do not know how many chunk docs
-    // currently exist, so cannot pass a value for expectedNumModified
-    const auto chunksQuery = BSON(ChunkType::collectionUUID() << collUUID);
-    const auto catalogClient = Grid::get(opCtx)->catalogClient();
+void cleanupSourceConfigCollections(OperationContext* opCtx,
+                                    const ReshardingCoordinatorDocument& coordinatorDoc) {
+    using Doc = Document;
+    using Arr = std::vector<Value>;
+    using V = Value;
 
-    uassertStatusOK(catalogClient->removeConfigDocuments(
-        opCtx, ChunkType::ConfigNS, chunksQuery, kMajorityWriteConcern));
+    auto createTagFilter = [](const V value) {
+        return V{Doc{{"$map",
+                      V{Doc{{"input", V{Doc{{"$objectToArray", value}}}},
+                            {"in", V{StringData("$$this.k")}}}}}}};
+    };
+
+
+    auto skipNewTagsFilter = Doc{
+        {"$ne",
+         Arr{createTagFilter(V{StringData("$min")}),
+             createTagFilter(V{Doc{{"$literal", coordinatorDoc.getReshardingKey().toBSON()}}})}}};
+
+    const auto removeTagsQuery =
+        BSON(TagsType::ns(coordinatorDoc.getSourceNss().ns()) << "$expr" << skipNewTagsFilter);
+
+    removeChunkAndTagsDocs(opCtx, removeTagsQuery, coordinatorDoc.getSourceUUID());
 }
 
 void writeDecisionPersistedState(OperationContext* opCtx,
@@ -600,49 +622,21 @@ void writeDecisionPersistedState(OperationContext* opCtx,
                                  Timestamp newCollectionTimestamp) {
 
     // No need to bump originalNss version because its epoch will be changed.
-    executeMetadataChangesInTxn(
-        opCtx,
-        [&coordinatorDoc, &newCollectionEpoch, &newCollectionTimestamp](OperationContext* opCtx,
-                                                                        TxnNumber txnNumber) {
-            // Update the config.reshardingOperations entry
-            writeToCoordinatorStateNss(opCtx, coordinatorDoc, txnNumber);
+    executeMetadataChangesInTxn(opCtx, [&](OperationContext* opCtx, TxnNumber txnNumber) {
+        // Update the config.reshardingOperations entry
+        writeToCoordinatorStateNss(opCtx, coordinatorDoc, txnNumber);
 
-            // Remove the config.collections entry for the temporary collection
-            writeToConfigCollectionsForTempNss(
-                opCtx, coordinatorDoc, boost::none, boost::none, txnNumber);
-
-            // Update the config.collections entry for the original namespace to reflect the new
-            // shard key, new epoch, and new UUID
-            updateConfigCollectionsForOriginalNss(
-                opCtx, coordinatorDoc, newCollectionEpoch, newCollectionTimestamp, txnNumber);
+        // Remove the config.collections entry for the temporary collection
+        writeToConfigCollectionsForTempNss(
+            opCtx, coordinatorDoc, boost::none, boost::none, txnNumber);
 
-            // Delete all of the config.tags entries for the user collection namespace.
-            const auto removeTagsQuery = BSON(TagsType::ns(coordinatorDoc.getSourceNss().ns()));
-            removeTagsDocs(opCtx, removeTagsQuery, txnNumber);
+        // Update the config.collections entry for the original namespace to reflect the new
+        // shard key, new epoch, and new UUID
+        updateConfigCollectionsForOriginalNss(
+            opCtx, coordinatorDoc, newCollectionEpoch, newCollectionTimestamp, txnNumber);
 
-            // Update all of the config.tags entries for the temporary resharding namespace
-            // to refer to the user collection namespace.
-            updateTagsDocsForTempNss(opCtx, coordinatorDoc, txnNumber);
-        });
-}
-
-void updateTagsDocsForTempNss(OperationContext* opCtx,
-                              const ReshardingCoordinatorDocument& coordinatorDoc,
-                              TxnNumber txnNumber) {
-    auto hint = BSON("ns" << 1 << "min" << 1);
-    auto tagsRequest = BatchedCommandRequest::buildUpdateOp(
-        TagsType::ConfigNS,
-        BSON(TagsType::ns(coordinatorDoc.getTempReshardingNss().ns())),    // query
-        BSON("$set" << BSON("ns" << coordinatorDoc.getSourceNss().ns())),  // update
-        false,                                                             // upsert
-        true,                                                              // multi
-        hint                                                               // hint
-    );
-
-    // Update the 'ns' field to be the original collection namespace for all tags documents that
-    // currently have 'ns' as the temporary collection namespace.
-    auto tagsRes = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn(
-        opCtx, TagsType::ConfigNS, tagsRequest, txnNumber);
+        updateChunkAndTagsDocsForTempNss(opCtx, coordinatorDoc, newCollectionEpoch, txnNumber);
+    });
 }
 
 void insertCoordDocAndChangeOrigCollEntry(OperationContext* opCtx,
@@ -754,21 +748,16 @@ void writeStateTransitionAndCatalogUpdatesThenBumpShardVersions(
         ShardingCatalogClient::kLocalWriteConcern);
 }
 
-ReshardingCoordinatorDocument removeCoordinatorDocAndReshardingFields(
-    OperationContext* opCtx,
-    const ReshardingCoordinatorDocument& coordinatorDoc,
-    boost::optional<Status> abortReason) {
+void removeCoordinatorDocAndReshardingFields(OperationContext* opCtx,
+                                             const ReshardingCoordinatorDocument& coordinatorDoc,
+                                             boost::optional<Status> abortReason) {
     // If the coordinator needs to abort and isn't in kInitializing, additional collections need to
     // be cleaned up in the final transaction. Otherwise, cleanup for abort and success are the
     // same.
     const bool wasDecisionPersisted =
-        coordinatorDoc.getState() >= CoordinatorStateEnum::kCommitting;
+        coordinatorDoc.getState() == CoordinatorStateEnum::kCommitting;
     invariant((wasDecisionPersisted && !abortReason) || abortReason);
 
-    if (coordinatorDoc.getState() > CoordinatorStateEnum::kCommitting) {
-        return coordinatorDoc;
-    }
-
     ReshardingCoordinatorDocument updatedCoordinatorDoc = coordinatorDoc;
     updatedCoordinatorDoc.setState(CoordinatorStateEnum::kDone);
     emplaceTruncatedAbortReasonIfExists(updatedCoordinatorDoc, abortReason);
@@ -802,7 +791,6 @@ ReshardingCoordinatorDocument removeCoordinatorDocAndReshardingFields(
                 opCtx, updatedCoordinatorDoc, boost::none, boost::none, txnNumber);
         },
         ShardingCatalogClient::kLocalWriteConcern);
-    return updatedCoordinatorDoc;
 }
 }  // namespace resharding
 
@@ -855,12 +843,7 @@ ReshardingCoordinatorExternalStateImpl::calculateParticipantShardsAndChunks(
 
     // The database primary must always be a recipient to ensure it ends up with consistent
     // collection metadata.
-    const auto dbPrimaryShard =
-        uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabaseWithRefresh(
-                            opCtx, coordinatorDoc.getSourceNss().db()))
-            ->getPrimary();
-
-    recipientShardIds.emplace(dbPrimaryShard);
+    recipientShardIds.emplace(cm.dbPrimary());
 
     if (const auto& chunks = coordinatorDoc.getPresetReshardedChunks()) {
         auto version = calculateChunkVersionForInitialChunks(opCtx);
@@ -1089,7 +1072,6 @@ ReshardingCoordinatorService::ReshardingCoordinator::_tellAllParticipantsReshard
                        _cancelableOpCtxFactory.emplace(_ctHolder->getStepdownToken(),
                                                        _markKilledExecutor);
                    })
-                   .then([this] { return _waitForMajority(_ctHolder->getStepdownToken()); })
                    .then([this, executor]() {
                        pauseBeforeTellDonorToRefresh.pauseWhileSet();
                        _establishAllDonorsAsParticipants(executor);
@@ -1120,7 +1102,8 @@ ExecutorFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::_initi
     return resharding::WithAutomaticRetry([this, executor] {
                return ExecutorFuture<void>(**executor)
                    .then([this, executor] { _insertCoordDocAndChangeOrigCollEntry(); })
-                   .then([this, executor] { _calculateParticipantsAndChunksThenWriteToDisk(); });
+                   .then([this, executor] { _calculateParticipantsAndChunksThenWriteToDisk(); })
+                   .then([this] { return _waitForMajority(_ctHolder->getAbortToken()); });
            })
         .onTransientError([](const Status& status) {
             LOGV2(5093703,
@@ -1259,11 +1242,33 @@ ReshardingCoordinatorService::ReshardingCoordinator::_commitAndFinishReshardOper
     const ReshardingCoordinatorDocument& updatedCoordinatorDoc) noexcept {
     return resharding::WithAutomaticRetry([this, executor, updatedCoordinatorDoc] {
                return ExecutorFuture<void>(**executor)
-                   .then(
-                       [this, executor, updatedCoordinatorDoc] { _commit(updatedCoordinatorDoc); });
+                   .then([this, executor, updatedCoordinatorDoc] {
+                       return _commit(updatedCoordinatorDoc);
+                   })
+                   .then([this] { return _waitForMajority(_ctHolder->getStepdownToken()); })
+                   .thenRunOn(**executor)
+                   .then([this, executor] {
+                       _tellAllParticipantsToCommit(_coordinatorDoc.getSourceNss(), executor);
+                   })
+                   .then([this] { _updateChunkImbalanceMetrics(_coordinatorDoc.getSourceNss()); })
+                   .then([this, updatedCoordinatorDoc] {
+                       auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
+                       resharding::cleanupSourceConfigCollections(opCtx.get(),
+                                                                  updatedCoordinatorDoc);
+                       return Status::OK();
+                   })
+                   .then([this, executor] { return _awaitAllParticipantShardsDone(executor); })
+                   .then([this, executor] {
+                       // Best-effort attempt to trigger a refresh on the participant shards so
+                       // they see the collection metadata without reshardingFields and no longer
+                       // throw ReshardCollectionInProgress. There is no guarantee this logic ever
+                       // runs if the config server primary steps down after having removed the
+                       // coordinator state document.
+                       return _tellAllRecipientsToRefresh(executor);
+                   });
            })
         .onTransientError([](const Status& status) {
-            LOGV2(7698801,
+            LOGV2(5093705,
                   "Resharding coordinator encountered transient error while committing",
                   "error"_attr = status);
         })
@@ -1271,83 +1276,25 @@ ReshardingCoordinatorService::ReshardingCoordinator::_commitAndFinishReshardOper
         .until<Status>([](const Status& status) { return status.isOK(); })
         .on(**executor, _ctHolder->getStepdownToken())
         .onError([this, executor](Status status) {
-            if (status == ErrorCodes::TransactionTooLargeForCache) {
-                return _onAbortCoordinatorAndParticipants(executor, status);
+            {
+                auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
+                reshardingPauseCoordinatorBeforeStartingErrorFlow.pauseWhileSet(opCtx.get());
             }
-            return ExecutorFuture<void>(**executor, status);
-        })
-        .then([this, executor, updatedCoordinatorDoc] {
-            return resharding::WithAutomaticRetry([this, executor, updatedCoordinatorDoc] {
-                       return ExecutorFuture<void>(**executor)
-                           .then([this] { return _waitForMajority(_ctHolder->getStepdownToken()); })
-                           .thenRunOn(**executor)
-                           .then([this, executor] {
-                               _tellAllParticipantsToCommit(_coordinatorDoc.getSourceNss(),
-                                                            executor);
-                           })
-                           .then([this] {
-                               _updateChunkImbalanceMetrics(_coordinatorDoc.getSourceNss());
-                           })
-                           .then([this, updatedCoordinatorDoc] {
-                               auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
-                               resharding::removeChunkDocs(opCtx.get(),
-                                                           updatedCoordinatorDoc.getSourceUUID());
-                               return Status::OK();
-                           })
-                           .then([this, executor] {
-                               return _awaitAllParticipantShardsDone(executor);
-                           })
-                           .then([this, executor] {
-                               // Best-effort attempt to trigger a refresh on the participant shards
-                               // so they see the collection metadata without reshardingFields and
-                               // no longer throw ReshardCollectionInProgress. There is no guarantee
-                               // this logic ever runs if the config server primary steps down after
-                               // having removed the coordinator state document.
-                               return _tellAllRecipientsToRefresh(executor);
-                           });
-                   })
-                .onTransientError([](const Status& status) {
-                    LOGV2(5093705,
-                          "Resharding coordinator encountered transient error while committing",
-                          "error"_attr = status);
-                })
-                .onUnrecoverableError([](const Status& status) {})
-                .until<Status>([](const Status& status) { return status.isOK(); })
-                .on(**executor, _ctHolder->getStepdownToken())
-                .onError([this, executor](Status status) {
-                    {
-                        auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
-                        reshardingPauseCoordinatorBeforeStartingErrorFlow.pauseWhileSet(
-                            opCtx.get());
-                    }
 
-                    if (_ctHolder->isSteppingOrShuttingDown()) {
-                        return status;
-                    }
+            if (_ctHolder->isSteppingOrShuttingDown()) {
+                return status;
+            }
 
-                    LOGV2_FATAL(
-                        5277000,
+            LOGV2_FATAL(5277000,
                         "Unrecoverable error past the point resharding was guaranteed to succeed",
                         "error"_attr = redact(status));
-                });
         });
 }
 
 SemiFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::run(
     std::shared_ptr<executor::ScopedTaskExecutor> executor,
     const CancellationToken& stepdownToken) noexcept {
-    pauseBeforeCTHolderInitialization.pauseWhileSet();
-
-    auto abortCalled = [&] {
-        stdx::lock_guard<Latch> lk(_abortCalledMutex);
-        _ctHolder = std::make_unique<CoordinatorCancellationTokenHolder>(stepdownToken);
-        return _abortCalled;
-    }();
-
-    if (abortCalled) {
-        _ctHolder->abort();
-    }
-
+    _ctHolder = std::make_unique<CoordinatorCancellationTokenHolder>(stepdownToken);
     _markKilledExecutor->startup();
     _cancelableOpCtxFactory.emplace(_ctHolder->getAbortToken(), _markKilledExecutor);
 
@@ -1358,15 +1305,8 @@ SemiFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::run(
         })
         .onCompletion([this, executor](Status status) {
             auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
-            reshardingPauseCoordinatorBeforeCompletion.executeIf(
-                [&](const BSONObj&) {
-                    reshardingPauseCoordinatorBeforeCompletion.pauseWhileSetAndNotCanceled(
-                        opCtx.get(), _ctHolder->getStepdownToken());
-                },
-                [&](const BSONObj& data) {
-                    auto ns = data.getStringField("sourceNamespace");
-                    return ns.empty() ? true : ns.toString() == _coordinatorDoc.getSourceNss().ns();
-                });
+            reshardingPauseCoordinatorBeforeCompletion.pauseWhileSetAndNotCanceled(
+                opCtx.get(), _ctHolder->getStepdownToken());
 
             {
                 auto lg = stdx::lock_guard(_fulfillmentMutex);
@@ -1407,8 +1347,6 @@ SemiFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::run(
                     ->onStepDown(ReshardingMetrics::Role::kCoordinator);
             }
 
-            _logStatsOnCompletion(status.isOK());
-
             if (!status.isOK()) {
                 {
                     auto lg = stdx::lock_guard(_fulfillmentMutex);
@@ -1505,15 +1443,7 @@ ReshardingCoordinatorService::ReshardingCoordinator::_onAbortCoordinatorAndParti
 }
 
 void ReshardingCoordinatorService::ReshardingCoordinator::abort() {
-    auto ctHolderInitialized = [&] {
-        stdx::lock_guard<Latch> lk(_abortCalledMutex);
-        _abortCalled = true;
-        return !(_ctHolder == nullptr);
-    }();
-
-    if (ctHolderInitialized) {
-        _ctHolder->abort();
-    }
+    _ctHolder->abort();
 }
 
 boost::optional<BSONObj> ReshardingCoordinatorService::ReshardingCoordinator::reportForCurrentOp(
@@ -1612,12 +1542,10 @@ void ReshardingCoordinatorService::ReshardingCoordinator::
 
     // Remove the presetReshardedChunks and zones from the coordinator document to reduce
     // the possibility of the document reaching the BSONObj size constraint.
-    ShardKeyPattern shardKey(updatedCoordinatorDoc.getReshardingKey());
     std::vector<BSONObj> zones;
     if (updatedCoordinatorDoc.getZones()) {
         zones = buildTagsDocsFromZones(updatedCoordinatorDoc.getTempReshardingNss(),
-                                       *updatedCoordinatorDoc.getZones(),
-                                       shardKey);
+                                       *updatedCoordinatorDoc.getZones());
     }
     updatedCoordinatorDoc.setPresetReshardedChunks(boost::none);
     updatedCoordinatorDoc.setZones(boost::none);
@@ -1811,11 +1739,11 @@ ReshardingCoordinatorService::ReshardingCoordinator::_awaitAllRecipientsInStrict
         .thenRunOn(**executor);
 }
 
-void ReshardingCoordinatorService::ReshardingCoordinator::_commit(
+Future<void> ReshardingCoordinatorService::ReshardingCoordinator::_commit(
     const ReshardingCoordinatorDocument& coordinatorDoc) {
     if (_coordinatorDoc.getState() > CoordinatorStateEnum::kBlockingWrites) {
         invariant(_coordinatorDoc.getState() != CoordinatorStateEnum::kAborting);
-        return;
+        return Status::OK();
     }
 
     ReshardingCoordinatorDocument updatedCoordinatorDoc = coordinatorDoc;
@@ -1841,6 +1769,8 @@ void ReshardingCoordinatorService::ReshardingCoordinator::_commit(
 
     // Update the in memory state
     installCoordinatorDoc(opCtx.get(), updatedCoordinatorDoc);
+
+    return Status::OK();
 }
 
 ExecutorFuture<void>
@@ -2073,22 +2003,4 @@ void ReshardingCoordinatorService::ReshardingCoordinator::_updateChunkImbalanceM
     }
 }
 
-void ReshardingCoordinatorService::ReshardingCoordinator::_logStatsOnCompletion(bool success) {
-    BSONObjBuilder builder;
-    BSONObjBuilder statsBuilder;
-    builder.append("uuid", _coordinatorDoc.getReshardingUUID().toBSON());
-    builder.append("status", success ? "success" : "failed");
-    statsBuilder.append("ns", _coordinatorDoc.getSourceNss().toString());
-    statsBuilder.append("sourceUUID", _coordinatorDoc.getSourceUUID().toBSON());
-    statsBuilder.append("newUUID", _coordinatorDoc.getReshardingUUID().toBSON());
-    statsBuilder.append("newShardKey", _coordinatorDoc.getReshardingKey().toBSON());
-    if (_coordinatorDoc.getStartTime()) {
-        statsBuilder.append("startTime", *_coordinatorDoc.getStartTime());
-    }
-    statsBuilder.append("endTime", getCurrentTime());
-
-    builder.append("statistics", statsBuilder.obj());
-    LOGV2(7763800, "Resharding complete", "info"_attr = builder.obj());
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service.h b/src/mongo/db/s/resharding/resharding_coordinator_service.h
index 694070c126c..a24569ecc44 100644
--- a/src/mongo/db/s/resharding/resharding_coordinator_service.h
+++ b/src/mongo/db/s/resharding/resharding_coordinator_service.h
@@ -50,17 +50,14 @@ CollectionType createTempReshardingCollectionType(
     const ChunkVersion& chunkVersion,
     const BSONObj& collation);
 
-void removeChunkDocs(OperationContext* opCtx, const UUID& collUUID);
+void cleanupSourceConfigCollections(OperationContext* opCtx,
+                                    const ReshardingCoordinatorDocument& coordinatorDoc);
 
 void writeDecisionPersistedState(OperationContext* opCtx,
                                  const ReshardingCoordinatorDocument& coordinatorDoc,
                                  OID newCollectionEpoch,
                                  Timestamp newCollectionTimestamp);
 
-void updateTagsDocsForTempNss(OperationContext* opCtx,
-                              const ReshardingCoordinatorDocument& coordinatorDoc,
-                              TxnNumber txnNumber);
-
 void insertCoordDocAndChangeOrigCollEntry(OperationContext* opCtx,
                                           const ReshardingCoordinatorDocument& coordinatorDoc);
 
@@ -72,10 +69,9 @@ void writeParticipantShardsAndTempCollInfo(OperationContext* opCtx,
 void writeStateTransitionAndCatalogUpdatesThenBumpShardVersions(
     OperationContext* opCtx, const ReshardingCoordinatorDocument& coordinatorDoc);
 
-ReshardingCoordinatorDocument removeCoordinatorDocAndReshardingFields(
-    OperationContext* opCtx,
-    const ReshardingCoordinatorDocument& coordinatorDoc,
-    boost::optional<Status> abortReason = boost::none);
+void removeCoordinatorDocAndReshardingFields(OperationContext* opCtx,
+                                             const ReshardingCoordinatorDocument& coordinatorDoc,
+                                             boost::optional<Status> abortReason = boost::none);
 }  // namespace resharding
 
 class ReshardingCoordinatorExternalState {
@@ -408,10 +404,11 @@ private:
      * Does the following writes:
      * 1. Updates the config.collections entry for the new sharded collection
      * 2. Updates config.chunks entries for the new sharded collection
+     * 3. Updates config.tags for the new sharded collection
      *
      * Transitions to 'kCommitting'.
      */
-    void _commit(const ReshardingCoordinatorDocument& updatedDoc);
+    Future<void> _commit(const ReshardingCoordinatorDocument& updatedDoc);
 
     /**
      * Waits on _reshardingCoordinatorObserver to notify that:
@@ -501,11 +498,6 @@ private:
     // Waits for majority replication of the latest opTime unless token is cancelled.
     SemiFuture<void> _waitForMajority(const CancellationToken& token);
 
-    /**
-     * Print a log containing the information of this resharding operation.
-     */
-    void _logStatsOnCompletion(bool success);
-
     // The unique key for a given resharding operation. InstanceID is an alias for BSONObj. The
     // value of this is the UUID that will be used as the collection UUID for the new sharded
     // collection. The object looks like: {_id: 'reshardingUUID'}
@@ -542,13 +534,6 @@ private:
         MONGO_MAKE_LATCH("ReshardingCoordinatorService::_fulfillmentMutex");
 
     /**
-     * Must be locked while the _abortCalled is being set to true.
-     */
-    mutable Mutex _abortCalledMutex =
-        MONGO_MAKE_LATCH("ReshardingCoordinatorService::_abortCalledMutex");
-
-
-    /**
      * Coordinator does not enter the critical section until this is fulfilled.
      * Can be set by "commitReshardCollection" command or by metrics determining
      * that it's okay to proceed.
@@ -568,10 +553,6 @@ private:
     std::shared_ptr<resharding::CoordinatorCommitMonitor> _commitMonitor;
 
     std::shared_ptr<ReshardingCoordinatorExternalState> _reshardingCoordinatorExternalState;
-
-    // Used to catch the case when an abort() is called but the cancellation source (_ctHolder) has
-    // not been initialized.
-    bool _abortCalled{false};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp b/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp
index 33511df9fd4..6a5197b4c41 100644
--- a/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp
@@ -128,6 +128,10 @@ class ReshardingCoordinatorServiceTest : public ConfigServerTestFixture {
 public:
     using ReshardingCoordinator = ReshardingCoordinatorService::ReshardingCoordinator;
 
+    // TODO (SERVER-65302): Use wiredTiger.
+    ReshardingCoordinatorServiceTest()
+        : ConfigServerTestFixture(Options{}.engine("ephemeralForTest")) {}
+
     std::unique_ptr<repl::PrimaryOnlyService> makeService(ServiceContext* serviceContext) {
         return std::make_unique<ReshardingCoordinatorServiceForTest>(serviceContext);
     }
@@ -164,6 +168,7 @@ public:
             dynamic_cast<OpObserverRegistry*>(getServiceContext()->getOpObserver());
         invariant(_opObserverRegistry);
 
+        _opObserverRegistry->addObserver(std::make_unique<OpObserverImpl>());
         _opObserverRegistry->addObserver(std::make_unique<ReshardingOpObserver>());
         _opObserverRegistry->addObserver(
             std::make_unique<CoordinatorOpObserverForTest>(_controller));
@@ -213,9 +218,14 @@ public:
 
     std::shared_ptr<ReshardingCoordinatorService::ReshardingCoordinator> getCoordinatorIfExists(
         OperationContext* opCtx, repl::PrimaryOnlyService::InstanceID instanceId) {
-        auto [coordinatorOpt, _] = ReshardingCoordinatorService::ReshardingCoordinator::lookup(
+        auto coordinatorOpt = ReshardingCoordinatorService::ReshardingCoordinator::lookup(
             opCtx, _service, instanceId);
-        return coordinatorOpt ? *coordinatorOpt : nullptr;
+        if (!coordinatorOpt) {
+            return nullptr;
+        }
+
+        auto coordinator = *coordinatorOpt;
+        return coordinator ? coordinator : nullptr;
     }
 
     ReshardingCoordinatorDocument getCoordinatorDoc(OperationContext* opCtx) {
@@ -913,16 +923,5 @@ TEST_F(ReshardingCoordinatorServiceTest, ReshardingCoordinatorFailsIfMigrationNo
     }
 }
 
-TEST_F(ReshardingCoordinatorServiceTest, SuccessfullyAbortReshardOperationImmediately) {
-    auto pauseBeforeCTHolderInitialization =
-        globalFailPointRegistry().find("pauseBeforeCTHolderInitialization");
-    auto timesEnteredFailPoint = pauseBeforeCTHolderInitialization->setMode(FailPoint::alwaysOn, 0);
-    auto coordinator = initializeAndGetCoordinator();
-    coordinator->abort();
-    pauseBeforeCTHolderInitialization->waitForTimesEntered(timesEnteredFailPoint + 1);
-    pauseBeforeCTHolderInitialization->setMode(FailPoint::off, 0);
-    coordinator->getCompletionFuture().wait();
-}
-
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_coordinator_test.cpp b/src/mongo/db/s/resharding/resharding_coordinator_test.cpp
index 521cb84471e..a8fb4d83889 100644
--- a/src/mongo/db/s/resharding/resharding_coordinator_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_coordinator_test.cpp
@@ -584,8 +584,7 @@ protected:
         std::vector<BSONObj> zones;
         if (expectedCoordinatorDoc.getZones()) {
             zones = buildTagsDocsFromZones(expectedCoordinatorDoc.getTempReshardingNss(),
-                                           *expectedCoordinatorDoc.getZones(),
-                                           _newShardKey);
+                                           *expectedCoordinatorDoc.getZones());
         }
         expectedCoordinatorDoc.setZones(boost::none);
         expectedCoordinatorDoc.setPresetReshardedChunks(boost::none);
@@ -637,7 +636,7 @@ protected:
                                               ReshardingCoordinatorDocument expectedCoordinatorDoc,
                                               std::vector<ChunkType> expectedChunks,
                                               std::vector<TagsType> expectedZones) {
-        removeChunkDocs(opCtx, expectedCoordinatorDoc.getSourceUUID());
+        cleanupSourceConfigCollections(opCtx, expectedCoordinatorDoc);
         // Check that chunks and tags entries previously under the temporary namespace have been
         // correctly updated to the original namespace
 
@@ -648,13 +647,10 @@ protected:
 
     void removeCoordinatorDocAndReshardingFieldsExpectSuccess(
         OperationContext* opCtx, const ReshardingCoordinatorDocument& coordinatorDoc) {
-        auto updatedCoordinatorDoc = removeCoordinatorDocAndReshardingFields(opCtx, coordinatorDoc);
-        ASSERT_EQUALS(int(updatedCoordinatorDoc.getState()), int(CoordinatorStateEnum::kDone));
+        removeCoordinatorDocAndReshardingFields(opCtx, coordinatorDoc);
 
-        // Check that the on disk document is same as the in memory document returned above.
         auto expectedCoordinatorDoc = coordinatorDoc;
         expectedCoordinatorDoc.setState(CoordinatorStateEnum::kDone);
-        ASSERT_BSONOBJ_EQ(updatedCoordinatorDoc.toBSON(), expectedCoordinatorDoc.toBSON());
 
         // Check that the entry is removed from config.reshardingOperations
         DBDirectClient client(opCtx);
@@ -665,12 +661,12 @@ protected:
         // Check that the resharding fields are removed from the config.collections entry and
         // allowMigrations is set back to true.
         auto expectedOriginalCollType = makeOriginalCollectionCatalogEntry(
-            updatedCoordinatorDoc,
+            expectedCoordinatorDoc,
             boost::none,
             _finalEpoch,
             opCtx->getServiceContext()->getPreciseClockSource()->now());
         assertOriginalCollectionCatalogEntryMatchesExpected(
-            opCtx, expectedOriginalCollType, updatedCoordinatorDoc);
+            opCtx, expectedOriginalCollType, expectedCoordinatorDoc);
     }
 
     void transitionToErrorExpectSuccess(ErrorCodes::Error errorCode) {
diff --git a/src/mongo/db/s/resharding/resharding_data_copy_util.cpp b/src/mongo/db/s/resharding/resharding_data_copy_util.cpp
index 8635e389cf4..9893b2b0f2e 100644
--- a/src/mongo/db/s/resharding/resharding_data_copy_util.cpp
+++ b/src/mongo/db/s/resharding/resharding_data_copy_util.cpp
@@ -33,7 +33,7 @@
 
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/namespace_string.h"
@@ -44,7 +44,6 @@
 #include "mongo/db/s/resharding/resharding_txn_cloner_progress_gen.h"
 #include "mongo/db/s/resharding/resharding_util.h"
 #include "mongo/db/s/session_catalog_migration.h"
-#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/session_catalog_mongod.h"
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/db/storage/write_unit_of_work.h"
@@ -72,6 +71,28 @@ void ensureCollectionExists(OperationContext* opCtx,
     });
 }
 
+void ensureCollectionDropped(OperationContext* opCtx,
+                             const NamespaceString& nss,
+                             const boost::optional<UUID>& uuid) {
+    invariant(!opCtx->lockState()->isLocked());
+    invariant(!opCtx->lockState()->inAWriteUnitOfWork());
+
+    writeConflictRetry(
+        opCtx, "resharding::data_copy::ensureCollectionDropped", nss.toString(), [&] {
+            AutoGetCollection coll(opCtx, nss, MODE_X);
+            if (!coll || (uuid && coll->uuid() != uuid)) {
+                // If the collection doesn't exist or exists with a different UUID, then the
+                // requested collection has been dropped already.
+                return;
+            }
+
+            WriteUnitOfWork wuow(opCtx);
+            uassertStatusOK(coll.getDb()->dropCollectionEvenIfSystem(
+                opCtx, nss, {} /* dropOpTime */, true /* markFromMigrate */));
+            wuow.commit();
+        });
+}
+
 void ensureOplogCollectionsDropped(OperationContext* opCtx,
                                    const UUID& reshardingUUID,
                                    const UUID& sourceUUID,
@@ -98,11 +119,11 @@ void ensureOplogCollectionsDropped(OperationContext* opCtx,
 
         // Drop the conflict stash collection for this donor.
         auto stashNss = getLocalConflictStashNamespace(sourceUUID, donor.getShardId());
-        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, stashNss);
+        ensureCollectionDropped(opCtx, stashNss);
 
         // Drop the oplog buffer collection for this donor.
         auto oplogBufferNss = getLocalOplogBufferNamespace(sourceUUID, donor.getShardId());
-        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, oplogBufferNss);
+        ensureCollectionDropped(opCtx, oplogBufferNss);
     }
 }
 
@@ -282,8 +303,7 @@ void updateSessionRecord(OperationContext* opCtx,
                          BSONObj o2Field,
                          std::vector<StmtId> stmtIds,
                          boost::optional<repl::OpTime> preImageOpTime,
-                         boost::optional<repl::OpTime> postImageOpTime,
-                         NamespaceString sourceNss) {
+                         boost::optional<repl::OpTime> postImageOpTime) {
     invariant(opCtx->getLogicalSessionId());
     invariant(opCtx->getTxnNumber());
 
@@ -297,7 +317,7 @@ void updateSessionRecord(OperationContext* opCtx,
     oplogEntry.setOpType(repl::OpTypeEnum::kNoop);
     oplogEntry.setObject(SessionCatalogMigration::kSessionOplogTag);
     oplogEntry.setObject2(std::move(o2Field));
-    oplogEntry.setNss(std::move(sourceNss));
+    oplogEntry.setNss({});
     oplogEntry.setSessionId(sessionId);
     oplogEntry.setTxnNumber(txnNumber);
     oplogEntry.setStatementIds(stmtIds);
diff --git a/src/mongo/db/s/resharding/resharding_data_copy_util.h b/src/mongo/db/s/resharding/resharding_data_copy_util.h
index af24152aaf2..9f2a332ef6c 100644
--- a/src/mongo/db/s/resharding/resharding_data_copy_util.h
+++ b/src/mongo/db/s/resharding/resharding_data_copy_util.h
@@ -60,6 +60,16 @@ void ensureCollectionExists(OperationContext* opCtx,
                             const CollectionOptions& options);
 
 /**
+ * Drops the specified collection or returns without error if the collection has already been
+ * dropped. A particular incarnation of the collection can be dropped by specifying its UUID.
+ *
+ * This functions assumes the collection being dropped doesn't have any two-phase index builds
+ * active on it.
+ */
+void ensureCollectionDropped(OperationContext* opCtx,
+                             const NamespaceString& nss,
+                             const boost::optional<UUID>& uuid = boost::none);
+/**
  * Removes documents from the oplog applier progress and transaction applier progress collections
  * that are associated with an in-progress resharding operation. Also drops all oplog buffer
  * collections and conflict stash collections that are associated with the in-progress resharding
@@ -144,14 +154,13 @@ void updateSessionRecord(OperationContext* opCtx,
                          BSONObj o2Field,
                          std::vector<StmtId> stmtIds,
                          boost::optional<repl::OpTime> preImageOpTime,
-                         boost::optional<repl::OpTime> postImageOpTime,
-                         NamespaceString sourceNss);
+                         boost::optional<repl::OpTime> postImageOpTime);
 
 /**
  * Calls and returns the value from the supplied lambda function.
  *
- * If a StaleConfig error is thrown during its execution, then this function will attempt to refresh
- * the collection and invoke the supplied lambda function a second time.
+ * If a StaleConfig exception is thrown during its execution, then this function will attempt to
+ * refresh the collection and invoke the supplied lambda function a second time.
  */
 template <typename Callable>
 auto withOneStaleConfigRetry(OperationContext* opCtx, Callable&& callable) {
diff --git a/src/mongo/db/s/resharding/resharding_data_replication.cpp b/src/mongo/db/s/resharding/resharding_data_replication.cpp
index 5223b1c5c0b..a5918e3ea77 100644
--- a/src/mongo/db/s/resharding/resharding_data_replication.cpp
+++ b/src/mongo/db/s/resharding/resharding_data_replication.cpp
@@ -475,9 +475,4 @@ ReshardingDonorOplogId ReshardingDataReplication::getOplogApplierResumeId(
                            : ReshardingDonorOplogId{minFetchTimestamp, minFetchTimestamp};
 }
 
-ReshardingDataReplication::~ReshardingDataReplication() {
-    shutdown();
-    join();
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_data_replication.h b/src/mongo/db/s/resharding/resharding_data_replication.h
index 61355447a40..a2705abd6f6 100644
--- a/src/mongo/db/s/resharding/resharding_data_replication.h
+++ b/src/mongo/db/s/resharding/resharding_data_replication.h
@@ -136,7 +136,6 @@ private:
     struct TrustedInitTag {};
 
 public:
-    virtual ~ReshardingDataReplication();
     static std::unique_ptr<ReshardingDataReplicationInterface> make(
         OperationContext* opCtx,
         ReshardingMetrics* metrics,
diff --git a/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp b/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp
index 0c26b0acd02..e942dcd139f 100644
--- a/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp
@@ -131,16 +131,14 @@ public:
         StaticCatalogClient(std::vector<ShardType> shards) : _shards(std::move(shards)) {}
 
         StatusWith<repl::OpTimeWith<std::vector<ShardType>>> getAllShards(
-            OperationContext* opCtx,
-            repl::ReadConcernLevel readConcern,
-            bool excludeDraining) override {
+            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 {
+        std::vector<CollectionType> getCollections(
+            OperationContext* opCtx,
+            StringData dbName,
+            repl::ReadConcernLevel readConcernLevel) override {
             return _colls;
         }
 
@@ -239,7 +237,7 @@ protected:
             createChunks(env.version.epoch(), env.sourceUuid, env.version.getTimestamp(), "y"),
             boost::none);
 
-        ASSERT_OK(onDbVersionMismatchNoExcept(opCtx, kNss.db(), boost::none));
+        forceDatabaseRefresh(opCtx, kNss.db());
         forceShardFilteringMetadataRefresh(opCtx, kNss);
 
         if (refreshTempNss)
@@ -312,18 +310,22 @@ TEST_F(DestinedRecipientTest, TestGetDestinedRecipientThrowsOnBlockedRefresh) {
     auto opCtx = operationContext();
     auto env = setupReshardingEnv(opCtx, false);
 
-    AutoGetCollection coll(opCtx, kNss, MODE_IX);
-    OperationShardingState::setShardRole(opCtx, kNss, env.version, env.dbVersion);
+    {
+        AutoGetCollection coll(opCtx, kNss, MODE_IX);
+        OperationShardingState::setShardRole(opCtx, kNss, env.version, env.dbVersion);
+
+        FailPointEnableBlock failPoint("blockCollectionCacheLookup");
+        ASSERT_THROWS_WITH_CHECK(ShardingWriteRouter(opCtx, kNss, Grid::get(opCtx)->catalogCache()),
+                                 ShardCannotRefreshDueToLocksHeldException,
+                                 [&](const ShardCannotRefreshDueToLocksHeldException& ex) {
+                                     const auto refreshInfo =
+                                         ex.extraInfo<ShardCannotRefreshDueToLocksHeldInfo>();
+                                     ASSERT(refreshInfo);
+                                     ASSERT_EQ(refreshInfo->getNss(), env.tempNss);
+                                 });
+    }
 
-    FailPointEnableBlock failPoint("blockCollectionCacheLookup");
-    ASSERT_THROWS_WITH_CHECK(ShardingWriteRouter(opCtx, kNss, Grid::get(opCtx)->catalogCache()),
-                             ShardCannotRefreshDueToLocksHeldException,
-                             [&](const ShardCannotRefreshDueToLocksHeldException& ex) {
-                                 const auto refreshInfo =
-                                     ex.extraInfo<ShardCannotRefreshDueToLocksHeldInfo>();
-                                 ASSERT(refreshInfo);
-                                 ASSERT_EQ(refreshInfo->getNss(), env.tempNss);
-                             });
+    auto sw = catalogCache()->getCollectionRoutingInfoWithRefresh(opCtx, env.tempNss);
 }
 
 TEST_F(DestinedRecipientTest, TestOpObserverSetsDestinedRecipientOnInserts) {
diff --git a/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp b/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp
index 99ab47fc11a..e9a466a1cb6 100644
--- a/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp
+++ b/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp
@@ -51,35 +51,10 @@ using DonorStateMachine = ReshardingDonorService::DonorStateMachine;
 using RecipientStateMachine = ReshardingRecipientService::RecipientStateMachine;
 
 namespace {
-MONGO_FAIL_POINT_DEFINE(reshardingInterruptAfterInsertStateMachineDocument);
-
 using namespace fmt::literals;
 
 const Backoff kExponentialBackoff(Seconds(1), Milliseconds::max());
 
-template <class StateMachine, class ReshardingDocument>
-void ensureStateDocumentInserted(OperationContext* opCtx, const ReshardingDocument& doc) {
-    try {
-        StateMachine::insertStateDocument(opCtx, doc);
-    } catch (const ExceptionFor<ErrorCodes::DuplicateKey>& ex) {
-        // It's possible that the state document was already previously inserted in the following
-        // cases:
-        // 1. The document was inserted previously, but the opCtx was interrupted before the
-        // state machine was started in-memory with getOrCreate(), e.g. due to a chunk migration
-        // (see SERVER-74647)
-        // 2. Similar to the ErrorCategory::NotPrimaryError clause below, it is
-        // theoretically possible for a series of stepdowns and step-ups to lead a scenario where a
-        // stale but now re-elected primary attempts to insert the state document when another node
-        // which was primary had already done so. Again, rather than attempt to prevent replica set
-        // member state transitions during the shard version refresh, we instead swallow the
-        // DuplicateKey exception. This is safe because PrimaryOnlyService::onStepUp() will have
-        // constructed a new instance of the resharding state machine.
-        auto dupeKeyInfo = ex.extraInfo<DuplicateKeyErrorInfo>();
-        invariant(dupeKeyInfo->getDuplicatedKeyValue().binaryEqual(
-            BSON("_id" << doc.getReshardingUUID())));
-    }
-}
-
 /*
  * Creates a ReshardingStateMachine if this node is primary and the ReshardingStateMachine doesn't
  * already exist.
@@ -92,10 +67,7 @@ void createReshardingStateMachine(OperationContext* opCtx, const ReshardingDocum
         // Inserting the resharding state document must happen synchronously with the shard version
         // refresh for the w:majority wait from the resharding coordinator to mean that this replica
         // set shard cannot forget about being a participant.
-        ensureStateDocumentInserted<StateMachine>(opCtx, doc);
-
-        reshardingInterruptAfterInsertStateMachineDocument.execute(
-            [&opCtx](const BSONObj& data) { opCtx->markKilled(); });
+        StateMachine::insertStateDocument(opCtx, doc);
 
         auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext());
         auto service = registry->lookupServiceByName(Service::kServiceName);
@@ -110,6 +82,17 @@ void createReshardingStateMachine(OperationContext* opCtx, const ReshardingDocum
         // secondary (or primary which stepped down) must do for an active resharding operation upon
         // refreshing its shard version. The primary is solely responsible for advancing the
         // participant state as a result of the shard version refresh.
+    } catch (const ExceptionFor<ErrorCodes::DuplicateKey>& ex) {
+        // Similar to the ErrorCategory::NotPrimaryError clause above, it is theoretically possible
+        // for a series of stepdowns and step-ups to lead a scenario where a stale but now
+        // re-elected primary attempts to insert the state document when another node which was
+        // primary had already done so. Again, rather than attempt to prevent replica set member
+        // state transitions during the shard version refresh, we instead swallow the DuplicateKey
+        // exception. This is safe because PrimaryOnlyService::onStepUp() will have constructed a
+        // new instance of the resharding state machine.
+        auto dupeKeyInfo = ex.extraInfo<DuplicateKeyErrorInfo>();
+        invariant(dupeKeyInfo->getDuplicatedKeyValue().binaryEqual(
+            BSON("_id" << doc.getReshardingUUID())));
     }
 }
 
@@ -155,13 +138,6 @@ void processReshardingFieldsForDonorCollection(OperationContext* opCtx,
         return;
     }
 
-    // We clear the routing information for the temporary resharding namespace to ensure this donor
-    // shard primary will refresh from the config server and see the chunk distribution for the new
-    // resharding operation.
-    auto* catalogCache = Grid::get(opCtx)->catalogCache();
-    catalogCache->invalidateCollectionEntry_LINEARIZABLE(
-        reshardingFields.getDonorFields()->getTempReshardingNss());
-
     auto donorDoc = constructDonorDocumentFromReshardingFields(nss, metadata, reshardingFields);
     createReshardingStateMachine<ReshardingDonorService,
                                  DonorStateMachine,
@@ -352,16 +328,7 @@ void clearFilteringMetadata(OperationContext* opCtx, bool scheduleAsyncRefresh)
         });
     }
 
-    auto* catalogCache = Grid::get(opCtx)->catalogCache();
-
     for (const auto& nss : namespacesToRefresh) {
-        if (nss.isTemporaryReshardingCollection()) {
-            // We clear the routing information for the temporary resharding namespace to ensure all
-            // new donor shard primaries will refresh from the config server and see the chunk
-            // distribution for the ongoing resharding operation.
-            catalogCache->invalidateCollectionEntry_LINEARIZABLE(nss);
-        }
-
         AutoGetCollection autoColl(opCtx, nss, MODE_IX);
         CollectionShardingRuntime::get(opCtx, nss)->clearFilteringMetadata(opCtx);
 
diff --git a/src/mongo/db/s/resharding/resharding_donor_recipient_common.h b/src/mongo/db/s/resharding/resharding_donor_recipient_common.h
index f6d6a259282..2efba26f659 100644
--- a/src/mongo/db/s/resharding/resharding_donor_recipient_common.h
+++ b/src/mongo/db/s/resharding/resharding_donor_recipient_common.h
@@ -38,15 +38,11 @@ namespace resharding {
 using ReshardingFields = TypeCollectionReshardingFields;
 
 /**
- * Looks up the StateMachine by the 'reshardingUUID'. Returns boost::none in the following cases:
- * 1. The state machine does not exist.
- * 2. In certain cases when the node is shutting down.
- * Additionally returns a bool indicating if the node is stepping or shutting down to disambiguate
- * the two.
+ * Looks up the StateMachine by the 'reshardingUUID'. If it does not exist, returns boost::none.
  */
 template <class Service, class StateMachine, class ReshardingDocument>
-std::pair<boost::optional<std::shared_ptr<StateMachine>>, bool>
-tryGetReshardingStateMachineAndShutdownState(OperationContext* opCtx, const UUID& reshardingUUID) {
+boost::optional<std::shared_ptr<StateMachine>> tryGetReshardingStateMachine(
+    OperationContext* opCtx, const UUID& reshardingUUID) {
     auto instanceId = BSON(ReshardingDocument::kReshardingUUIDFieldName << reshardingUUID);
     auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext());
     auto service = registry->lookupServiceByName(Service::kServiceName);
@@ -54,39 +50,6 @@ tryGetReshardingStateMachineAndShutdownState(OperationContext* opCtx, const UUID
 }
 
 /**
- * Same as tryGetReshardingStateMachineAndShutdownState, except does not return the shutdown state.
- */
-template <class Service, class StateMachine, class ReshardingDocument>
-boost::optional<std::shared_ptr<StateMachine>> tryGetReshardingStateMachine(
-    OperationContext* opCtx, const UUID& reshardingUUID) {
-    auto [instance, _] =
-        tryGetReshardingStateMachineAndShutdownState<Service, StateMachine, ReshardingDocument>(
-            opCtx, reshardingUUID);
-    return instance;
-}
-
-/**
- * Same as tryGetReshardingStateMachine, except throws if we were stepping or shutting down when we
- * tried to access the PrimaryOnlyService. Use this function in situations where you need to
- * guarantee that a return of boost::none means that there is no state document on disk for the
- * associated state machine.
- */
-template <class Service, class StateMachine, class ReshardingDocument>
-boost::optional<std::shared_ptr<StateMachine>> tryGetReshardingStateMachineAndThrowIfShuttingDown(
-    OperationContext* opCtx, const UUID& reshardingUUID) {
-    auto [instance, steppingOrShuttingDown] =
-        tryGetReshardingStateMachineAndShutdownState<Service, StateMachine, ReshardingDocument>(
-            opCtx, reshardingUUID);
-
-    uassert(ErrorCodes::InterruptedDueToReplStateChange,
-            "Unable to get resharding state machine, if it exists, because the node is "
-            "stepping or shutting down.",
-            !steppingOrShuttingDown);
-
-    return instance;
-}
-
-/**
  * The following functions construct a ReshardingDocument from the given 'reshardingFields'.
  */
 ReshardingDonorDocument constructDonorDocumentFromReshardingFields(
diff --git a/src/mongo/db/s/resharding/resharding_donor_service.cpp b/src/mongo/db/s/resharding/resharding_donor_service.cpp
index 2e75751ba01..ce60e58f0e1 100644
--- a/src/mongo/db/s/resharding/resharding_donor_service.cpp
+++ b/src/mongo/db/s/resharding/resharding_donor_service.cpp
@@ -37,7 +37,7 @@
 #include "mongo/db/catalog/drop_collection.h"
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/dbhelpers.h"
@@ -47,7 +47,6 @@
 #include "mongo/db/persistent_task_store.h"
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/repl/wait_for_majority_service.h"
-#include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/recoverable_critical_section_service.h"
 #include "mongo/db/s/resharding/resharding_change_event_o2_field_gen.h"
 #include "mongo/db/s/resharding/resharding_data_copy_util.h"
@@ -56,7 +55,6 @@
 #include "mongo/db/s/resharding/resharding_metrics.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
 #include "mongo/db/s/resharding/resharding_util.h"
-#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/write_block_bypass.h"
 #include "mongo/db/write_concern_options.h"
@@ -67,7 +65,6 @@
 
 namespace mongo {
 
-MONGO_FAIL_POINT_DEFINE(reshardingPauseDonorBeforeCatalogCacheRefresh);
 MONGO_FAIL_POINT_DEFINE(reshardingDonorFailsAfterTransitionToDonatingOplogEntries);
 MONGO_FAIL_POINT_DEFINE(removeDonorDocFailpoint);
 
@@ -180,9 +177,8 @@ public:
         }
     }
 
-    void refreshCollectionPlacementInfo(OperationContext* opCtx,
-                                        const NamespaceString& sourceNss) override {
-        onShardVersionMismatch(opCtx, sourceNss, boost::none);
+    void clearFilteringMetadata(OperationContext* opCtx) {
+        resharding::clearFilteringMetadata(opCtx, true /* scheduleAsyncRefresh */);
     }
 };
 
@@ -371,15 +367,8 @@ ExecutorFuture<void> ReshardingDonorService::DonorStateMachine::_finishReshardin
 
                {
                    auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
-                   std::initializer_list<NamespaceString> namespacesToRefresh{
-                       _metadata.getSourceNss(), _metadata.getTempReshardingNss()};
-
-                   // Clear filtering metadata for the source and temp resharding nss.
-                   for (const auto& nss : namespacesToRefresh) {
-                       AutoGetCollection autoColl(opCtx.get(), nss, MODE_IX);
-                       CollectionShardingRuntime::get(opCtx.get(), nss)
-                           ->clearFilteringMetadata(opCtx.get());
-                   }
+
+                   _externalState->clearFilteringMetadata(opCtx.get());
 
                    RecoverableCriticalSectionService::get(opCtx.get())
                        ->releaseRecoverableCriticalSection(
@@ -389,13 +378,6 @@ ExecutorFuture<void> ReshardingDonorService::DonorStateMachine::_finishReshardin
                            ShardingCatalogClient::kLocalWriteConcern);
 
                    _metrics()->leaveCriticalSection(getCurrentTime());
-
-                   // We force a refresh to make sure that the placement information is updated in
-                   // cache after abort decision before the donor state document is deleted.
-                   for (const auto& nss : namespacesToRefresh) {
-                       _externalState->refreshCollectionPlacementInfo(opCtx.get(), nss);
-                       _externalState->waitForCollectionFlush(opCtx.get(), nss);
-                   }
                }
 
                auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
@@ -585,8 +567,6 @@ void ReshardingDonorService::DonorStateMachine::
     // with a SnapshotUnavailable error response.
     {
         auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc());
-        reshardingPauseDonorBeforeCatalogCacheRefresh.pauseWhileSet(opCtx.get());
-
         _externalState->refreshCatalogCache(opCtx.get(), _metadata.getTempReshardingNss());
         _externalState->waitForCollectionFlush(opCtx.get(), _metadata.getTempReshardingNss());
     }
@@ -815,7 +795,7 @@ void ReshardingDonorService::DonorStateMachine::_dropOriginalCollectionThenTrans
         // Allow bypassing user write blocking. The check has already been performed on the
         // db-primary shard's ReshardCollectionCoordinator.
         WriteBlockBypass::get(opCtx.get()).set(true);
-        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(
+        resharding::data_copy::ensureCollectionDropped(
             opCtx.get(), _metadata.getSourceNss(), _metadata.getSourceUUID());
     }
 
diff --git a/src/mongo/db/s/resharding/resharding_donor_service.h b/src/mongo/db/s/resharding/resharding_donor_service.h
index 7b5331f93ac..b50c88b6af5 100644
--- a/src/mongo/db/s/resharding/resharding_donor_service.h
+++ b/src/mongo/db/s/resharding/resharding_donor_service.h
@@ -298,8 +298,7 @@ public:
                                            const BSONObj& query,
                                            const BSONObj& update) = 0;
 
-    virtual void refreshCollectionPlacementInfo(OperationContext* opCtx,
-                                                const NamespaceString& sourceNss) = 0;
+    virtual void clearFilteringMetadata(OperationContext* opCtx) = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp
index 16e1507be1c..663b8c28136 100644
--- a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp
@@ -34,6 +34,7 @@
 #include <boost/optional/optional_io.hpp>
 #include <utility>
 
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/op_observer_noop.h"
 #include "mongo/db/op_observer_registry.h"
@@ -52,7 +53,6 @@
 #include "mongo/db/s/resharding/resharding_donor_service.h"
 #include "mongo/db/s/resharding/resharding_service_test_helpers.h"
 #include "mongo/db/s/resharding/resharding_util.h"
-#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/unittest/death_test.h"
@@ -84,8 +84,7 @@ public:
                                    const BSONObj& query,
                                    const BSONObj& update) override {}
 
-    void refreshCollectionPlacementInfo(OperationContext* opCtx,
-                                        const NamespaceString& sourceNss) override {}
+    void clearFilteringMetadata(OperationContext* opCtx) override {}
 };
 
 class DonorOpObserverForTest : public OpObserverForTest {
@@ -162,8 +161,7 @@ public:
     void createSourceCollection(OperationContext* opCtx, const ReshardingDonorDocument& donorDoc) {
         CollectionOptions options;
         options.uuid = donorDoc.getSourceUUID();
-        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx,
-                                                                       donorDoc.getSourceNss());
+        resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getSourceNss());
         resharding::data_copy::ensureCollectionExists(opCtx, donorDoc.getSourceNss(), options);
     }
 
@@ -171,8 +169,7 @@ public:
                                              const ReshardingDonorDocument& donorDoc) {
         CollectionOptions options;
         options.uuid = donorDoc.getReshardingUUID();
-        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(
-            opCtx, donorDoc.getTempReshardingNss());
+        resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getTempReshardingNss());
         resharding::data_copy::ensureCollectionExists(
             opCtx, donorDoc.getTempReshardingNss(), options);
     }
@@ -418,10 +415,8 @@ TEST_F(ReshardingDonorServiceTest, StepDownStepUpEachTransition) {
                     DonorStateMachine::insertStateDocument(opCtx.get(), doc);
                     return DonorStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON());
                 } else {
-                    auto [maybeDonor, isPausedOrShutdown] =
-                        DonorStateMachine::lookup(opCtx.get(), _service, instanceId);
-                    ASSERT_TRUE(maybeDonor);
-                    ASSERT_FALSE(isPausedOrShutdown);
+                    auto maybeDonor = DonorStateMachine::lookup(opCtx.get(), _service, instanceId);
+                    ASSERT_TRUE(bool(maybeDonor));
 
                     // Allow the transition to prevState to succeed on this primary-only service
                     // instance.
@@ -482,10 +477,8 @@ TEST_F(ReshardingDonorServiceTest, StepDownStepUpEachTransition) {
         }
 
         // Finally complete the operation and ensure its success.
-        auto [maybeDonor, isPausedOrShutdown] =
-            DonorStateMachine::lookup(opCtx.get(), _service, instanceId);
-        ASSERT_TRUE(maybeDonor);
-        ASSERT_FALSE(isPausedOrShutdown);
+        auto maybeDonor = DonorStateMachine::lookup(opCtx.get(), _service, instanceId);
+        ASSERT_TRUE(bool(maybeDonor));
 
         auto donor = *maybeDonor;
         stateTransitionsGuard.unset(DonorStateEnum::kDone);
@@ -615,10 +608,8 @@ TEST_F(ReshardingDonorServiceTest, CompletesWithStepdownAfterAbort) {
         donor.reset();
         stepUp(opCtx.get());
 
-        auto [maybeDonor, isPausedOrShutdown] =
-            DonorStateMachine::lookup(opCtx.get(), _service, instanceId);
-        ASSERT_TRUE(maybeDonor);
-        ASSERT_FALSE(isPausedOrShutdown);
+        auto maybeDonor = DonorStateMachine::lookup(opCtx.get(), _service, instanceId);
+        ASSERT_TRUE(bool(maybeDonor));
 
         donor = *maybeDonor;
         doneTransitionGuard.reset();
diff --git a/src/mongo/db/s/resharding/resharding_future_util.cpp b/src/mongo/db/s/resharding/resharding_future_util.cpp
index 62c95fbaf3f..849a14bd80e 100644
--- a/src/mongo/db/s/resharding/resharding_future_util.cpp
+++ b/src/mongo/db/s/resharding/resharding_future_util.cpp
@@ -51,42 +51,19 @@ ExecutorFuture<void> whenAllSucceedOn(const std::vector<SharedSemiFuture<void>>&
                             : ExecutorFuture(executor);
 }
 
-std::vector<Future<void>> runAllInlineUnsafe(const std::vector<SharedSemiFuture<void>>& futures) {
-    std::vector<Future<void>> result;
-    result.reserve(futures.size());
-
-    for (const auto& future : futures) {
-        result.emplace_back(future.unsafeToInlineFuture());
-    }
-
-    return result;
-}
-
 ExecutorFuture<void> cancelWhenAnyErrorThenQuiesce(
     const std::vector<SharedSemiFuture<void>>& futures,
     ExecutorPtr executor,
     CancellationSource cancelSource) {
-    if (futures.empty()) {
-        return ExecutorFuture(executor);
-    }
-    // Run all futures inline so that the onError callback is called even if that error was caused
-    // by the executor shutting down. This causes the logic for whenAllSucceed, whenAll, and the
-    // onError callback to potentially run on the threads of the setters of the promises
-    // associated with the input futures. Since this logic is thread safe, not blocking, and does
-    // not acquire additional resources, this is safe, but beware if making further changes to this
-    // function.
-    return whenAllSucceed(runAllInlineUnsafe(futures))
-        .unsafeToInlineFuture()
-        .onError([futures, cancelSource](Status originalError) mutable {
+    return whenAllSucceedOn(futures, executor)
+        .onError([futures, executor, cancelSource](Status originalError) mutable {
             cancelSource.cancel();
 
-            return whenAll(runAllInlineUnsafe(futures))
+            return whenAll(thenRunAllOn(futures, executor))
                 .ignoreValue()
-                .unsafeToInlineFuture()
+                .thenRunOn(executor)
                 .onCompletion([originalError](auto) { return originalError; });
-        })
-        .thenRunOn(executor);
+        });
 }
 
-
 }  // namespace mongo::resharding
diff --git a/src/mongo/db/s/resharding/resharding_future_util.h b/src/mongo/db/s/resharding/resharding_future_util.h
index 78847350d0c..0bb858a963c 100644
--- a/src/mongo/db/s/resharding/resharding_future_util.h
+++ b/src/mongo/db/s/resharding/resharding_future_util.h
@@ -131,8 +131,7 @@ public:
                     status.isA<ErrorCategory::CursorInvalidatedError>() ||
                     status == ErrorCodes::Interrupted ||
                     status.isA<ErrorCategory::CancellationError>() ||
-                    status.isA<ErrorCategory::NotPrimaryError>() ||
-                    status.isA<ErrorCategory::NetworkTimeoutError>()) {
+                    status.isA<ErrorCategory::NotPrimaryError>()) {
                     // Always attempt to retry on any type of retryable error. Also retry on errors
                     // from stray killCursors and killOp commands being run. Cancellation and
                     // NotPrimary errors may indicate the primary-only service Instance will be shut
diff --git a/src/mongo/db/s/resharding/resharding_future_util_test.cpp b/src/mongo/db/s/resharding/resharding_future_util_test.cpp
deleted file mode 100644
index e37a13a314b..00000000000
--- a/src/mongo/db/s/resharding/resharding_future_util_test.cpp
+++ /dev/null
@@ -1,100 +0,0 @@
-/**
- *    Copyright (C) 2022-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the Server Side Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#include "mongo/db/s/resharding/resharding_future_util.h"
-#include "mongo/unittest/unittest.h"
-#include "mongo/util/concurrency/thread_pool.h"
-
-namespace mongo {
-namespace {
-class ReshardingFutureUtilTest : public unittest::Test {
-protected:
-    void setUp() override {
-        _executor = std::make_shared<ThreadPool>([]() {
-            ThreadPool::Options options;
-            options.maxThreads = 2;
-            return options;
-        }());
-        _executor->startup();
-    }
-
-    void tearDown() override {
-        _executor->shutdown();
-        _executor->join();
-    }
-
-    std::shared_ptr<ThreadPool> getExecutor() const {
-        return _executor;
-    }
-
-private:
-    std::shared_ptr<ThreadPool> _executor;
-};
-
-TEST_F(ReshardingFutureUtilTest, CancelWhenAnyErrorThenQuiesceDuringExecutorShutdown) {
-    CancellationSource cancelSource;
-    auto token = cancelSource.token();
-    PromiseAndFuture<void> taskThreadsReady;
-    AtomicWord<int> tasksRunningCount{0};
-    AtomicWord<bool> taskWasCancelled{false};
-    auto checkSignalReady = [&]() {
-        auto running = tasksRunningCount.addAndFetch(1);
-        if (running == 2) {
-            taskThreadsReady.promise.emplaceValue();
-        }
-    };
-    PromiseAndFuture<void> executorShutDownTriggered;
-    auto quiesced = ExecutorFuture(getExecutor()).then([&]() {
-        return resharding::cancelWhenAnyErrorThenQuiesce(
-            {ExecutorFuture(getExecutor())
-                 .then([&]() {
-                     checkSignalReady();
-                     executorShutDownTriggered.future.wait();
-                     uasserted(6791600, "Executor shut down");
-                 })
-                 .share(),
-             ExecutorFuture(getExecutor())
-                 .then([&]() {
-                     checkSignalReady();
-                     token.onCancel().wait();
-                     taskWasCancelled.store(true);
-                 })
-                 .share()},
-            getExecutor(),
-            cancelSource);
-    });
-    taskThreadsReady.future.wait();
-    getExecutor()->shutdown();
-    executorShutDownTriggered.promise.emplaceValue();
-    auto status = quiesced.getNoThrow();
-    ASSERT_EQ(status.code(), 6791600);
-    ASSERT_TRUE(taskWasCancelled.load());
-}
-}  // namespace
-}  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_metrics.cpp b/src/mongo/db/s/resharding/resharding_metrics.cpp
index b42c9df5c9e..6e4e4e041e9 100644
--- a/src/mongo/db/s/resharding/resharding_metrics.cpp
+++ b/src/mongo/db/s/resharding/resharding_metrics.cpp
@@ -406,60 +406,6 @@ void ReshardingMetrics::onStepUp(Role role) noexcept {
     // instead of starting from the current time.
 }
 
-void ReshardingMetrics::onStepUp(RecipientStateEnum state,
-                                 const ReshardingRecipientCountsAndMetrics& recipientMetrics) {
-    stdx::lock_guard<Latch> lk(_mutex);
-
-    _emplaceCurrentOpForRole(Role::kRecipient, boost::none);
-    _onStepUpCalled = true;
-
-    invariant(_currentOp, kNoOperationInProgress);
-    invariant(_currentOp->documentsCopied == 0, kMetricsSetBeforeRestore);
-    invariant(_currentOp->bytesCopied == 0, kMetricsSetBeforeRestore);
-    invariant(_currentOp->oplogEntriesFetched == 0, kMetricsSetBeforeRestore);
-    invariant(_currentOp->oplogEntriesApplied == 0, kMetricsSetBeforeRestore);
-
-    _currentOp->recipientState = state;
-    _currentOp->documentsCopied = recipientMetrics.documentCountCopied;
-    _currentOp->bytesCopied = recipientMetrics.documentBytesCopied;
-    _currentOp->oplogEntriesFetched = recipientMetrics.oplogEntriesFetched;
-    _currentOp->oplogEntriesApplied = recipientMetrics.oplogEntriesApplied;
-
-    if (recipientMetrics.approxBytesToCopy)
-        _currentOp->bytesToCopy = recipientMetrics.approxBytesToCopy.get();
-
-
-    const auto& timeIntervals = recipientMetrics.metrics;
-
-    // Restore in memory state of document copy metrics.
-    // Not calling startCopyingDocuments or endCopyingDocuments because they acquire a mutex that we
-    // already have.
-    //
-    // Also, note that it is possible for documentCopyInterval->getStart() to be none and for
-    // documentCopyInterval->getStop() to be not none.  That can happen if the cluster is upgraded
-    // to include code for persisting time intervals during a resharding operation.
-    // In that case, restore neither the start nor stop time. The resharding coordinator will still
-    // treat this scenario as the recipient shard being completely caught up after a primary
-    // failover and engage the critical section too early.
-    const auto& documentCopyInterval = timeIntervals.getDocumentCopy();
-    if (documentCopyInterval && documentCopyInterval->getStart()) {
-        _currentOp->copyingDocuments.start(documentCopyInterval->getStart().get());
-        if (documentCopyInterval->getStop()) {
-            _currentOp->copyingDocuments.end(documentCopyInterval->getStop().get());
-        }
-    }
-    // Restore in memory state of oplog application metrics.
-    // Not calling startApplyingOplogEntries or endApplyingOplogEntries because they acquire a mutex
-    // that we already have.
-    const auto& oplogApplicationInterval = timeIntervals.getOplogApplication();
-    if (oplogApplicationInterval && oplogApplicationInterval->getStart()) {
-        _currentOp->applyingOplogEntries.start(oplogApplicationInterval->getStart().get());
-        if (oplogApplicationInterval->getStop()) {
-            _currentOp->applyingOplogEntries.end(oplogApplicationInterval->getStop().get());
-        }
-    }
-}
-
 void ReshardingMetrics::onStepUp(DonorStateEnum state, ReshardingDonorMetrics donorMetrics) {
     stdx::lock_guard<Latch> lk(_mutex);
     auto operationRuntime = donorMetrics.getOperationRuntime();
@@ -538,9 +484,6 @@ void ReshardingMetrics::setDonorState(DonorStateEnum state) noexcept {
 
 void ReshardingMetrics::setRecipientState(RecipientStateEnum state) noexcept {
     stdx::lock_guard<Latch> lk(_mutex);
-    if (!_currentOp && state == RecipientStateEnum::kDone) {
-        return;
-    }
     invariant(_currentOp, kNoOperationInProgress);
 
     const auto oldState = std::exchange(_currentOp->recipientState, state);
@@ -691,10 +634,6 @@ void ReshardingMetrics::enterCriticalSection(Date_t start) {
 
 void ReshardingMetrics::leaveCriticalSection(Date_t end) {
     stdx::lock_guard<Latch> lk(_mutex);
-    if (!_currentOp) {
-        return;
-    }
-
     _currentOp->inCriticalSection.forceEnd(end);
 }
 
@@ -723,6 +662,22 @@ void ReshardingMetrics::onOplogEntriesApplied(int64_t entries) noexcept {
     _cumulativeOp->oplogEntriesApplied += entries;
 }
 
+void ReshardingMetrics::restoreForCurrentOp(int64_t documentCountCopied,
+                                            int64_t documentBytesCopied,
+                                            int64_t oplogEntriesFetched,
+                                            int64_t oplogEntriesApplied) noexcept {
+    invariant(_currentOp, kNoOperationInProgress);
+    invariant(_currentOp->documentsCopied == 0, kMetricsSetBeforeRestore);
+    invariant(_currentOp->bytesCopied == 0, kMetricsSetBeforeRestore);
+    invariant(_currentOp->oplogEntriesFetched == 0, kMetricsSetBeforeRestore);
+    invariant(_currentOp->oplogEntriesApplied == 0, kMetricsSetBeforeRestore);
+
+    _currentOp->documentsCopied = documentCountCopied;
+    _currentOp->bytesCopied = documentBytesCopied;
+    _currentOp->oplogEntriesFetched = oplogEntriesFetched;
+    _currentOp->oplogEntriesApplied = oplogEntriesApplied;
+}
+
 void ReshardingMetrics::onWriteDuringCriticalSection(int64_t writes) noexcept {
     stdx::lock_guard<Latch> lk(_mutex);
     if (!_currentOp)
diff --git a/src/mongo/db/s/resharding/resharding_metrics.h b/src/mongo/db/s/resharding/resharding_metrics.h
index 9ba80d35417..a6964c9d611 100644
--- a/src/mongo/db/s/resharding/resharding_metrics.h
+++ b/src/mongo/db/s/resharding/resharding_metrics.h
@@ -35,7 +35,6 @@
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/s/resharding/donor_document_gen.h"
-#include "mongo/db/s/resharding/recipient_document_gen.h"
 #include "mongo/db/service_context.h"
 #include "mongo/platform/mutex.h"
 #include "mongo/s/resharding/common_types_gen.h"
@@ -72,29 +71,6 @@ public:
 
     void onStepUp(DonorStateEnum state, ReshardingDonorMetrics donorMetrics);
 
-    struct ReshardingRecipientCountsAndMetrics {
-        ReshardingRecipientCountsAndMetrics(int64_t documentCountCopied,
-                                            int64_t documentBytesCopied,
-                                            int64_t oplogEntriesFetched,
-                                            int64_t oplogEntriesApplied,
-                                            boost::optional<int64_t> approxBytesToCopy,
-                                            ReshardingRecipientMetrics metrics)
-            : documentCountCopied{documentCountCopied},
-              documentBytesCopied{documentBytesCopied},
-              oplogEntriesFetched{oplogEntriesFetched},
-              oplogEntriesApplied{oplogEntriesApplied},
-              approxBytesToCopy{approxBytesToCopy},
-              metrics{metrics} {}
-        int64_t documentCountCopied;
-        int64_t documentBytesCopied;
-        int64_t oplogEntriesFetched;
-        int64_t oplogEntriesApplied;
-        boost::optional<int64_t> approxBytesToCopy;
-        ReshardingRecipientMetrics metrics;
-    };
-
-    void onStepUp(RecipientStateEnum, const ReshardingRecipientCountsAndMetrics&);
-
     // So long as a resharding operation is in progress, the following may be used to update the
     // state of a donor, a recipient, and a coordinator, respectively.
     void setDonorState(DonorStateEnum) noexcept;
@@ -137,6 +113,11 @@ public:
     // Allows restoring "oplog entries to apply" metrics.
     void onOplogEntriesApplied(int64_t entries) noexcept;
 
+    void restoreForCurrentOp(int64_t documentCountCopied,
+                             int64_t documentBytesCopied,
+                             int64_t oplogEntriesFetched,
+                             int64_t oplogEntriesApplied) noexcept;
+
     // Allows tracking writes during a critical section when the donor's state is either of
     // "donating-oplog-entries" or "blocking-writes".
     void onWriteDuringCriticalSection(int64_t writes) noexcept;
diff --git a/src/mongo/db/s/resharding/resharding_op_observer.cpp b/src/mongo/db/s/resharding/resharding_op_observer.cpp
index cd2fb324c8e..ba1f768f552 100644
--- a/src/mongo/db/s/resharding/resharding_op_observer.cpp
+++ b/src/mongo/db/s/resharding/resharding_op_observer.cpp
@@ -48,7 +48,7 @@ std::shared_ptr<ReshardingCoordinatorObserver> getReshardingCoordinatorObserver(
     OperationContext* opCtx, const BSONObj& reshardingId) {
     auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext());
     auto service = registry->lookupServiceByName(ReshardingCoordinatorService::kServiceName);
-    auto [instance, _] =
+    auto instance =
         ReshardingCoordinatorService::ReshardingCoordinator::lookup(opCtx, service, reshardingId);
 
     iassert(
diff --git a/src/mongo/db/s/resharding/resharding_op_observer.h b/src/mongo/db/s/resharding/resharding_op_observer.h
index e8affe3ef4a..30d319a041d 100644
--- a/src/mongo/db/s/resharding/resharding_op_observer.h
+++ b/src/mongo/db/s/resharding/resharding_op_observer.h
@@ -228,10 +228,6 @@ public:
         size_t numberOfPrePostImagesToWrite,
         Date_t wallClockTime) override {}
 
-    void onTransactionPrepareNonPrimary(OperationContext* opCtx,
-                                        const std::vector<repl::OplogEntry>& statements,
-                                        const repl::OpTime& prepareOpTime) override {}
-
     void onTransactionAbort(OperationContext* opCtx,
                             boost::optional<OplogSlot> abortOplogEntryOpTime) override {}
 
diff --git a/src/mongo/db/s/resharding/resharding_oplog_application.cpp b/src/mongo/db/s/resharding/resharding_oplog_application.cpp
index 42c5a1543e5..9a2f6f2750e 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_application.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_application.cpp
@@ -29,9 +29,11 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kResharding
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/s/resharding/resharding_oplog_application.h"
 
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/logical_session_cache.h"
@@ -77,8 +79,8 @@ void runWithTransaction(OperationContext* opCtx,
 
     // ReshardingOpObserver depends on the collection metadata being known when processing writes to
     // the temporary resharding collection. We attach shard version IGNORED to the write operations
-    // and leave it to ReshardingOplogBatchApplier::applyBatch() to retry on a StaleConfig error to
-    // allow the collection metadata information to be recovered.
+    // and leave it to ReshardingOplogBatchApplier::applyBatch() to retry on a StaleConfig exception
+    // to allow the collection metadata information to be recovered.
     ScopedSetShardRole scopedSetShardRole(asr.opCtx(),
                                           nss,
                                           ChunkVersion::IGNORED() /* shardVersion */,
diff --git a/src/mongo/db/s/resharding/resharding_oplog_applier.cpp b/src/mongo/db/s/resharding/resharding_oplog_applier.cpp
index 71d156170d4..766ac67f266 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_applier.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_applier.cpp
@@ -236,8 +236,7 @@ void ReshardingOplogApplier::_clearAppliedOpsAndStoreProgress(OperationContext*
     store.upsert(
         opCtx,
         BSON(ReshardingOplogApplierProgress::kOplogSourceIdFieldName << _sourceId.toBSON()),
-        builder.obj(),
-        WriteConcerns::kLocalWriteConcern);
+        builder.obj());
     _env->metrics()->onOplogEntriesApplied(_currentBatchToApply.size());
     if (ShardingDataTransformMetrics::isEnabled()) {
         _env->metricsNew()->onOplogEntriesApplied(_currentBatchToApply.size());
diff --git a/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp b/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp
index 774a654ce8f..02334d1f4bf 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp
@@ -249,7 +249,6 @@ public:
                                         kCrudNs,
                                         kCrudUUID,
                                         false /* fromMigrate */,
-                                        boost::none,  // checkExistenceForDiffInsert
                                         0 /* version */,
                                         obj1,
                                         obj2,
diff --git a/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp b/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp
index c7efa6d33e4..4ff29b42d30 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp
@@ -29,8 +29,12 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kResharding
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/s/resharding/resharding_oplog_batch_applier.h"
 
+#include <memory>
+
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/s/resharding/resharding_data_copy_util.h"
 #include "mongo/db/s/resharding/resharding_future_util.h"
@@ -79,8 +83,8 @@ SemiFuture<void> ReshardingOplogBatchApplier::applyBatch(
                            // ReshardingOpObserver depends on the collection metadata being known
                            // when processing writes to the temporary resharding collection. We
                            // attach shard version IGNORED to the write operations and retry once
-                           // on a StaleConfig error to allow the collection metadata information to
-                           // be recovered.
+                           // on a StaleConfig exception to allow the collection metadata
+                           // information to be recovered.
                            ScopedSetShardRole scopedSetShardRole(
                                opCtx.get(),
                                _crudApplication.getOutputNss(),
diff --git a/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp b/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp
index 064ca34c9fa..30811c8d2aa 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp
@@ -41,7 +41,7 @@
 #include "mongo/client/dbclient_connection.h"
 #include "mongo/client/remote_command_targeter.h"
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/pipeline/aggregate_command_gen.h"
 #include "mongo/db/repl/read_concern_args.h"
 #include "mongo/db/repl/read_concern_level.h"
@@ -194,22 +194,23 @@ ExecutorFuture<void> ReshardingOplogFetcher::_reschedule(
 }
 
 bool ReshardingOplogFetcher::iterate(Client* client, CancelableOperationContextFactory factory) {
-    try {
-        std::shared_ptr<Shard> targetShard;
-        {
-            auto opCtxRaii = factory.makeOperationContext(client);
-            opCtxRaii->checkForInterrupt();
-
-            StatusWith<std::shared_ptr<Shard>> swDonor =
-                Grid::get(opCtxRaii.get())->shardRegistry()->getShard(opCtxRaii.get(), _donorShard);
-            if (!swDonor.isOK()) {
-                LOGV2_WARNING(5127203,
-                              "Error finding shard in registry, retrying.",
-                              "error"_attr = swDonor.getStatus());
-                return true;
-            }
-            targetShard = swDonor.getValue();
+    std::shared_ptr<Shard> targetShard;
+    {
+        auto opCtxRaii = factory.makeOperationContext(client);
+        opCtxRaii->checkForInterrupt();
+
+        StatusWith<std::shared_ptr<Shard>> swDonor =
+            Grid::get(opCtxRaii.get())->shardRegistry()->getShard(opCtxRaii.get(), _donorShard);
+        if (!swDonor.isOK()) {
+            LOGV2_WARNING(5127203,
+                          "Error finding shard in registry, retrying.",
+                          "error"_attr = swDonor.getStatus());
+            return true;
         }
+        targetShard = swDonor.getValue();
+    }
+
+    try {
         return consume(client, factory, targetShard.get());
     } catch (const ExceptionForCat<ErrorCategory::Interruption>&) {
         // Defer to the cancellation token for whether the Interruption exception should be retried
diff --git a/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp b/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp
index 99707c61b96..e7c512e2669 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp
@@ -36,7 +36,7 @@
 
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/dbhelpers.h"
@@ -144,9 +144,7 @@ public:
             StaticCatalogClient(std::vector<ShardId> shardIds) : _shardIds(std::move(shardIds)) {}
 
             StatusWith<repl::OpTimeWith<std::vector<ShardType>>> getAllShards(
-                OperationContext* opCtx,
-                repl::ReadConcernLevel readConcern,
-                bool excludeDraining) override {
+                OperationContext* opCtx, repl::ReadConcernLevel readConcern) override {
                 std::vector<ShardType> shardTypes;
                 for (const auto& shardId : _shardIds) {
                     const ConnectionString cs = ConnectionString::forReplicaSet(
@@ -668,44 +666,6 @@ TEST_F(ReshardingOplogFetcherTest, RetriesOnRemoteInterruptionError) {
     ASSERT_TRUE(moreToCome);
 }
 
-TEST_F(ReshardingOplogFetcherTest, RetriesOnNetworkTimeoutError) {
-    const NamespaceString outputCollectionNss("dbtests.outputCollection");
-    const NamespaceString dataCollectionNss("dbtests.runFetchIteration");
-
-    create(outputCollectionNss);
-    create(dataCollectionNss);
-    _fetchTimestamp = repl::StorageInterface::get(_svcCtx)->getLatestOplogTimestamp(_opCtx);
-
-    const auto& collectionUUID = [&] {
-        AutoGetCollection dataColl(_opCtx, dataCollectionNss, LockMode::MODE_IX);
-        return dataColl->uuid();
-    }();
-
-    auto fetcherJob = launchAsync([&, this] {
-        ThreadClient tc("RunnerForFetcher", _svcCtx, nullptr);
-
-        ReshardingDonorOplogId startAt{_fetchTimestamp, _fetchTimestamp};
-        ReshardingOplogFetcher fetcher(makeFetcherEnv(),
-                                       _reshardingUUID,
-                                       collectionUUID,
-                                       startAt,
-                                       _donorShard,
-                                       _destinationShard,
-                                       outputCollectionNss);
-
-        auto factory = makeCancelableOpCtx();
-        return fetcher.iterate(&cc(), factory);
-    });
-
-    onCommand([&](const executor::RemoteCommandRequest& request) -> StatusWith<BSONObj> {
-        // Inject network timeout error.
-        return {ErrorCodes::NetworkInterfaceExceededTimeLimit, "exceeded network time limit"};
-    });
-
-    auto moreToCome = fetcherJob.timed_get(Seconds(5));
-    ASSERT_TRUE(moreToCome);
-}
-
 TEST_F(ReshardingOplogFetcherTest, ImmediatelyDoneWhenFinalOpHasAlreadyBeenFetched) {
     const NamespaceString outputCollectionNss("dbtests.outputCollection");
     const NamespaceString dataCollectionNss("dbtests.runFetchIteration");
diff --git a/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp b/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp
index 95c29d0825c..44251fa39dc 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp
@@ -32,7 +32,7 @@
 #include "mongo/db/s/resharding/resharding_oplog_session_application.h"
 
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/oplog_entry.h"
@@ -108,7 +108,6 @@ boost::optional<SharedSemiFuture<void>> ReshardingOplogSessionApplication::tryAp
     invariant(op.getTxnNumber());
     invariant(op.get_id());
 
-    auto sourceNss = op.getNss();
     auto lsid = *op.getSessionId();
     if (isInternalSessionForNonRetryableWrite(lsid)) {
         // Skip internal sessions for non-retryable writes since they only support transactions
@@ -158,8 +157,7 @@ boost::optional<SharedSemiFuture<void>> ReshardingOplogSessionApplication::tryAp
                                                        std::move(o2Field),
                                                        std::move(stmtIds),
                                                        std::move(preImageOpTime),
-                                                       std::move(postImageOpTime),
-                                                       std::move(sourceNss));
+                                                       std::move(postImageOpTime));
         });
 }
 
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.cpp b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
index 087e3cb603f..5d49e0cdb22 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
@@ -36,7 +36,7 @@
 #include "mongo/db/cancelable_operation_context.h"
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/ops/delete.h"
@@ -56,7 +56,6 @@
 #include "mongo/db/s/resharding/resharding_recipient_service_external_state.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
 #include "mongo/db/s/shard_key_util.h"
-#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/write_block_bypass.h"
 #include "mongo/executor/network_interface_factory.h"
@@ -146,8 +145,6 @@ ReshardingRecipientService::RecipientStateMachine::RecipientStateMachine(
       _recipientCtx{recipientDoc.getMutableState()},
       _donorShards{recipientDoc.getDonorShards()},
       _cloneTimestamp{recipientDoc.getCloneTimestamp()},
-      _timeIntervals{recipientDoc.getMetrics().get_value_or({})},
-      _approxBytesToCopy{recipientDoc.getApproxBytesToCopy()},
       _externalState{std::move(externalState)},
       _startConfigTxnCloneAt{recipientDoc.getStartConfigTxnCloneTime()},
       _markKilledExecutor(std::make_shared<ThreadPool>([] {
@@ -169,7 +166,6 @@ ReshardingRecipientService::RecipientStateMachine::RecipientStateMachine(
                                   return donor.getShardId() == myShardId;
                               }) != _donorShards.end();
       }()) {
-
     invariant(_externalState);
 }
 
@@ -557,11 +553,6 @@ ReshardingRecipientService::RecipientStateMachine::_makeDataReplication(Operatio
                                                                         bool cloningDone) {
     invariant(_cloneTimestamp);
 
-    // We refresh the routing information for the source collection to ensure the
-    // ReshardingOplogApplier is making its decisions according to the chunk distribution after the
-    // sharding metadata was frozen.
-    _externalState->refreshCatalogCache(opCtx, _metadata.getSourceNss());
-
     auto myShardId = _externalState->myShardId(opCtx->getServiceContext());
     auto sourceChunkMgr =
         _externalState->getShardedCollectionRoutingInfo(opCtx, _metadata.getSourceNss());
@@ -761,7 +752,7 @@ void ReshardingRecipientService::RecipientStateMachine::_cleanupReshardingCollec
         opCtx.get(), _metadata.getReshardingUUID(), _metadata.getSourceUUID(), _donorShards);
 
     if (aborted) {
-        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(
+        resharding::data_copy::ensureCollectionDropped(
             opCtx.get(), _metadata.getTempReshardingNss(), _metadata.getReshardingUUID());
     }
 }
@@ -817,53 +808,27 @@ void ReshardingRecipientService::RecipientStateMachine::_transitionToCloning(
     const CancelableOperationContextFactory& factory) {
     auto newRecipientCtx = _recipientCtx;
     newRecipientCtx.setState(RecipientStateEnum::kCloning);
-    auto cloningStartTime = getCurrentTime();
-
-    // Record cloning start time.
-    ReshardingMetricsTimeInterval interval;
-    interval.setStart(cloningStartTime);
-    _timeIntervals.setDocumentCopy(interval);
-
     _transitionState(std::move(newRecipientCtx), boost::none, boost::none, factory);
-    _metrics()->startCopyingDocuments(cloningStartTime);
+    _metrics()->startCopyingDocuments(getCurrentTime());
 }
 
 void ReshardingRecipientService::RecipientStateMachine::_transitionToApplying(
     const CancelableOperationContextFactory& factory) {
     auto newRecipientCtx = _recipientCtx;
     newRecipientCtx.setState(RecipientStateEnum::kApplying);
-    auto oplogApplicationStartTime = getCurrentTime();
-
-    // Record oplog application start time.
-    ReshardingMetricsTimeInterval interval;
-    interval.setStart(oplogApplicationStartTime);
-    _timeIntervals.setOplogApplication(interval);
-
-    // Record document copy stop time.
-    ReshardingMetricsTimeInterval documentCopy{_timeIntervals.getDocumentCopy().get_value_or({})};
-    documentCopy.setStop(oplogApplicationStartTime);
-    _timeIntervals.setDocumentCopy(documentCopy);
-
     _transitionState(std::move(newRecipientCtx), boost::none, boost::none, factory);
-    _metrics()->endCopyingDocuments(oplogApplicationStartTime);
-    _metrics()->startApplyingOplogEntries(oplogApplicationStartTime);
+    auto currentTime = getCurrentTime();
+    _metrics()->endCopyingDocuments(currentTime);
+    _metrics()->startApplyingOplogEntries(currentTime);
 }
 
 void ReshardingRecipientService::RecipientStateMachine::_transitionToStrictConsistency(
     const CancelableOperationContextFactory& factory) {
     auto newRecipientCtx = _recipientCtx;
     newRecipientCtx.setState(RecipientStateEnum::kStrictConsistency);
-    auto oplogApplicationStopTime = getCurrentTime();
-
-    // Record oplog application stop time
-    ReshardingMetricsTimeInterval oplogApplication{
-        _timeIntervals.getOplogApplication().get_value_or({})};
-    oplogApplication.setStop(oplogApplicationStopTime);
-    _timeIntervals.setOplogApplication(oplogApplication);
-
-
     _transitionState(std::move(newRecipientCtx), boost::none, boost::none, factory);
-    _metrics()->endApplyingOplogEntries(oplogApplicationStopTime);
+    auto currentTime = getCurrentTime();
+    _metrics()->endApplyingOplogEntries(currentTime);
 }
 
 void ReshardingRecipientService::RecipientStateMachine::_transitionToError(
@@ -1015,9 +980,6 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument
 
             setBuilder.append(ReshardingRecipientDocument::kDonorShardsFieldName,
                               donorShardsArrayBuilder.arr());
-
-            setBuilder.append(ReshardingRecipientDocument::kApproxBytesToCopyFieldName,
-                              cloneDetails->approxBytesToCopy);
         }
 
         if (configStartTime) {
@@ -1025,8 +987,6 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument
                               *configStartTime);
         }
 
-        setBuilder.append(ReshardingRecipientDocument::kMetricsFieldName, _timeIntervals.toBSON());
-
         setBuilder.doneFast();
     }
 
@@ -1044,7 +1004,6 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument
     if (cloneDetails) {
         _cloneTimestamp = cloneDetails->cloneTimestamp;
         _donorShards = std::move(cloneDetails->donorShards);
-        _approxBytesToCopy = cloneDetails->approxBytesToCopy;
     }
 
     if (configStartTime) {
@@ -1104,6 +1063,7 @@ ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_startMe
     const std::shared_ptr<executor::ScopedTaskExecutor>& executor,
     const CancellationToken& abortToken) {
     if (_recipientCtx.getState() > RecipientStateEnum::kAwaitingFetchTimestamp) {
+        _metrics()->onStepUp(ReshardingMetrics::Role::kRecipient);
         return _restoreMetricsWithRetry(executor, abortToken);
     }
     _metrics()->onStart(ReshardingMetrics::Role::kRecipient, getCurrentTime());
@@ -1113,6 +1073,7 @@ ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_startMe
 ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_restoreMetricsWithRetry(
     const std::shared_ptr<executor::ScopedTaskExecutor>& executor,
     const CancellationToken& abortToken) {
+    _metrics()->setRecipientState(_recipientCtx.getState());
     return _retryingCancelableOpCtxFactory
         ->withAutomaticRetry(
             [this, executor, abortToken](const auto& factory) { _restoreMetrics(factory); })
@@ -1178,13 +1139,8 @@ void ReshardingRecipientService::RecipientStateMachine::_restoreMetrics(
         }
     }
 
-    _metrics()->onStepUp(_recipientCtx.getState(),
-                         ReshardingMetrics::ReshardingRecipientCountsAndMetrics{documentCountCopied,
-                                                                                documentBytesCopied,
-                                                                                oplogEntriesFetched,
-                                                                                oplogEntriesApplied,
-                                                                                _approxBytesToCopy,
-                                                                                _timeIntervals});
+    _metrics()->restoreForCurrentOp(
+        documentCountCopied, documentBytesCopied, oplogEntriesFetched, oplogEntriesApplied);
 }
 
 CancellationToken ReshardingRecipientService::RecipientStateMachine::_initAbortSource(
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.h b/src/mongo/db/s/resharding/resharding_recipient_service.h
index 86d3236b9c9..e1fe504a970 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service.h
+++ b/src/mongo/db/s/resharding/resharding_recipient_service.h
@@ -288,8 +288,6 @@ private:
     RecipientShardContext _recipientCtx;
     std::vector<DonorShardFetchTimestamp> _donorShards;
     boost::optional<Timestamp> _cloneTimestamp;
-    ReshardingRecipientMetrics _timeIntervals;
-    boost::optional<int64_t> _approxBytesToCopy;
 
     const std::unique_ptr<RecipientStateMachineExternalState> _externalState;
 
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service_external_state.cpp b/src/mongo/db/s/resharding/resharding_recipient_service_external_state.cpp
index b99f1f60437..2dc584953d6 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service_external_state.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service_external_state.cpp
@@ -124,8 +124,13 @@ RecipientStateMachineExternalStateImpl::getCollectionOptions(OperationContext* o
                                                              StringData reason) {
     // Load the collection options from the primary shard for the database.
     return _withShardVersionRetry(opCtx, nss, reason, [&] {
+        auto cm = getShardedCollectionRoutingInfo(opCtx, nss);
         return MigrationDestinationManager::getCollectionOptions(
-            opCtx, NamespaceStringOrUUID{nss.db().toString(), uuid}, afterClusterTime);
+            opCtx,
+            NamespaceStringOrUUID{nss.db().toString(), uuid},
+            cm.dbPrimary(),
+            cm,
+            afterClusterTime);
     });
 }
 
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
index b598f32baf1..43ccab32bec 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
@@ -48,13 +48,10 @@
 #include "mongo/db/s/resharding/resharding_recipient_service.h"
 #include "mongo/db/s/resharding/resharding_recipient_service_external_state.h"
 #include "mongo/db/s/resharding/resharding_service_test_helpers.h"
-#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/unittest/death_test.h"
-#include "mongo/util/clock_source_mock.h"
 #include "mongo/util/fail_point.h"
 
-
 namespace mongo {
 namespace {
 
@@ -211,8 +208,6 @@ public:
  */
 class ReshardingRecipientServiceTest : public repl::PrimaryOnlyServiceMongoDTest {
 public:
-    ReshardingRecipientServiceTest() : PrimaryOnlyServiceMongoDTest(Options{}.useMockClock(true)) {}
-
     using RecipientStateMachine = ReshardingRecipientService::RecipientStateMachine;
 
     std::unique_ptr<repl::PrimaryOnlyService> makeService(ServiceContext* serviceContext) override {
@@ -227,6 +222,7 @@ public:
         repl::DropPendingCollectionReaper::set(
             serviceContext, std::make_unique<repl::DropPendingCollectionReaper>(storageMock.get()));
         repl::StorageInterface::set(serviceContext, std::move(storageMock));
+
         _controller = std::make_shared<RecipientStateTransitionController>();
         _opObserverRegistry->addObserver(std::make_unique<RecipientOpObserverForTest>(_controller));
     }
@@ -267,8 +263,7 @@ public:
                                 const ReshardingRecipientDocument& recipientDoc) {
         CollectionOptions options;
         options.uuid = recipientDoc.getSourceUUID();
-        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx,
-                                                                       recipientDoc.getSourceNss());
+        resharding::data_copy::ensureCollectionDropped(opCtx, recipientDoc.getSourceNss());
         resharding::data_copy::ensureCollectionExists(opCtx, recipientDoc.getSourceNss(), options);
     }
 
@@ -379,10 +374,9 @@ TEST_F(ReshardingRecipientServiceTest, StepDownStepUpEachTransition) {
                     RecipientStateMachine::insertStateDocument(opCtx.get(), doc);
                     return RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON());
                 } else {
-                    auto [maybeRecipient, isPausedOrShutdown] =
+                    auto maybeRecipient =
                         RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
-                    ASSERT_TRUE(maybeRecipient);
-                    ASSERT_FALSE(isPausedOrShutdown);
+                    ASSERT_TRUE(bool(maybeRecipient));
 
                     // Allow the transition to prevState to succeed on this primary-only service
                     // instance.
@@ -429,10 +423,8 @@ TEST_F(ReshardingRecipientServiceTest, StepDownStepUpEachTransition) {
         }
 
         // Finally complete the operation and ensure its success.
-        auto [maybeRecipient, isPausedOrShutdown] =
-            RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
-        ASSERT_TRUE(maybeRecipient);
-        ASSERT_FALSE(isPausedOrShutdown);
+        auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
+        ASSERT_TRUE(bool(maybeRecipient));
 
         auto recipient = *maybeRecipient;
 
@@ -477,10 +469,8 @@ TEST_F(ReshardingRecipientServiceTest, OpCtxKilledWhileRestoringMetrics) {
         stepUp(opCtx.get());
 
         // After the failpoint is disabled, the operation should succeed.
-        auto [maybeRecipient, isPausedOrShutdown] =
-            RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
-        ASSERT_TRUE(maybeRecipient);
-        ASSERT_FALSE(isPausedOrShutdown);
+        auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
+        ASSERT_TRUE(bool(maybeRecipient));
         recipient = *maybeRecipient;
         notifyReshardingCommitting(opCtx.get(), *recipient, doc);
         ASSERT_OK(recipient->getCompletionFuture().getNoThrow());
@@ -506,89 +496,69 @@ DEATH_TEST_REGEX_F(ReshardingRecipientServiceTest, CommitFn, "4457001.*tripwire"
 TEST_F(ReshardingRecipientServiceTest, DropsTemporaryReshardingCollectionOnAbort) {
     auto metrics = ReshardingRecipientServiceTest::metrics();
     for (bool isAlsoDonor : {false, true}) {
-        for (bool waitForMetricsInitialized : {false, true}) {
-            LOGV2(5551107,
-                  "Running case",
-                  "test"_attr = _agent.getTestName(),
-                  "isAlsoDonor"_attr = isAlsoDonor,
-                  "waitForMetricsInitialized"_attr = waitForMetricsInitialized);
-
-            boost::optional<PauseDuringStateTransitions> stateTransitionGuard;
-            if (waitForMetricsInitialized) {
-                std::vector<RecipientStateEnum> recipientStates{
-                    RecipientStateEnum::kDone, RecipientStateEnum::kCreatingCollection};
-                stateTransitionGuard.emplace(controller(), recipientStates);
-            } else {
-                stateTransitionGuard.emplace(controller(), RecipientStateEnum::kDone);
-            }
-
-            auto doc = makeStateDocument(isAlsoDonor);
-            auto instanceId = BSON(ReshardingRecipientDocument::kReshardingUUIDFieldName
-                                   << doc.getReshardingUUID());
+        LOGV2(5551107,
+              "Running case",
+              "test"_attr = _agent.getTestName(),
+              "isAlsoDonor"_attr = isAlsoDonor);
 
-            auto opCtx = makeOperationContext();
+        boost::optional<PauseDuringStateTransitions> doneTransitionGuard;
+        doneTransitionGuard.emplace(controller(), RecipientStateEnum::kDone);
 
-            if (isAlsoDonor) {
-                // If the recipient is also a donor, the original collection should already exist on
-                // this shard.
-                createSourceCollection(opCtx.get(), doc);
-            }
+        auto doc = makeStateDocument(isAlsoDonor);
+        auto instanceId =
+            BSON(ReshardingRecipientDocument::kReshardingUUIDFieldName << doc.getReshardingUUID());
 
-            RecipientStateMachine::insertStateDocument(opCtx.get(), doc);
-            auto recipient =
-                RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON());
+        auto opCtx = makeOperationContext();
 
-            notifyToStartCloning(opCtx.get(), *recipient, doc);
-            if (waitForMetricsInitialized) {
-                // Waiting for the metrics to be initialized here causes the second abort to occur
-                // before the metrics are initialized after the step up, thereby testing a different
-                // code path.
-                stateTransitionGuard->wait(RecipientStateEnum::kCreatingCollection);
-            }
+        if (isAlsoDonor) {
+            // If the recipient is also a donor, the original collection should already exist on
+            // this shard.
+            createSourceCollection(opCtx.get(), doc);
+        }
 
-            recipient->abort(false);
+        RecipientStateMachine::insertStateDocument(opCtx.get(), doc);
+        auto recipient = RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON());
 
-            stateTransitionGuard->wait(RecipientStateEnum::kDone);
+        notifyToStartCloning(opCtx.get(), *recipient, doc);
+        recipient->abort(false);
 
-            stepDown();
+        doneTransitionGuard->wait(RecipientStateEnum::kDone);
+        stepDown();
 
-            ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(),
-                      ErrorCodes::InterruptedDueToReplStateChange);
+        ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(),
+                  ErrorCodes::InterruptedDueToReplStateChange);
 
-            recipient.reset();
-            stepUp(opCtx.get());
+        recipient.reset();
+        stepUp(opCtx.get());
 
-            auto [maybeRecipient, isPausedOrShutdown] =
-                RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
-            ASSERT_TRUE(maybeRecipient);
-            ASSERT_FALSE(isPausedOrShutdown);
-            recipient = *maybeRecipient;
+        auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
+        ASSERT_TRUE(bool(maybeRecipient));
+        recipient = *maybeRecipient;
 
-            stateTransitionGuard.reset();
-            recipient->abort(false);
+        doneTransitionGuard.reset();
+        recipient->abort(false);
 
-            ASSERT_OK(recipient->getCompletionFuture().getNoThrow());
-            checkStateDocumentRemoved(opCtx.get());
+        ASSERT_OK(recipient->getCompletionFuture().getNoThrow());
+        checkStateDocumentRemoved(opCtx.get());
 
-            if (isAlsoDonor) {
-                // Verify original collection still exists after aborting.
-                AutoGetCollection coll(opCtx.get(), doc.getSourceNss(), MODE_IS);
-                ASSERT_TRUE(bool(coll));
-                ASSERT_EQ(coll->uuid(), doc.getSourceUUID());
-            }
+        if (isAlsoDonor) {
+            // Verify original collection still exists after aborting.
+            AutoGetCollection coll(opCtx.get(), doc.getSourceNss(), MODE_IS);
+            ASSERT_TRUE(bool(coll));
+            ASSERT_EQ(coll->uuid(), doc.getSourceUUID());
+        }
 
-            // Verify the temporary collection no longer exists.
-            {
-                AutoGetCollection coll(opCtx.get(), doc.getTempReshardingNss(), MODE_IS);
-                ASSERT_FALSE(bool(coll));
-            }
+        // Verify the temporary collection no longer exists.
+        {
+            AutoGetCollection coll(opCtx.get(), doc.getTempReshardingNss(), MODE_IS);
+            ASSERT_FALSE(bool(coll));
         }
     }
 
     BSONObjBuilder result;
     metrics->serializeCumulativeOpMetrics(&result);
 
-    ASSERT_LESS_THAN_OR_EQUALS(result.obj().getField("countReshardingFailures").numberLong(), 4);
+    ASSERT_EQ(result.obj().getField("countReshardingFailures").numberLong(), 2);
 }
 
 TEST_F(ReshardingRecipientServiceTest, RenamesTemporaryReshardingCollectionWhenDone) {
@@ -778,10 +748,9 @@ TEST_F(ReshardingRecipientServiceTest, RestoreMetricsAfterStepUp) {
                 RecipientStateMachine::insertStateDocument(opCtx.get(), doc);
                 return RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON());
             } else {
-                auto [maybeRecipient, isPausedOrShutdown] =
+                auto maybeRecipient =
                     RecipientStateMachine::lookup(opCtx.get(), _service, instanceId);
-                ASSERT_TRUE(maybeRecipient);
-                ASSERT_FALSE(isPausedOrShutdown);
+                ASSERT_TRUE(bool(maybeRecipient));
 
                 // Allow the transition to prevState to succeed on this primary-only service
                 // instance.
@@ -847,33 +816,22 @@ TEST_F(ReshardingRecipientServiceTest, RestoreMetricsAfterStepUp) {
         }
         // Step down before the transition to state can complete.
         stateTransitionsGuard.wait(state);
-
-        dynamic_cast<ClockSourceMock*>(getServiceContext()->getFastClockSource())
-            ->advance(Seconds(1));
-        auto currOp =
-            recipient
-                ->reportForCurrentOp(MongoProcessInterface::CurrentOpConnectionsMode::kExcludeIdle,
-                                     MongoProcessInterface::CurrentOpSessionsMode::kExcludeIdle)
-                .get();
-
-
-        if (state == RecipientStateEnum::kApplying) {
-            ASSERT_EQ(currOp.getField("totalApplyTimeElapsedSecs").Long(), 0);
-            ASSERT_EQ(currOp.getStringField("recipientState"),
-                      RecipientState_serializer(RecipientStateEnum::kCloning));
-            ASSERT_GT(currOp.getField("totalCopyTimeElapsedSecs").Long(), 0);
-            ASSERT_GT(currOp.getField("approxBytesToCopy").Long(), 0);
-
-        } else if (state == RecipientStateEnum::kStrictConsistency) {
+        if (state == RecipientStateEnum::kStrictConsistency) {
+            auto currOp = recipient
+                              ->reportForCurrentOp(
+                                  MongoProcessInterface::CurrentOpConnectionsMode::kExcludeIdle,
+                                  MongoProcessInterface::CurrentOpSessionsMode::kExcludeIdle)
+                              .get();
             ASSERT_EQ(currOp.getField("documentsCopied").Long(), 1L);
             ASSERT_EQ(currOp.getField("bytesCopied").Long(), (long)reshardedDoc.objsize());
             ASSERT_EQ(currOp.getStringField("recipientState"),
                       RecipientState_serializer(RecipientStateEnum::kApplying));
-            ASSERT_GT(currOp.getField("totalCopyTimeElapsedSecs").Long(), 0);
-            ASSERT_GT(currOp.getField("totalApplyTimeElapsedSecs").Long(), 0);
-            ASSERT_GT(currOp.getField("approxBytesToCopy").Long(), 0);
-
         } else if (state == RecipientStateEnum::kDone) {
+            auto currOp = recipient
+                              ->reportForCurrentOp(
+                                  MongoProcessInterface::CurrentOpConnectionsMode::kExcludeIdle,
+                                  MongoProcessInterface::CurrentOpSessionsMode::kExcludeIdle)
+                              .get();
             ASSERT_EQ(currOp.getField("documentsCopied").Long(), 1L);
             ASSERT_EQ(currOp.getField("bytesCopied").Long(), (long)reshardedDoc.objsize());
             ASSERT_EQ(currOp.getField("oplogEntriesFetched").Long(),
@@ -882,11 +840,7 @@ TEST_F(ReshardingRecipientServiceTest, RestoreMetricsAfterStepUp) {
                       oplogEntriesAppliedOnEachDonor * doc.getDonorShards().size());
             ASSERT_EQ(currOp.getStringField("recipientState"),
                       RecipientState_serializer(RecipientStateEnum::kStrictConsistency));
-            ASSERT_GT(currOp.getField("totalCopyTimeElapsedSecs").Long(), 0);
-            ASSERT_GT(currOp.getField("totalApplyTimeElapsedSecs").Long(), 0);
-            ASSERT_GT(currOp.getField("approxBytesToCopy").Long(), 0);
         }
-
         stepDown();
 
         ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(),
diff --git a/src/mongo/db/s/resharding/resharding_txn_cloner.cpp b/src/mongo/db/s/resharding/resharding_txn_cloner.cpp
index 96a08c210fa..7ab491286f3 100644
--- a/src/mongo/db/s/resharding/resharding_txn_cloner.cpp
+++ b/src/mongo/db/s/resharding/resharding_txn_cloner.cpp
@@ -40,6 +40,7 @@
 #include "mongo/client/read_preference.h"
 #include "mongo/db/client.h"
 #include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/logical_session_id.h"
 #include "mongo/db/namespace_string.h"
@@ -190,8 +191,7 @@ boost::optional<SharedSemiFuture<void>> ReshardingTxnCloner::doOneRecord(
                                                        TransactionParticipant::kDeadEndSentinel,
                                                        {kIncompleteHistoryStmtId},
                                                        boost::none /* preImageOpTime */,
-                                                       boost::none /* postImageOpTime */,
-                                                       {});
+                                                       boost::none /* postImageOpTime */);
         });
 }
 
diff --git a/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp b/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp
index b77db34ef11..e4f4f1f2b61 100644
--- a/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp
@@ -40,7 +40,6 @@
 #include "mongo/db/logical_session_cache_noop.h"
 #include "mongo/db/persistent_task_store.h"
 #include "mongo/db/pipeline/process_interface/shardsvr_process_interface.h"
-#include "mongo/db/query/cursor_response.h"
 #include "mongo/db/repl/storage_interface_impl.h"
 #include "mongo/db/repl/wait_for_majority_service.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
@@ -130,9 +129,7 @@ class ReshardingTxnClonerTest : public ShardServerTestFixture {
             StaticCatalogClient(std::vector<ShardId> shardIds) : _shardIds(std::move(shardIds)) {}
 
             StatusWith<repl::OpTimeWith<std::vector<ShardType>>> getAllShards(
-                OperationContext* opCtx,
-                repl::ReadConcernLevel readConcern,
-                bool excludeDraining) override {
+                OperationContext* opCtx, repl::ReadConcernLevel readConcern) override {
                 std::vector<ShardType> shardTypes;
                 for (const auto& shardId : _shardIds) {
                     const ConnectionString cs = ConnectionString::forReplicaSet(
diff --git a/src/mongo/db/s/resharding/resharding_util.cpp b/src/mongo/db/s/resharding/resharding_util.cpp
index 3e0db3b383e..27d319264dd 100644
--- a/src/mongo/db/s/resharding/resharding_util.cpp
+++ b/src/mongo/db/s/resharding/resharding_util.cpp
@@ -38,7 +38,7 @@
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/json.h"
 #include "mongo/bson/util/bson_extract.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/exec/document_value/document.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/op_observer.h"
@@ -207,14 +207,11 @@ void checkForOverlappingZones(std::vector<ReshardingZoneType>& zones) {
 }
 
 std::vector<BSONObj> buildTagsDocsFromZones(const NamespaceString& tempNss,
-                                            const std::vector<ReshardingZoneType>& zones,
-                                            const ShardKeyPattern& shardKey) {
+                                            const std::vector<ReshardingZoneType>& zones) {
     std::vector<BSONObj> tags;
     tags.reserve(zones.size());
-    for (auto& zone : zones) {
-        auto min = shardKey.getKeyPattern().extendRangeBound(zone.getMin(), false);
-        auto max = shardKey.getKeyPattern().extendRangeBound(zone.getMax(), false);
-        ChunkRange range(min, max);
+    for (const auto& zone : zones) {
+        ChunkRange range(zone.getMin(), zone.getMax());
         TagsType tag(tempNss, zone.getZone().toString(), range);
         tags.push_back(tag.toBSON());
     }
diff --git a/src/mongo/db/s/resharding/resharding_util.h b/src/mongo/db/s/resharding/resharding_util.h
index f0e60e99555..856d7cbb081 100644
--- a/src/mongo/db/s/resharding/resharding_util.h
+++ b/src/mongo/db/s/resharding/resharding_util.h
@@ -260,8 +260,7 @@ void checkForOverlappingZones(std::vector<ReshardingZoneType>& zones);
  * Builds documents to insert into config.tags from zones provided to reshardCollection cmd.
  */
 std::vector<BSONObj> buildTagsDocsFromZones(const NamespaceString& tempNss,
-                                            const std::vector<ReshardingZoneType>& zones,
-                                            const ShardKeyPattern& shardKey);
+                                            const std::vector<ReshardingZoneType>& zones);
 
 /**
  * Creates a pipeline that can be serialized into a query for fetching oplog entries. `startAfter`