Update upstream source from tag 'upstream/6.0.0'master

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

48 files changed, 671 insertions, 1454 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index d5210d9603c..143c5eae3ef 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -231,9 +231,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/commands',
-        '$BUILD_DIR/mongo/db/concurrency/exception_util',
-        '$BUILD_DIR/mongo/db/dbdirectclient',
-        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
+        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/storage/backup_cursor_hooks',
         'fsync_locked',
     ]
@@ -373,8 +372,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/multi_index_block',
         '$BUILD_DIR/mongo/db/command_can_run_here',
         '$BUILD_DIR/mongo/db/commands',
-        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
+        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/curop_failpoint_helpers',
         '$BUILD_DIR/mongo/db/exec/sbe/query_sbe_abt',
         '$BUILD_DIR/mongo/db/fle_crud_mongod',
@@ -387,8 +386,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/query/ce/query_ce',
         '$BUILD_DIR/mongo/db/query/command_request_response',
         '$BUILD_DIR/mongo/db/query/cursor_response_idl',
-        '$BUILD_DIR/mongo/db/query/query_shape/query_shape',
-        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
+        '$BUILD_DIR/mongo/db/query/optimizer/optimizer',
         '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/repl/replica_set_messages',
         '$BUILD_DIR/mongo/db/repl/tenant_migration_access_blocker',
@@ -403,7 +401,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/timeseries/catalog_helper',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_collmod',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_conversion_util',
-        '$BUILD_DIR/mongo/db/timeseries/timeseries_extended_range',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_stats',
         '$BUILD_DIR/mongo/db/transaction',
@@ -497,7 +494,6 @@ env.Library(
         'shutdown.idl',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/bson/bson_validate',
         '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/util/fail_point',
     ],
@@ -579,7 +575,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/repl/tenant_migration_donor_service',
         '$BUILD_DIR/mongo/db/repl/tenant_migration_recipient_service',
         '$BUILD_DIR/mongo/db/rw_concern_d',
-        '$BUILD_DIR/mongo/db/s/balancer_stats_registry',
         '$BUILD_DIR/mongo/db/s/sharding_api_d',
         '$BUILD_DIR/mongo/db/s/sharding_catalog_manager',
         '$BUILD_DIR/mongo/db/s/sharding_commands_d',
@@ -666,7 +661,6 @@ env.Library(
         'profile_common.cpp',
         'profile.idl',
         '$BUILD_DIR/mongo/db/profile_filter_impl.cpp',
-        'set_profiling_filter_globally_cmd.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/commands',
@@ -674,8 +668,8 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
+        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/matcher/expressions',
-        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/server_options_core',
     ],
 )
@@ -742,7 +736,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/commands/servers',
         '$BUILD_DIR/mongo/db/db_raii',
-        '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/index/index_access_methods',
         '$BUILD_DIR/mongo/db/pipeline/process_interface/mongo_process_interface',
         '$BUILD_DIR/mongo/db/pipeline/process_interface/mongod_process_interface_factory',
         '$BUILD_DIR/mongo/db/query/map_reduce_output_format',
@@ -815,7 +809,6 @@ env.CppUnitTest(
         "$BUILD_DIR/mongo/db/auth/authmocks",
         "$BUILD_DIR/mongo/db/catalog/collection",
         "$BUILD_DIR/mongo/db/commands/list_collections_filter",
-        "$BUILD_DIR/mongo/db/concurrency/exception_util",
         "$BUILD_DIR/mongo/db/dbdirectclient",
         "$BUILD_DIR/mongo/db/fle_crud",
         "$BUILD_DIR/mongo/db/fle_mocks",
diff --git a/src/mongo/db/commands/apply_ops_cmd.cpp b/src/mongo/db/commands/apply_ops_cmd.cpp
index 1a31cb0d8f8..2b379ee085e 100644
--- a/src/mongo/db/commands/apply_ops_cmd.cpp
+++ b/src/mongo/db/commands/apply_ops_cmd.cpp
@@ -39,6 +39,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/oplog_application_checks.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/jsobj.h"
diff --git a/src/mongo/db/commands/authentication_commands.cpp b/src/mongo/db/commands/authentication_commands.cpp
index 81a844dacba..0fef67553d0 100644
--- a/src/mongo/db/commands/authentication_commands.cpp
+++ b/src/mongo/db/commands/authentication_commands.cpp
@@ -240,7 +240,7 @@ void _authenticateX509(OperationContext* opCtx, AuthenticationSession* session)
     auto user = [&] {
         if (session->getUserName().empty()) {
             auto user = UserName(clientName.toString(), session->getDatabase().toString());
-            session->updateUserName(user, true /* isMechX509 */);
+            session->updateUserName(user);
             return user;
         } else {
             uassert(ErrorCodes::AuthenticationFailed,
@@ -258,6 +258,10 @@ void _authenticateX509(OperationContext* opCtx, AuthenticationSession* session)
 
     auto sslConfiguration = opCtx->getClient()->session()->getSSLConfiguration();
 
+    uassert(ErrorCodes::AuthenticationFailed,
+            "Unable to verify x.509 certificate, as no CA has been provided.",
+            sslConfiguration->hasCA);
+
     uassert(ErrorCodes::ProtocolError,
             "X.509 authentication must always use the $external database.",
             user.getDB() == kExternalDB);
@@ -337,9 +341,9 @@ AuthenticateReply authCommand(OperationContext* opCtx,
         // Allows authenticating as the internal user against the admin database.  This is to
         // support the auth passthrough test framework on mongos (since you can't use the local
         // database on a mongos, so you can't auth as the internal user without this).
-        session->updateUserName(internalSecurityUser, mechanism == auth::kMechanismMongoX509);
+        session->updateUserName(internalSecurityUser);
     } else {
-        session->updateUserName(UserName{user, dbname}, mechanism == auth::kMechanismMongoX509);
+        session->updateUserName(UserName{user, dbname});
     }
 
     if (mechanism.empty()) {
diff --git a/src/mongo/db/commands/authentication_commands.h b/src/mongo/db/commands/authentication_commands.h
index 814a2ab17d2..2e82168e050 100644
--- a/src/mongo/db/commands/authentication_commands.h
+++ b/src/mongo/db/commands/authentication_commands.h
@@ -42,6 +42,6 @@ constexpr StringData kX509AuthMechanism = "MONGODB-X509"_sd;
 void disableX509Auth(ServiceContext* svcCtx);
 bool isX509AuthDisabled(ServiceContext* svcCtx);
 
-void doSpeculativeAuthenticate(OperationContext* opCtx, BSONObj helloCmd, BSONObjBuilder* result);
+void doSpeculativeAuthenticate(OperationContext* opCtx, BSONObj isMaster, BSONObjBuilder* result);
 
 }  // namespace mongo
diff --git a/src/mongo/db/commands/command_mirroring_test.cpp b/src/mongo/db/commands/command_mirroring_test.cpp
index 8222cc9b74e..286d20813a2 100644
--- a/src/mongo/db/commands/command_mirroring_test.cpp
+++ b/src/mongo/db/commands/command_mirroring_test.cpp
@@ -38,7 +38,6 @@
 #include "mongo/unittest/unittest.h"
 
 namespace mongo {
-
 namespace {
 
 class CommandMirroringTest : public unittest::Test {
@@ -109,7 +108,6 @@ public:
     void setUp() override {
         CommandMirroringTest::setUp();
         shardVersion = boost::none;
-        databaseVersion = boost::none;
     }
 
     std::string commandName() override {
@@ -121,10 +119,6 @@ public:
         if (shardVersion) {
             args.push_back(shardVersion.get());
         }
-        if (databaseVersion) {
-            args.push_back(databaseVersion.value());
-        }
-
         auto request = CommandMirroringTest::makeCommand(coll, args);
 
         // Directly add `updates` to `OpMsg::sequences` to emulate `OpMsg::parse()` behavior.
@@ -140,7 +134,6 @@ public:
     }
 
     boost::optional<BSONObj> shardVersion;
-    boost::optional<BSONObj> databaseVersion;
 };
 
 TEST_F(UpdateCommandTest, NoQuery) {
@@ -199,26 +192,21 @@ TEST_F(UpdateCommandTest, MultipleQueries) {
     ASSERT_EQ(mirroredObj["batchSize"].Int(), 1);
 }
 
-TEST_F(UpdateCommandTest, ValidateShardVersionAndDatabaseVersion) {
+TEST_F(UpdateCommandTest, ValidateShardVersion) {
     auto update = BSON("q" << BSONObj() << "u" << BSON("$set" << BSON("_id" << 1)));
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, {update});
 
         ASSERT_FALSE(mirroredObj.hasField("shardVersion"));
-        ASSERT_FALSE(mirroredObj.hasField("databaseVersion"));
     }
 
     const auto kShardVersion = 123;
-    const auto kDatabaseVersion = 456;
     shardVersion = BSON("shardVersion" << kShardVersion);
-    databaseVersion = BSON("databaseVersion" << kDatabaseVersion);
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, {update});
 
         ASSERT_TRUE(mirroredObj.hasField("shardVersion"));
-        ASSERT_TRUE(mirroredObj.hasField("databaseVersion"));
         ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion);
-        ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion);
     }
 }
 
@@ -240,8 +228,7 @@ public:
                 "max",
                 "batchSize",
                 "singleBatch",
-                "shardVersion",
-                "databaseVersion"};
+                "shardVersion"};
     }
 
     void checkFieldNamesAreAllowed(BSONObj& mirroredObj) {
@@ -276,8 +263,7 @@ TEST_F(FindCommandTest, MirrorableKeys) {
                      BSON("awaitData" << true),
                      BSON("allowPartialResults" << true),
                      BSON("collation" << BSONObj()),
-                     BSON("shardVersion" << BSONObj()),
-                     BSON("databaseVersion" << BSONObj())};
+                     BSON("shardVersion" << BSONObj())};
 
     auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs);
     checkFieldNamesAreAllowed(mirroredObj);
@@ -305,8 +291,7 @@ TEST_F(FindCommandTest, ValidateMirroredQuery) {
     const auto min = BSONObj();
     const auto max = BSONObj();
 
-    const auto shardVersion = BSON("v" << 123);
-    const auto databaseVersion = BSON("v" << 456);
+    const auto shardVersion = BSONObj();
 
     auto findArgs = {BSON("filter" << filter),
                      BSON("skip" << skip),
@@ -316,8 +301,7 @@ TEST_F(FindCommandTest, ValidateMirroredQuery) {
                      BSON("collation" << collation),
                      BSON("min" << min),
                      BSON("max" << max),
-                     BSON("shardVersion" << shardVersion),
-                     BSON("databaseVersion" << databaseVersion)};
+                     BSON("shardVersion" << shardVersion)};
 
     auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs);
 
@@ -331,27 +315,21 @@ TEST_F(FindCommandTest, ValidateMirroredQuery) {
     ASSERT(compareBSONObjs(mirroredObj["min"].Obj(), min));
     ASSERT(compareBSONObjs(mirroredObj["max"].Obj(), max));
     ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion));
-    ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion));
 }
 
-TEST_F(FindCommandTest, ValidateShardVersionAndDatabaseVersion) {
+TEST_F(FindCommandTest, ValidateShardVersion) {
     std::vector<BSONObj> findArgs = {BSON("filter" << BSONObj())};
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs);
         ASSERT_FALSE(mirroredObj.hasField("shardVersion"));
-        ASSERT_FALSE(mirroredObj.hasField("databaseVersion"));
     }
 
     const auto kShardVersion = 123;
-    const auto kDatabaseVersion = 456;
     findArgs.push_back(BSON("shardVersion" << kShardVersion));
-    findArgs.push_back(BSON("databaseVersion" << kDatabaseVersion));
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, findArgs);
         ASSERT_TRUE(mirroredObj.hasField("shardVersion"));
-        ASSERT_TRUE(mirroredObj.hasField("databaseVersion"));
         ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion);
-        ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion);
     }
 }
 
@@ -362,14 +340,7 @@ public:
     }
 
     std::vector<std::string> getAllowedKeys() const override {
-        return {"sort",
-                "collation",
-                "find",
-                "filter",
-                "batchSize",
-                "singleBatch",
-                "shardVersion",
-                "databaseVersion"};
+        return {"sort", "collation", "find", "filter", "batchSize", "singleBatch", "shardVersion"};
     }
 };
 
@@ -385,9 +356,7 @@ TEST_F(FindAndModifyCommandTest, MirrorableKeys) {
                               BSON("writeConcern" << BSONObj()),
                               BSON("maxTimeMS" << 100),
                               BSON("collation" << BSONObj()),
-                              BSON("arrayFilters" << BSONArray()),
-                              BSON("shardVersion" << 123),
-                              BSON("databaseVersion" << 456)};
+                              BSON("arrayFilters" << BSONArray())};
 
     auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs);
     checkFieldNamesAreAllowed(mirroredObj);
@@ -412,16 +381,12 @@ TEST_F(FindAndModifyCommandTest, ValidateMirroredQuery) {
     constexpr auto upsert = true;
     const auto collation = BSON("locale"
                                 << "\"fr\"");
-    const auto shardVersion = BSON("v" << 123);
-    const auto databaseVersion = BSON("v" << 456);
 
     auto findAndModifyArgs = {BSON("query" << query),
                               BSON("sort" << sortObj),
                               BSON("update" << update),
                               BSON("upsert" << upsert),
-                              BSON("collation" << collation),
-                              BSON("shardVersion" << shardVersion),
-                              BSON("databaseVersion" << databaseVersion)};
+                              BSON("collation" << collation)};
 
     auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs);
 
@@ -430,11 +395,9 @@ TEST_F(FindAndModifyCommandTest, ValidateMirroredQuery) {
     ASSERT(compareBSONObjs(mirroredObj["filter"].Obj(), query));
     ASSERT(compareBSONObjs(mirroredObj["sort"].Obj(), sortObj));
     ASSERT(compareBSONObjs(mirroredObj["collation"].Obj(), collation));
-    ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion));
-    ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion));
 }
 
-TEST_F(FindAndModifyCommandTest, ValidateShardVersionAndDatabaseVersion) {
+TEST_F(FindAndModifyCommandTest, ValidateShardVersion) {
     std::vector<BSONObj> findAndModifyArgs = {BSON("query" << BSON("name"
                                                                    << "Andy")),
                                               BSON("update" << BSON("$inc" << BSON("score" << 1)))};
@@ -442,19 +405,14 @@ TEST_F(FindAndModifyCommandTest, ValidateShardVersionAndDatabaseVersion) {
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs);
         ASSERT_FALSE(mirroredObj.hasField("shardVersion"));
-        ASSERT_FALSE(mirroredObj.hasField("databaseVersion"));
     }
 
     const auto kShardVersion = 123;
-    const auto kDatabaseVersion = 456;
     findAndModifyArgs.push_back(BSON("shardVersion" << kShardVersion));
-    findAndModifyArgs.push_back(BSON("databaseVersion" << kDatabaseVersion));
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, findAndModifyArgs);
         ASSERT_TRUE(mirroredObj.hasField("shardVersion"));
-        ASSERT_TRUE(mirroredObj.hasField("databaseVersion"));
         ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion);
-        ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion);
     }
 }
 
@@ -465,7 +423,7 @@ public:
     }
 
     std::vector<std::string> getAllowedKeys() const override {
-        return {"distinct", "key", "query", "collation", "shardVersion", "databaseVersion"};
+        return {"distinct", "key", "query", "collation", "shardVersion"};
     }
 };
 
@@ -475,8 +433,7 @@ TEST_F(DistinctCommandTest, MirrorableKeys) {
                          BSON("query" << BSONObj()),
                          BSON("readConcern" << BSONObj()),
                          BSON("collation" << BSONObj()),
-                         BSON("shardVersion" << BSONObj()),
-                         BSON("databaseVersion" << BSONObj())};
+                         BSON("shardVersion" << BSONObj())};
 
     auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs);
     checkFieldNamesAreAllowed(mirroredObj);
@@ -489,15 +446,13 @@ TEST_F(DistinctCommandTest, ValidateMirroredQuery) {
     const auto readConcern = BSON("level"
                                   << "majority");
     const auto collation = BSON("strength" << 1);
-    const auto shardVersion = BSON("v" << 123);
-    const auto databaseVersion = BSON("v" << 456);
+    const auto shardVersion = BSONObj();
 
     auto distinctArgs = {BSON("key" << key),
                          BSON("query" << query),
                          BSON("readConcern" << readConcern),
                          BSON("collation" << collation),
-                         BSON("shardVersion" << shardVersion),
-                         BSON("databaseVersion" << databaseVersion)};
+                         BSON("shardVersion" << shardVersion)};
 
     auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs);
 
@@ -507,27 +462,23 @@ TEST_F(DistinctCommandTest, ValidateMirroredQuery) {
     ASSERT(compareBSONObjs(mirroredObj["query"].Obj(), query));
     ASSERT(compareBSONObjs(mirroredObj["collation"].Obj(), collation));
     ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion));
-    ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion));
 }
 
-TEST_F(DistinctCommandTest, ValidateShardVersionAndDatabaseVersion) {
+TEST_F(DistinctCommandTest, ValidateShardVersion) {
+    const auto kCollection = "test";
+
     std::vector<BSONObj> distinctArgs = {BSON("distinct" << BSONObj())};
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs);
         ASSERT_FALSE(mirroredObj.hasField("shardVersion"));
-        ASSERT_FALSE(mirroredObj.hasField("databaseVersion"));
     }
 
     const auto kShardVersion = 123;
-    const auto kDatabaseVersion = 456;
     distinctArgs.push_back(BSON("shardVersion" << kShardVersion));
-    distinctArgs.push_back(BSON("databaseVersion" << kDatabaseVersion));
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, distinctArgs);
         ASSERT_TRUE(mirroredObj.hasField("shardVersion"));
-        ASSERT_TRUE(mirroredObj.hasField("databaseVersion"));
         ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion);
-        ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion);
     }
 }
 
@@ -538,14 +489,7 @@ public:
     }
 
     std::vector<std::string> getAllowedKeys() const override {
-        return {"count",
-                "query",
-                "skip",
-                "limit",
-                "hint",
-                "collation",
-                "shardVersion",
-                "databaseVersion"};
+        return {"count", "query", "skip", "limit", "hint", "collation", "shardVersion"};
     }
 };
 
@@ -556,8 +500,7 @@ TEST_F(CountCommandTest, MirrorableKeys) {
                       BSON("hint" << BSONObj()),
                       BSON("readConcern" << BSONObj()),
                       BSON("collation" << BSONObj()),
-                      BSON("shardVersion" << BSONObj()),
-                      BSON("databaseVersion" << BSONObj())};
+                      BSON("shardVersion" << BSONObj())};
 
     auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs);
     checkFieldNamesAreAllowed(mirroredObj);
@@ -568,14 +511,12 @@ TEST_F(CountCommandTest, ValidateMirroredQuery) {
                             << "Delivered");
     const auto hint = BSON("status" << 1);
     constexpr auto limit = 1000;
-    const auto shardVersion = BSON("v" << 123);
-    const auto databaseVersion = BSON("v" << 456);
+    const auto shardVersion = BSONObj();
 
     auto countArgs = {BSON("query" << query),
                       BSON("hint" << hint),
                       BSON("limit" << limit),
-                      BSON("shardVersion" << shardVersion),
-                      BSON("databaseVersion" << databaseVersion)};
+                      BSON("shardVersion" << shardVersion)};
     auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs);
 
     ASSERT_EQ(mirroredObj["count"].String(), kCollection);
@@ -585,27 +526,21 @@ TEST_F(CountCommandTest, ValidateMirroredQuery) {
     ASSERT(compareBSONObjs(mirroredObj["hint"].Obj(), hint));
     ASSERT_EQ(mirroredObj["limit"].Int(), limit);
     ASSERT(compareBSONObjs(mirroredObj["shardVersion"].Obj(), shardVersion));
-    ASSERT(compareBSONObjs(mirroredObj["databaseVersion"].Obj(), databaseVersion));
 }
 
-TEST_F(CountCommandTest, ValidateShardVersionAndDatabaseVersion) {
+TEST_F(CountCommandTest, ValidateShardVersion) {
     std::vector<BSONObj> countArgs = {BSON("count" << BSONObj())};
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs);
         ASSERT_FALSE(mirroredObj.hasField("shardVersion"));
-        ASSERT_FALSE(mirroredObj.hasField("databaseVersion"));
     }
 
     const auto kShardVersion = 123;
-    const auto kDatabaseVersion = 456;
     countArgs.push_back(BSON("shardVersion" << kShardVersion));
-    countArgs.push_back(BSON("databaseVersion" << kDatabaseVersion));
     {
         auto mirroredObj = createCommandAndGetMirrored(kCollection, countArgs);
         ASSERT_TRUE(mirroredObj.hasField("shardVersion"));
-        ASSERT_TRUE(mirroredObj.hasField("databaseVersion"));
         ASSERT_EQ(mirroredObj["shardVersion"].Int(), kShardVersion);
-        ASSERT_EQ(mirroredObj["databaseVersion"].Int(), kDatabaseVersion);
     }
 }
 
diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp
index 26ee8ab5d4b..6aa009ea216 100644
--- a/src/mongo/db/commands/compact.cpp
+++ b/src/mongo/db/commands/compact.cpp
@@ -95,9 +95,6 @@ public:
             return false;
         }
 
-        // This command is internal to the storage engine and should not block oplog application.
-        ShouldNotConflictWithSecondaryBatchApplicationBlock noPBWMBlock(opCtx->lockState());
-
         StatusWith<int64_t> status = compactCollection(opCtx, nss);
         uassertStatusOK(status.getStatus());
 
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index 57291c7b782..c548bf3ef1c 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -110,7 +110,7 @@ public:
                 Status::OK()};
     }
 
-    bool shouldAffectReadOptionCounters() const override {
+    bool shouldAffectReadConcernCounter() const override {
         return true;
     }
 
@@ -182,8 +182,12 @@ public:
 
             // An empty PrivilegeVector is acceptable because these privileges are only checked on
             // getMore and explain will not open a cursor.
-            return runAggregate(
-                opCtx, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result);
+            return runAggregate(opCtx,
+                                viewAggRequest.getNamespace(),
+                                viewAggRequest,
+                                viewAggregation.getValue(),
+                                PrivilegeVector(),
+                                result);
         }
 
         const auto& collection = ctx->getCollection();
@@ -232,8 +236,6 @@ public:
             &hangBeforeCollectionCount, opCtx, "hangBeforeCollectionCount", []() {}, nss);
 
         auto request = CountCommandRequest::parse(IDLParserErrorContext("count"), cmdObj);
-        auto curOp = CurOp::get(opCtx);
-        curOp->beginQueryPlanningTimer();
         if (shouldDoFLERewrite(request)) {
             processFLECountD(opCtx, nss, &request);
         }
@@ -283,6 +285,7 @@ public:
         auto exec = std::move(statusWithPlanExecutor.getValue());
 
         // Store the plan summary string in CurOp.
+        auto curOp = CurOp::get(opCtx);
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
             curOp->setPlanSummary_inlock(exec->getPlanExplainer().getPlanSummary());
@@ -319,7 +322,6 @@ public:
             keyBob.append("hint", 1);
             keyBob.append("collation", 1);
             keyBob.append("shardVersion", 1);
-            keyBob.append("databaseVersion", 1);
 
             return keyBob.obj();
         }();
diff --git a/src/mongo/db/commands/cqf/cqf_aggregate.cpp b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
index 30cdc8c3846..100da0be582 100644
--- a/src/mongo/db/commands/cqf/cqf_aggregate.cpp
+++ b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
@@ -29,7 +29,6 @@
 
 #include "mongo/db/commands/cqf/cqf_aggregate.h"
 
-#include "mongo/db/curop.h"
 #include "mongo/db/exec/sbe/abt/abt_lower.h"
 #include "mongo/db/pipeline/abt/abt_document_source_visitor.h"
 #include "mongo/db/pipeline/abt/match_expression_visitor.h"
@@ -63,8 +62,7 @@ static opt::unordered_map<std::string, optimizer::IndexDefinition> buildIndexSpe
 
     const IndexCatalog& indexCatalog = *collection->getIndexCatalog();
     opt::unordered_map<std::string, IndexDefinition> result;
-    auto indexIterator =
-        indexCatalog.getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
+    auto indexIterator = indexCatalog.getIndexIterator(opCtx, false /*includeUnfinished*/);
 
     while (indexIterator->more()) {
         const IndexCatalogEntry& catalogEntry = *indexIterator->next();
@@ -257,8 +255,7 @@ static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> optimizeAndCreateExe
     }
 
     auto yieldPolicy =
-        std::make_unique<PlanYieldPolicySBE>(opCtx,
-                                             PlanYieldPolicy::YieldPolicy::YIELD_AUTO,
+        std::make_unique<PlanYieldPolicySBE>(PlanYieldPolicy::YieldPolicy::YIELD_AUTO,
                                              opCtx->getServiceContext()->getFastClockSource(),
                                              internalQueryExecYieldIterations.load(),
                                              Milliseconds{internalQueryExecYieldPeriodMS.load()},
@@ -338,9 +335,6 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> getSBEExecutorViaCascadesOp
     const std::string collNameStr = nss.coll().toString();
     const std::string scanDefName = collNameStr + "_" + uuidStr;
 
-    auto curOp = CurOp::get(opCtx);
-    curOp->debug().cqfUsed = true;
-
     QueryHints queryHints = getHintsFromQueryKnobs();
 
     PrefixId prefixId;
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index f6684487a21..c743a405714 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -50,7 +50,7 @@
 #include "mongo/db/catalog/uncommitted_catalog_updates.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/create_indexes_gen.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -321,12 +321,11 @@ bool indexesAlreadyExist(OperationContext* opCtx,
  * Checks database sharding state. Throws exception on error.
  */
 void checkDatabaseShardingState(OperationContext* opCtx, const NamespaceString& ns) {
-    Lock::CollectionLock collLock(opCtx, ns, MODE_IS);
-
     auto dss = DatabaseShardingState::get(opCtx, ns.db());
     auto dssLock = DatabaseShardingState::DSSLock::lockShared(opCtx, dss);
     dss->checkDbVersion(opCtx, dssLock);
 
+    Lock::CollectionLock collLock(opCtx, ns, MODE_IS);
     try {
         const auto collDesc =
             CollectionShardingState::get(opCtx, ns)->getCollectionDescription(opCtx);
@@ -448,7 +447,7 @@ CreateIndexesReply runCreateIndexesOnNewCollection(
 }
 
 bool isCreatingInternalConfigTxnsPartialIndex(const CreateIndexesCommand& cmd) {
-    if (cmd.getIndexes().size() != 1) {
+    if (cmd.getIndexes().size() > 1) {
         return false;
     }
     const auto& index = cmd.getIndexes()[0];
diff --git a/src/mongo/db/commands/current_op.cpp b/src/mongo/db/commands/current_op.cpp
index b4a16bc40cb..78fec805202 100644
--- a/src/mongo/db/commands/current_op.cpp
+++ b/src/mongo/db/commands/current_op.cpp
@@ -77,7 +77,12 @@ public:
             privileges = {Privilege(ResourcePattern::forClusterResource(), ActionType::inprog)};
         }
 
-        auto status = runAggregate(opCtx, request, std::move(aggCmdObj), privileges, &replyBuilder);
+        auto status = runAggregate(opCtx,
+                                   request.getNamespace(),
+                                   request,
+                                   std::move(aggCmdObj),
+                                   privileges,
+                                   &replyBuilder);
 
         if (!status.isOK()) {
             return status;
diff --git a/src/mongo/db/commands/dbcheck.cpp b/src/mongo/db/commands/dbcheck.cpp
index 5d4579691dd..c26b09d8831 100644
--- a/src/mongo/db/commands/dbcheck.cpp
+++ b/src/mongo/db/commands/dbcheck.cpp
@@ -36,10 +36,10 @@
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/database.h"
-#include "mongo/db/catalog/health_log_interface.h"
+#include "mongo/db/catalog/health_log.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/test_commands_enabled.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/jsobj.h"
 #include "mongo/db/operation_context.h"
@@ -52,10 +52,6 @@
 
 #include "mongo/logv2/log.h"
 
-MONGO_FAIL_POINT_DEFINE(sleepAfterExtraIndexKeysHashing);
-MONGO_FAIL_POINT_DEFINE(hangBeforeProcessingDbCheckRun);
-MONGO_FAIL_POINT_DEFINE(hangBeforeAddingDBCheckBatchToOplog);
-
 namespace mongo {
 
 namespace {
@@ -97,7 +93,7 @@ public:
                                                               OplogEntriesEnum::Start,
                                                               boost::none /*data*/
             );
-            HealthLogInterface::get(_opCtx->getServiceContext())->log(*healthLogEntry);
+            HealthLog::get(_opCtx->getServiceContext()).log(*healthLogEntry);
 
             DbCheckOplogStartStop oplogEntry;
             const auto nss = NamespaceString("admin.$cmd");
@@ -123,7 +119,7 @@ public:
                                                               OplogEntriesEnum::Stop,
                                                               boost::none /*data*/
             );
-            HealthLogInterface::get(_opCtx->getServiceContext())->log(*healthLogEntry);
+            HealthLog::get(_opCtx->getServiceContext()).log(*healthLogEntry);
         } catch (const DBException&) {
             LOGV2(6202201, "Could not log stop event");
         }
@@ -290,18 +286,13 @@ protected:
 
         DbCheckStartAndStopLogger startStop(opCtx);
 
-        if (MONGO_unlikely(hangBeforeProcessingDbCheckRun.shouldFail())) {
-            LOGV2(7949000, "Hanging dbcheck due to failpoint 'hangBeforeProcessingDbCheckRun'");
-            hangBeforeProcessingDbCheckRun.pauseWhileSet();
-        }
-
         for (const auto& coll : *_run) {
             try {
                 _doCollection(opCtx, coll);
             } catch (const DBException& e) {
                 auto logEntry = dbCheckErrorHealthLogEntry(
                     coll.nss, "dbCheck failed", OplogEntriesEnum::Batch, e.toStatus());
-                HealthLogInterface::get(Client::getCurrent()->getServiceContext())->log(*logEntry);
+                HealthLog::get(Client::getCurrent()->getServiceContext()).log(*logEntry);
                 return;
             }
 
@@ -332,7 +323,7 @@ private:
                     "abandoning dbCheck batch because collection no longer exists",
                     OplogEntriesEnum::Batch,
                     Status(ErrorCodes::NamespaceNotFound, "collection not found"));
-                HealthLogInterface::get(Client::getCurrent()->getServiceContext())->log(*entry);
+                HealthLog::get(Client::getCurrent()->getServiceContext()).log(*entry);
                 return;
             }
         }
@@ -410,7 +401,7 @@ private:
                                                        OplogEntriesEnum::Batch,
                                                        result.getStatus());
                 }
-                HealthLogInterface::get(opCtx)->log(*entry);
+                HealthLog::get(opCtx).log(*entry);
                 if (retryable) {
                     continue;
                 }
@@ -433,28 +424,17 @@ private:
                 (_batchesProcessed % gDbCheckHealthLogEveryNBatches.load() == 0)) {
                 // On debug builds, health-log every batch result; on release builds, health-log
                 // every N batches.
-                HealthLogInterface::get(opCtx)->log(*entry);
-            }
-
-            if (MONGO_unlikely(sleepAfterExtraIndexKeysHashing.shouldFail())) {
-                LOGV2_DEBUG(
-                    3083201,
-                    3,
-                    "Sleeping for 1 second due to sleepAfterExtraIndexKeysHashing failpoint");
-                opCtx->sleepFor(Milliseconds(1000));
+                HealthLog::get(opCtx).log(*entry);
             }
 
             WriteConcernResult unused;
             auto status = waitForWriteConcern(opCtx, stats.time, info.writeConcern, &unused);
             if (!status.isOK()) {
-                // TODO SERVER-89817: Add context with batch ID and lastKey once those are
-                // backported.
-                auto entry = dbCheckErrorHealthLogEntry(info.nss,
-                                                        "dbCheck failed waiting for writeConcern",
-                                                        OplogEntriesEnum::Batch,
-                                                        status);
-                HealthLogInterface::get(opCtx)->log(*entry);
-                return;
+                auto entry = dbCheckWarningHealthLogEntry(info.nss,
+                                                          "dbCheck failed waiting for writeConcern",
+                                                          OplogEntriesEnum::Batch,
+                                                          status);
+                HealthLog::get(opCtx).log(*entry);
             }
 
             start = stats.lastKey;
@@ -585,12 +565,6 @@ private:
                 batch.setMaxKey(BSONKey(hasher->lastKey()));
                 batch.setReadTimestamp(readTimestamp);
 
-                if (MONGO_unlikely(hangBeforeAddingDBCheckBatchToOplog.shouldFail())) {
-                    LOGV2(8589000,
-                          "Hanging dbCheck due to failpoint 'hangBeforeAddingDBCheckBatchToOplog'");
-                    hangBeforeAddingDBCheckBatchToOplog.pauseWhileSet();
-                }
-
                 // Send information on this batch over the oplog.
                 result.time = _logOp(opCtx, info.nss, collection->uuid(), batch.toBSON());
                 result.readTimestamp = readTimestamp;
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 54574139484..0e88f993f93 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -58,6 +58,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/commands/server_status.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
@@ -168,6 +169,41 @@ public:
     };
 } cmdDropDatabase;
 
+static const char* repairRemovedMessage =
+    "This command has been removed. If you would like to compact your data, use the 'compact' "
+    "command. If you would like to rebuild indexes, use the 'reIndex' command. If you need to "
+    "recover data, please see the documentation for repairing your database offline: "
+    "http://dochub.mongodb.org/core/repair";
+
+class CmdRepairDatabase : public ErrmsgCommandDeprecated {
+public:
+    AllowedOnSecondary secondaryAllowed(ServiceContext*) const override {
+        return AllowedOnSecondary::kAlways;
+    }
+    virtual bool maintenanceMode() const {
+        return false;
+    }
+
+    std::string help() const override {
+        return repairRemovedMessage;
+    }
+    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+        return false;
+    }
+
+    CmdRepairDatabase() : ErrmsgCommandDeprecated("repairDatabase") {}
+
+    bool errmsgRun(OperationContext* opCtx,
+                   const std::string& dbname,
+                   const BSONObj& cmdObj,
+                   std::string& errmsg,
+                   BSONObjBuilder& result) {
+
+        uasserted(ErrorCodes::CommandNotFound, repairRemovedMessage);
+        return false;
+    }
+} cmdRepairDatabase;
+
 /* drop collection */
 class CmdDrop : public DropCmdVersion1Gen<CmdDrop> {
 public:
@@ -294,19 +330,13 @@ public:
         bool estimate = jsobj["estimate"].trueValue();
 
         const NamespaceString nss(ns);
-        AutoGetCollectionForReadCommand autoColl(opCtx, nss);
-        const auto& collection = autoColl.getCollection();
+        AutoGetCollectionForReadCommand collection(opCtx, nss);
 
-        if (!collection) {
-            // Collection does not exist
-            result.appendNumber("size", 0);
-            result.appendNumber("numObjects", 0);
-            result.append("millis", timer.millis());
-            return true;
-        }
+        const auto collDesc =
+            CollectionShardingState::get(opCtx, nss)->getCollectionDescription(opCtx);
 
-        if (collection.isSharded()) {
-            const ShardKeyPattern shardKeyPattern(collection.getShardKeyPattern());
+        if (collDesc.isSharded()) {
+            const ShardKeyPattern shardKeyPattern(collDesc.getKeyPattern());
             uassert(ErrorCodes::BadValue,
                     "keyPattern must be empty or must be an object that equals the shard key",
                     keyPattern.isEmpty() ||
@@ -324,7 +354,10 @@ public:
             max = shardKeyPattern.normalizeShardKey(max);
         }
 
-        const long long numRecords = collection->numRecords(opCtx);
+        long long numRecords = 0;
+        if (collection) {
+            numRecords = collection->numRecords(opCtx);
+        }
 
         if (numRecords == 0) {
             result.appendNumber("size", 0);
@@ -344,7 +377,7 @@ public:
                 return 1;
             }
             exec = InternalPlanner::collectionScan(
-                opCtx, &collection, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+                opCtx, &collection.getCollection(), PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY);
         } else if (min.isEmpty() || max.isEmpty()) {
             errmsg = "only one of min or max specified";
             return false;
@@ -355,7 +388,7 @@ public:
             }
 
             auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
-                                                         collection,
+                                                         *collection,
                                                          collection->getIndexCatalog(),
                                                          keyPattern,
                                                          /*requireSingleKey=*/true);
@@ -370,12 +403,12 @@ public:
             max = Helpers::toKeyFormat(kp.extendRangeBound(max, false));
 
             exec = InternalPlanner::shardKeyIndexScan(opCtx,
-                                                      &collection,
+                                                      &collection.getCollection(),
                                                       *shardKeyIdx,
                                                       min,
                                                       max,
                                                       BoundInclusion::kIncludeStartKeyOnly,
-                                                      PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+                                                      PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY);
         }
 
         CurOpFailpointHelpers::waitWhileFailPointEnabled(
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index 339d442bcaf..41969950a43 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -57,8 +57,7 @@
 #include "mongo/db/commands/profile_common.h"
 #include "mongo/db/commands/profile_gen.h"
 #include "mongo/db/commands/server_status.h"
-#include "mongo/db/commands/set_profiling_filter_globally_cmd.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
@@ -205,8 +204,6 @@ protected:
 
 } cmdProfile;
 
-SetProfilingFilterGloballyCmd cmdSetProfilingFilterGlobally;
-
 class CmdFileMD5 : public BasicCommand {
 public:
     CmdFileMD5() : BasicCommand("filemd5") {}
@@ -382,7 +379,7 @@ public:
                     try {
                         // RELOCKED
                         ctx.reset(new AutoGetCollectionForReadCommand(opCtx, nss));
-                    } catch (const ExceptionFor<ErrorCodes::StaleConfig>&) {
+                    } catch (const StaleConfigException&) {
                         LOGV2_DEBUG(
                             20453,
                             1,
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index ef08e46aee7..57585a88401 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -100,7 +100,7 @@ public:
         return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
     }
 
-    bool shouldAffectReadOptionCounters() const override {
+    bool shouldAffectReadConcernCounter() const override {
         return true;
     }
 
@@ -172,7 +172,7 @@ public:
             // An empty PrivilegeVector is acceptable because these privileges are only checked on
             // getMore and explain will not open a cursor.
             return runAggregate(
-                opCtx, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result);
+                opCtx, nss, viewAggRequest, viewAggregation.getValue(), PrivilegeVector(), result);
         }
 
         const auto& collection = ctx->getCollection();
@@ -296,6 +296,8 @@ public:
             auto&& [stats, _] =
                 explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
             LOGV2_WARNING(23797,
+                          "Plan executor error during distinct command: {error}, "
+                          "stats: {stats}, cmd: {cmd}",
                           "Plan executor error during distinct command",
                           "error"_attr = exception.toStatus(),
                           "stats"_attr = redact(stats),
@@ -346,7 +348,6 @@ public:
             keyBob.append("query", 1);
             keyBob.append("collation", 1);
             keyBob.append("shardVersion", 1);
-            keyBob.append("databaseVersion", 1);
             return keyBob.obj();
         }();
 
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index a2c773af89d..f4595688706 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -44,7 +44,7 @@
 #include "mongo/db/catalog/multi_index_block.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/drop_indexes_gen.h"
@@ -236,8 +236,8 @@ public:
             collection.getWritableCollection()->getIndexCatalog()->dropAllIndexes(
                 opCtx, collection.getWritableCollection(), true, {});
 
-            swIndexesToRebuild = indexer->init(
-                opCtx, collection, all, MultiIndexBlock::kNoopOnInitFn, /*forRecovery=*/false);
+            swIndexesToRebuild =
+                indexer->init(opCtx, collection, all, MultiIndexBlock::kNoopOnInitFn);
             uassertStatusOK(swIndexesToRebuild.getStatus());
             wunit.commit();
         });
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index d94346896da..abbc0d834fd 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -41,7 +41,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/update_metrics.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/exec/update_stage.h"
@@ -95,17 +95,16 @@ boost::optional<BSONObj> advanceExecutor(OperationContext* opCtx,
     PlanExecutor::ExecState state;
     try {
         state = exec->getNext(&value, nullptr);
-    } catch (const WriteConflictException&) {
-        // Propagate the WCE to be retried at a higher-level without logging.
-        throw;
     } catch (DBException& exception) {
         auto&& explainer = exec->getPlanExplainer();
         auto&& [stats, _] = explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
-        LOGV2_WARNING(23802,
-                      "Plan executor error during findAndModify",
-                      "error"_attr = exception.toStatus(),
-                      "stats"_attr = redact(stats),
-                      "cmd"_attr = request.toBSON(BSONObj() /* commandPassthroughFields */));
+        LOGV2_WARNING(
+            23802,
+            "Plan executor error during findAndModify: {error}, stats: {stats}, cmd: {cmd}",
+            "Plan executor error during findAndModify",
+            "error"_attr = exception.toStatus(),
+            "stats"_attr = redact(stats),
+            "cmd"_attr = request.toBSON(BSONObj() /* commandPassthroughFields */));
 
         exception.addContext("Plan executor error during findAndModify");
         throw;
@@ -170,8 +169,9 @@ void makeUpdateRequest(OperationContext* opCtx,
     requestOut->setMulti(false);
     requestOut->setExplain(explain);
 
-    requestOut->setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
-    requestOut->setBypassEmptyTsReplacement(request.getBypassEmptyTsReplacement());
+    requestOut->setYieldPolicy(opCtx->inMultiDocumentTransaction()
+                                   ? PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY
+                                   : PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
 }
 
 void makeDeleteRequest(OperationContext* opCtx,
@@ -190,7 +190,9 @@ void makeDeleteRequest(OperationContext* opCtx,
     requestOut->setReturnDeleted(true);  // Always return the old value.
     requestOut->setIsExplain(explain);
 
-    requestOut->setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+    requestOut->setYieldPolicy(opCtx->inMultiDocumentTransaction()
+                                   ? PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY
+                                   : PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
 }
 
 write_ops::FindAndModifyCommandReply buildResponse(const PlanExecutor* exec,
@@ -761,9 +763,6 @@ void CmdFindAndModify::Invocation::appendMirrorableRequest(BSONObjBuilder* bob)
     if (const auto& shardVersion = rawCmd.getField("shardVersion"); !shardVersion.eoo()) {
         bob->append(shardVersion);
     }
-    if (const auto& databaseVersion = rawCmd.getField("databaseVersion"); !databaseVersion.eoo()) {
-        bob->append(databaseVersion);
-    }
 
     // Prevent the find from returning multiple documents since we can
     bob->append("batchSize", 1);
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index df96e09e2ce..f7ac781404f 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -36,7 +36,6 @@
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/client.h"
 #include "mongo/db/clientcursor.h"
-#include "mongo/db/collection_type.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/run_aggregate.h"
 #include "mongo/db/commands/test_commands_enabled.h"
@@ -55,10 +54,6 @@
 #include "mongo/db/query/find_common.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/query/query_knobs_gen.h"
-#include "mongo/db/query/query_shape/query_shape.h"
-#include "mongo/db/query/query_stats/find_key.h"
-#include "mongo/db/query/query_stats/key.h"
-#include "mongo/db/query/query_stats/query_stats.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/counters.h"
@@ -116,28 +111,66 @@ std::unique_ptr<FindCommandRequest> translateNtoReturnToLimitOrBatchSize(
     return findCmd;
 }
 
+// Parses the command object to a FindCommandRequest. If the client request did not specify any
+// runtime constants, make them available to the query here.
+std::unique_ptr<FindCommandRequest> parseCmdObjectToFindCommandRequest(OperationContext* opCtx,
+                                                                       NamespaceString nss,
+                                                                       BSONObj cmdObj) {
+    auto findCommand = query_request_helper::makeFromFindCommand(
+        std::move(cmdObj),
+        std::move(nss),
+        APIParameters::get(opCtx).getAPIStrict().value_or(false));
+
+    // Rewrite any FLE find payloads that exist in the query if this is a FLE 2 query.
+    if (shouldDoFLERewrite(findCommand)) {
+        invariant(findCommand->getNamespaceOrUUID().nss());
+        processFLEFindD(opCtx, findCommand->getNamespaceOrUUID().nss().get(), findCommand.get());
+    }
+
+    return translateNtoReturnToLimitOrBatchSize(std::move(findCommand));
+}
+
 boost::intrusive_ptr<ExpressionContext> makeExpressionContext(
     OperationContext* opCtx,
     const FindCommandRequest& findCommand,
-    const CollectionPtr& collPtr,
     boost::optional<ExplainOptions::Verbosity> verbosity) {
     std::unique_ptr<CollatorInterface> collator;
     if (!findCommand.getCollation().isEmpty()) {
         collator = uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext())
                                        ->makeFromBSON(findCommand.getCollation()));
-    } else if (collPtr && collPtr->getDefaultCollator()) {
-        // The 'collPtr' will be null for views, but we don't need to worry about views here. The
-        // views will get rewritten into aggregate command and will regenerate the
-        // ExpressionContext.
-        collator = collPtr->getDefaultCollator()->clone();
     }
-    auto expCtx =
-        make_intrusive<ExpressionContext>(opCtx,
-                                          findCommand,
-                                          std::move(collator),
-                                          CurOp::get(opCtx)->dbProfileLevel() > 0,  // mayDbProfile
-                                          verbosity,
-                                          allowDiskUseByDefault.load());
+
+    // Although both 'find' and 'aggregate' commands have an ExpressionContext, some of the data
+    // members in the ExpressionContext are used exclusively by the aggregation subsystem. This
+    // includes the following fields which here we simply initialize to some meaningless default
+    // value:
+    //  - explain
+    //  - fromMongos
+    //  - needsMerge
+    //  - bypassDocumentValidation
+    //  - mongoProcessInterface
+    //  - resolvedNamespaces
+    //  - uuid
+    //
+    // As we change the code to make the find and agg systems more tightly coupled, it would make
+    // sense to start initializing these fields for find operations as well.
+    auto expCtx = make_intrusive<ExpressionContext>(
+        opCtx,
+        verbosity,
+        false,  // fromMongos
+        false,  // needsMerge
+        findCommand.getAllowDiskUse().value_or(allowDiskUseByDefault.load()),
+        false,  // bypassDocumentValidation
+        false,  // isMapReduceCommand
+        findCommand.getNamespaceOrUUID().nss().value_or(NamespaceString()),
+        findCommand.getLegacyRuntimeConstants(),
+        std::move(collator),
+        nullptr,  // mongoProcessInterface
+        StringMap<ExpressionContext::ResolvedNamespace>{},
+        boost::none,                             // uuid
+        findCommand.getLet(),                    // let
+        CurOp::get(opCtx)->dbProfileLevel() > 0  // mayDbProfile
+    );
     if (storageGlobalParams.readOnly) {
         // Disallow disk use if in read-only mode.
         expCtx->allowDiskUse = false;
@@ -159,45 +192,6 @@ void beginQueryOp(OperationContext* opCtx, const NamespaceString& nss, const BSO
 }
 
 /**
- * Parses the grammar elements like 'filter', 'sort', and 'projection' from the raw
- * 'FindCommandRequest', and tracks internal state like begining the operation's timer and recording
- * query shape stats (if enabled).
- */
-std::unique_ptr<CanonicalQuery> parseQueryAndBeginOperation(
-    OperationContext* opCtx,
-    const AutoGetCollectionForReadCommandMaybeLockFree& ctx,
-    const NamespaceString& nss,
-    BSONObj requestBody,
-    std::unique_ptr<FindCommandRequest> findCommand,
-    const CollectionPtr& collection) {
-    // Fill out curop information.
-    beginQueryOp(opCtx, nss, requestBody);
-    // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
-    const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
-
-    auto expCtx =
-        makeExpressionContext(opCtx, *findCommand, collection, boost::none /* verbosity */);
-
-    auto parsedRequest = uassertStatusOK(
-        parsed_find_command::parse(expCtx,
-                                   std::move(findCommand),
-                                   extensionsCallback,
-                                   MatchExpressionParser::kAllowAllSpecialFeatures));
-
-    // Register query stats collection. Exclude queries against collections with encrypted fields.
-    // It is important to do this before canonicalizing and optimizing the query, each of which
-    // would alter the query shape.
-    if (!(collection && collection.get()->getCollectionOptions().encryptedFieldConfig)) {
-        query_stats::registerRequest(opCtx, nss, [&]() {
-            return std::make_unique<query_stats::FindKey>(
-                expCtx, *parsedRequest, ctx.getCollectionType());
-        });
-    }
-
-    return uassertStatusOK(
-        CanonicalQuery::canonicalize(std::move(expCtx), std::move(parsedRequest)));
-}
-/**
  * A command for running .find() queries.
  */
 class FindCmd final : public Command {
@@ -250,7 +244,7 @@ public:
         return false;
     }
 
-    bool shouldAffectReadOptionCounters() const override {
+    bool shouldAffectReadConcernCounter() const override {
         return true;
     }
 
@@ -322,20 +316,11 @@ public:
             const auto nss = ctx->getNss();
 
             // Parse the command BSON to a FindCommandRequest.
-            auto findCommand = _parseCmdObjectToFindCommandRequest(opCtx, nss, _request.body);
+            auto findCommand = parseCmdObjectToFindCommandRequest(opCtx, nss, _request.body);
 
             // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
             const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
-
-            // The collection may be NULL. If so, getExecutor() should handle it by returning an
-            // execution tree with an EOFStage.
-            const auto& collection = ctx->getCollection();
-            if (!ctx->getView()) {
-                const bool isClusteredCollection = collection && collection->isClustered();
-                uassertStatusOK(query_request_helper::validateResumeAfter(
-                    findCommand->getResumeAfter(), isClusteredCollection));
-            }
-            auto expCtx = makeExpressionContext(opCtx, *findCommand, collection, verbosity);
+            auto expCtx = makeExpressionContext(opCtx, *findCommand, verbosity);
             const bool isExplain = true;
             auto cq = uassertStatusOK(
                 CanonicalQuery::canonicalize(opCtx,
@@ -372,8 +357,8 @@ public:
                 try {
                     // An empty PrivilegeVector is acceptable because these privileges are only
                     // checked on getMore and explain will not open a cursor.
-                    uassertStatusOK(
-                        runAggregate(opCtx, aggRequest, viewAggCmd, PrivilegeVector(), result));
+                    uassertStatusOK(runAggregate(
+                        opCtx, nss, aggRequest, viewAggCmd, PrivilegeVector(), result));
                 } catch (DBException& error) {
                     if (error.code() == ErrorCodes::InvalidPipelineOperator) {
                         uasserted(ErrorCodes::InvalidPipelineOperator,
@@ -385,6 +370,10 @@ public:
                 return;
             }
 
+            // The collection may be NULL. If so, getExecutor() should handle it by returning an
+            // execution tree with an EOFStage.
+            const auto& collection = ctx->getCollection();
+
             // Get the execution plan for the query.
             bool permitYield = true;
             auto exec =
@@ -419,10 +408,10 @@ public:
             // Parse the command BSON to a FindCommandRequest. Pass in the parsedNss in case cmdObj
             // does not have a UUID.
             auto parsedNss = NamespaceString{CommandHelpers::parseNsFromCommand(_dbName, cmdObj)};
+            const bool isExplain = false;
             const bool isOplogNss = (parsedNss == NamespaceString::kRsOplogNamespace);
             auto findCommand =
-                _parseCmdObjectToFindCommandRequest(opCtx, std::move(parsedNss), cmdObj);
-            CurOp::get(opCtx)->beginQueryPlanningTimer();
+                parseCmdObjectToFindCommandRequest(opCtx, std::move(parsedNss), cmdObj);
 
             // Only allow speculative majority for internal commands that specify the correct flag.
             uassert(ErrorCodes::ReadConcernMajorityNotEnabled,
@@ -516,16 +505,14 @@ public:
             }
 
             // Tailing a replicated capped clustered collection requires majority read concern.
-            const auto& collection = ctx->getCollection();
-
-            bool isClusteredCollection = false;
-            if (collection) {
+            const auto coll = ctx->getCollection().get();
+            if (coll) {
                 const bool isTailable = findCommand->getTailable();
                 const bool isMajorityReadConcern = repl::ReadConcernArgs::get(opCtx).getLevel() ==
                     repl::ReadConcernLevel::kMajorityReadConcern;
-                isClusteredCollection = collection->isClustered();
-                const bool isCapped = collection->isCapped();
-                const bool isReplicated = collection->ns().isReplicated();
+                const bool isClusteredCollection = coll->isClustered();
+                const bool isCapped = coll->isCapped();
+                const bool isReplicated = coll->ns().isReplicated();
                 if (isClusteredCollection && isCapped && isReplicated && isTailable) {
                     uassert(ErrorCodes::Error(6049203),
                             "A tailable cursor on a capped clustered collection requires majority "
@@ -534,16 +521,19 @@ public:
                 }
             }
 
-            // Views use the aggregation system and the $_resumeAfter parameter is not allowed. A
-            // more descriptive error will be raised later, but we want to validate this parameter
-            // before beginning the operation.
-            if (!ctx->getView()) {
-                uassertStatusOK(query_request_helper::validateResumeAfter(
-                    findCommand->getResumeAfter(), isClusteredCollection));
-            }
+            // Fill out curop information.
+            beginQueryOp(opCtx, nss, _request.body);
 
-            auto cq = parseQueryAndBeginOperation(
-                opCtx, *ctx, nss, _request.body, std::move(findCommand), collection);
+            // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
+            const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
+            auto expCtx = makeExpressionContext(opCtx, *findCommand, boost::none /* verbosity */);
+            auto cq = uassertStatusOK(
+                CanonicalQuery::canonicalize(opCtx,
+                                             std::move(findCommand),
+                                             isExplain,
+                                             std::move(expCtx),
+                                             extensionsCallback,
+                                             MatchExpressionParser::kAllowAllSpecialFeatures));
 
             // If we are running a query against a view, or if we are trying to test the new
             // optimizer, redirect this query through the aggregation system.
@@ -560,9 +550,6 @@ public:
                 auto viewAggregationCommand =
                     uassertStatusOK(query_request_helper::asAggregationCommand(findCommand));
 
-                // This doesn't directly call 'runAggregate()' so it doesn't need to adapt to the
-                // new API on v6.0. @Alyssa this suggests we should look into view performance more
-                // carefully on v6.0. The perf of this code path may have different characteristics?
                 BSONObj aggResult = CommandHelpers::runCommandDirectly(
                     opCtx, OpMsgRequest::fromDBAndBody(_dbName, std::move(viewAggregationCommand)));
                 auto status = getStatusFromCommandResult(aggResult);
@@ -583,6 +570,8 @@ public:
             uassertStatusOK(replCoord->checkCanServeReadsFor(
                 opCtx, nss, ReadPreferenceSetting::get(opCtx).canRunOnSecondary()));
 
+            const auto& collection = ctx->getCollection();
+
             if (cq->getFindCommandRequest().getReadOnce()) {
                 // The readOnce option causes any storage-layer cursors created during plan
                 // execution to assume read data will not be needed again and need not be cached.
@@ -616,7 +605,7 @@ public:
                 // there is no ClientCursor id, and then return.
                 const long long numResults = 0;
                 const CursorId cursorId = 0;
-                endQueryOp(opCtx, collection, *exec, numResults, boost::none, cmdObj);
+                endQueryOp(opCtx, collection, *exec, numResults, cursorId);
                 auto bodyBuilder = result->getBodyBuilder();
                 appendCursorResponseObject(
                     cursorId, nss.ns(), BSONArray(), boost::none, &bodyBuilder);
@@ -667,6 +656,8 @@ public:
                 auto&& [stats, _] =
                     explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
                 LOGV2_WARNING(23798,
+                              "Plan executor error during find command: {error}, "
+                              "stats: {stats}, cmd: {cmd}",
                               "Plan executor error during find command",
                               "error"_attr = exception.toStatus(),
                               "stats"_attr = redact(stats),
@@ -715,24 +706,26 @@ public:
                     pinnedCursor.getCursor()->setLeftoverMaxTimeMicros(
                         opCtx->getRemainingMaxTimeMicros());
                 }
+                pinnedCursor.getCursor()->setNReturnedSoFar(numResults);
+                pinnedCursor.getCursor()->incNBatches();
 
                 // Fill out curop based on the results.
-                endQueryOp(opCtx, collection, *cursorExec, numResults, pinnedCursor, cmdObj);
+                endQueryOp(opCtx, collection, *cursorExec, numResults, cursorId);
 
                 if (stashResourcesForGetMore) {
                     // Collect storage stats now before we stash the recovery unit. These stats are
                     // normally collected in the service entry point layer just before a command
-                    // ends, but they must be collected before stashing the RecoveryUnit. Otherwise,
-                    // the service entry point layer will collect the stats from the new
-                    // RecoveryUnit, which wasn't actually used for the query.
+                    // ends, but they must be collected before stashing the
+                    // RecoveryUnit. Otherwise, the service entry point layer will collect the
+                    // stats from the new RecoveryUnit, which wasn't actually used for the query.
                     //
                     // The stats collected here will not get overwritten, as the service entry
                     // point layer will only set these stats when they're not empty.
                     CurOp::get(opCtx)->debug().storageStats =
-                        opCtx->recoveryUnit()->computeOperationStatisticsSinceLastCall();
+                        opCtx->recoveryUnit()->getOperationStatistics();
                 }
             } else {
-                endQueryOp(opCtx, collection, *exec, numResults, boost::none, cmdObj);
+                endQueryOp(opCtx, collection, *exec, numResults, cursorId);
             }
 
             // Generate the response object to send to the client.
@@ -759,7 +752,6 @@ public:
                 keyBob.append("min", 1);
                 keyBob.append("max", 1);
                 keyBob.append("shardVersion", 1);
-                keyBob.append("databaseVersion", 1);
                 return keyBob.obj();
             }();
 
@@ -773,25 +765,6 @@ public:
     private:
         const OpMsgRequest _request;
         const StringData _dbName;
-
-        // Parses the command object to a FindCommandRequest. If the client request did not specify
-        // any runtime constants, make them available to the query here.
-        std::unique_ptr<FindCommandRequest> _parseCmdObjectToFindCommandRequest(
-            OperationContext* opCtx, NamespaceString nss, BSONObj cmdObj) {
-            auto findCommand = query_request_helper::makeFromFindCommand(
-                std::move(cmdObj),
-                std::move(nss),
-                APIParameters::get(opCtx).getAPIStrict().value_or(false));
-
-            // Rewrite any FLE find payloads that exist in the query if this is a FLE 2 query.
-            if (shouldDoFLERewrite(findCommand)) {
-                invariant(findCommand->getNamespaceOrUUID().nss());
-                processFLEFindD(
-                    opCtx, findCommand->getNamespaceOrUUID().nss().value(), findCommand.get());
-            }
-
-            return translateNtoReturnToLimitOrBatchSize(std::move(findCommand));
-        }
     };
 
 } findCmd;
diff --git a/src/mongo/db/commands/fsync.cpp b/src/mongo/db/commands/fsync.cpp
index 35ae6323188..f1d92aa636a 100644
--- a/src/mongo/db/commands/fsync.cpp
+++ b/src/mongo/db/commands/fsync.cpp
@@ -47,13 +47,11 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/fsync_locked.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/exception_util.h"
-#include "mongo/db/dbdirectclient.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/backup_cursor_hooks.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/logv2/log.h"
-#include "mongo/s/sharding_feature_flags_gen.h"
 #include "mongo/stdx/condition_variable.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/background.h"
@@ -71,13 +69,10 @@ Lock::ResourceMutex commandMutex("fsyncCommandMutex");
  */
 class FSyncLockThread : public BackgroundJob {
 public:
-    FSyncLockThread(ServiceContext* serviceContext,
-                    bool allowFsyncFailure,
-                    const Milliseconds deadline)
+    FSyncLockThread(ServiceContext* serviceContext, bool allowFsyncFailure)
         : BackgroundJob(false),
           _serviceContext(serviceContext),
-          _allowFsyncFailure(allowFsyncFailure),
-          _deadline(deadline) {}
+          _allowFsyncFailure(allowFsyncFailure) {}
 
     std::string name() const override {
         return "FSyncLockThread";
@@ -89,7 +84,6 @@ private:
     ServiceContext* const _serviceContext;
     bool _allowFsyncFailure;
     static bool _shutdownTaskRegistered;
-    const Milliseconds _deadline;
 };
 
 class FSyncCommand : public ErrmsgCommandDeprecated {
@@ -131,20 +125,6 @@ public:
         actions.addAction(ActionType::fsync);
         out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
     }
-
-    virtual void checkForInProgressDDLOperations(OperationContext* opCtx) {
-        DBDirectClient client(opCtx);
-        const auto numDDLDocuments =
-            client.count(NamespaceString::kShardingDDLCoordinatorsNamespace);
-
-        if (numDDLDocuments != 0) {
-            LOGV2_WARNING(781541, "Cannot take lock while DDL operations is in progress");
-            releaseLock();
-            uasserted(ErrorCodes::IllegalOperation,
-                      "Cannot take lock while DDL operation is in progress");
-        }
-    }
-
     virtual bool errmsgRun(OperationContext* opCtx,
                            const std::string& dbname,
                            const BSONObj& cmdObj,
@@ -156,12 +136,7 @@ public:
         }
 
         const bool lock = cmdObj["lock"].trueValue();
-        const bool forBackup = cmdObj["forBackup"].trueValue();
-        LOGV2(20461,
-              "CMD fsync: lock:{lock}",
-              "CMD fsync",
-              "lock"_attr = lock,
-              "forBackup"_attr = forBackup);
+        LOGV2(20461, "CMD fsync: lock:{lock}", "CMD fsync", "lock"_attr = lock);
 
         // fsync + lock is sometimes used to block writes out of the system and does not care if
         // the `BackupCursorService::fsyncLock` call succeeds.
@@ -193,23 +168,8 @@ public:
                 stdx::unique_lock<Latch> lk(lockStateMutex);
                 threadStatus = Status::OK();
                 threadStarted = false;
-
-                Milliseconds deadline = Milliseconds::max();
-                if (forBackup) {
-                    // Set a default deadline of 90s for the fsyncLock to be acquired.
-                    deadline = Milliseconds(90000);
-                    // Parse the cmdObj and update the deadline if
-                    // "fsyncLockAcquisitionTimeoutMillis" exists.
-                    for (const auto& elem : cmdObj) {
-                        if (elem.fieldNameStringData() == "fsyncLockAcquisitionTimeoutMillis") {
-                            deadline = Milliseconds{uassertStatusOK(parseMaxTimeMS(elem))};
-                        }
-                    }
-                }
-
-                _lockThread = std::make_unique<FSyncLockThread>(
-                    opCtx->getServiceContext(), allowFsyncFailure, deadline);
-
+                _lockThread = std::make_unique<FSyncLockThread>(opCtx->getServiceContext(),
+                                                                allowFsyncFailure);
                 _lockThread->go();
 
                 while (!threadStarted && threadStatus.isOK()) {
@@ -229,13 +189,6 @@ public:
             }
         }
 
-        if (forBackup) {
-            // The check must be performed only if the fsync+lock command has been issued for backup
-            // purposes (through monogs). There are valid cases where fsync+lock can be invoked on
-            // the mongod while DDLs are in progress.
-            checkForInProgressDDLOperations(opCtx);
-        }
-
         LOGV2(20462,
               "mongod is locked and no writes are allowed. db.fsyncUnlock() to unlock, "
               "lock count is {lockCount}, for more info see {seeAlso}",
@@ -396,18 +349,7 @@ void FSyncLockThread::run() {
     try {
         const ServiceContext::UniqueOperationContext opCtxPtr = cc().makeOperationContext();
         OperationContext& opCtx = *opCtxPtr;
-
-        // If the deadline exists, set it on the opCtx and GlobalRead lock.
-        Date_t lockDeadline = Date_t::max();
-        if (_deadline < Milliseconds::max()) {
-            lockDeadline = Date_t::now() + _deadline;
-        }
-
-        opCtx.setDeadlineAfterNowBy(Milliseconds(_deadline), ErrorCodes::ExceededTimeLimit);
-        Lock::GlobalRead global(
-            &opCtx,
-            lockDeadline,
-            Lock::InterruptBehavior::kThrow);  // Block any writes in order to flush the files.
+        Lock::GlobalRead global(&opCtx);  // Block any writes in order to flush the files.
 
         StorageEngine* storageEngine = _serviceContext->getStorageEngine();
 
@@ -497,16 +439,7 @@ void FSyncLockThread::run() {
                 storageEngine->endBackup(&opCtx);
             }
         }
-    } catch (const ExceptionForCat<ErrorCategory::ExceededTimeLimitError>&) {
-        LOGV2_ERROR(204739, "Fsync timed out with ExceededTimeLimitError");
-        fsyncCmd.threadStatus = Status(ErrorCodes::Error::LockTimeout, "Fsync lock timed out");
-        fsyncCmd.acquireFsyncLockSyncCV.notify_one();
-        return;
-    } catch (const ExceptionFor<ErrorCodes::LockTimeout>&) {
-        LOGV2_ERROR(204740, "Fsync timed out with LockTimeout");
-        fsyncCmd.threadStatus = Status(ErrorCodes::Error::LockTimeout, "Fsync lock timed out");
-        fsyncCmd.acquireFsyncLockSyncCV.notify_one();
-        return;
+
     } catch (const std::exception& e) {
         LOGV2_FATAL(40350,
                     "FSyncLockThread exception: {error}",
diff --git a/src/mongo/db/commands/generic_servers.cpp b/src/mongo/db/commands/generic_servers.cpp
index 7a6b87c13fe..41448e90dde 100644
--- a/src/mongo/db/commands/generic_servers.cpp
+++ b/src/mongo/db/commands/generic_servers.cpp
@@ -156,11 +156,8 @@ HostInfoReply HostInfoCmd::Invocation::typedRun(OperationContext*) {
     system.setMemSizeMB(static_cast<long>(p.getSystemMemSizeMB()));
     system.setMemLimitMB(static_cast<long>(p.getMemSizeMB()));
     system.setNumCores(static_cast<int>(p.getNumAvailableCores()));
-    system.setNumPhysicalCores(static_cast<int>(p.getNumPhysicalCores()));
-    system.setNumCpuSockets(static_cast<int>(p.getNumCpuSockets()));
     system.setCpuArch(p.getArch());
     system.setNumaEnabled(p.hasNumaEnabled());
-    system.setNumNumaNodes(static_cast<int>(p.getNumNumaNodes()));
 
     HostInfoOsReply os;
     os.setType(p.getOsType());
diff --git a/src/mongo/db/commands/generic_servers.idl b/src/mongo/db/commands/generic_servers.idl
index c1f626a7f07..598ce277333 100644
--- a/src/mongo/db/commands/generic_servers.idl
+++ b/src/mongo/db/commands/generic_servers.idl
@@ -64,11 +64,8 @@ structs:
             memSizeMB: long
             memLimitMB: long
             numCores: int
-            numPhysicalCores: int
-            numCpuSockets: int
             cpuArch: string
             numaEnabled: bool
-            numNumaNodes: int
 
     hostInfoOsReply:
         description: "hostInfo.os reply fields"
diff --git a/src/mongo/db/commands/get_cluster_parameter_command.cpp b/src/mongo/db/commands/get_cluster_parameter_command.cpp
index 6f87839e88d..35060c90ed4 100644
--- a/src/mongo/db/commands/get_cluster_parameter_command.cpp
+++ b/src/mongo/db/commands/get_cluster_parameter_command.cpp
@@ -65,11 +65,6 @@ public:
         using InvocationBase::InvocationBase;
 
         Reply typedRun(OperationContext* opCtx) {
-            uassert(ErrorCodes::UnknownFeatureCompatibilityVersion,
-                    "FCV is not yet initialized, retry the command after FCV initialization has "
-                    "completed",
-                    serverGlobalParams.featureCompatibility.isVersionInitialized());
-
             uassert(
                 ErrorCodes::IllegalOperation,
                 "featureFlagClusterWideConfig not enabled",
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index 8d7510a7e2f..b3c00996ec6 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -431,6 +431,7 @@ public:
                 auto&& [stats, _] =
                     explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
                 LOGV2_WARNING(20478,
+                              "getMore command executor error: {error}, stats: {stats}, cmd: {cmd}",
                               "getMore command executor error",
                               "error"_attr = exception.toStatus(),
                               "stats"_attr = redact(stats),
@@ -616,6 +617,7 @@ public:
                 options.atClusterTime = repl::ReadConcernArgs::get(opCtx).getArgsAtClusterTime();
             }
             CursorResponseBuilder nextBatch(reply, options);
+            BSONObj obj;
             std::uint64_t numResults = 0;
             ResourceConsumption::DocumentUnitCounter docUnitsReturned;
 
@@ -632,7 +634,9 @@ public:
                     // Use the commit point of the last batch for exhaust cursors.
                     lastKnownCommittedOpTime = cursorPin->getLastKnownCommittedOpTime();
                 }
-                clientsLastKnownCommittedOpTime(opCtx) = lastKnownCommittedOpTime;
+                if (lastKnownCommittedOpTime) {
+                    clientsLastKnownCommittedOpTime(opCtx) = lastKnownCommittedOpTime.get();
+                }
 
                 awaitDataState(opCtx).shouldWaitForInserts = true;
             }
@@ -695,19 +699,10 @@ public:
 
                 cursorPin->setLeftoverMaxTimeMicros(opCtx->getRemainingMaxTimeMicros());
 
-                if (opCtx->isExhaust() && clientsLastKnownCommittedOpTime(opCtx)) {
-                    // Update the cursor's lastKnownCommittedOpTime to the current
-                    // lastCommittedOpTime. The lastCommittedOpTime now may be staler than the
-                    // actual lastCommittedOpTime returned in the metadata of this latest batch (see
-                    // appendReplyMetadata).  As a result, we may sometimes return more empty
-                    // batches than we need to. But it is fine to be conservative in this.
+                if (opCtx->isExhaust() && !clientsLastKnownCommittedOpTime(opCtx).isNull()) {
+                    // Set the commit point of the latest batch.
                     auto replCoord = repl::ReplicationCoordinator::get(opCtx);
-                    auto myLastCommittedOpTime = replCoord->getLastCommittedOpTime();
-                    auto clientsLastKnownCommittedOpTime = cursorPin->getLastKnownCommittedOpTime();
-                    if (!clientsLastKnownCommittedOpTime.has_value() ||
-                        clientsLastKnownCommittedOpTime.value() < myLastCommittedOpTime) {
-                        cursorPin->setLastKnownCommittedOpTime(myLastCommittedOpTime);
-                    }
+                    cursorPin->setLastKnownCommittedOpTime(replCoord->getLastCommittedOpTime());
                 }
             } else {
                 curOp->debug().cursorExhausted = true;
@@ -719,9 +714,12 @@ public:
             // documents.
             auto& metricsCollector = ResourceConsumption::MetricsCollector::get(opCtx);
             metricsCollector.incrementDocUnitsReturned(docUnitsReturned);
-            curOp->debug().additiveMetrics.nBatches = 1;
-            curOp->setEndOfOpMetrics(numResults);
-            collectQueryStatsMongod(opCtx, cursorPin);
+            cursorPin->incNReturnedSoFar(numResults);
+            cursorPin->incNBatches();
+
+            // Ensure log and profiler include the number of results returned in this getMore's
+            // response batch.
+            curOp->debug().nreturned = numResults;
 
             if (respondWithId) {
                 cursorDeleter.dismiss();
diff --git a/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp b/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp
index ffb4b0c8555..f9648db8105 100644
--- a/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp
+++ b/src/mongo/db/commands/internal_rename_if_options_and_indexes_match_cmd.cpp
@@ -112,9 +112,8 @@ public:
             RenameCollectionOptions options;
             options.dropTarget = true;
             options.stayTemp = false;
-            options.originalIndexes = indexList;
-            options.originalCollectionOptions = collectionOptions;
-            doLocalRenameIfOptionsAndIndexesHaveNotChanged(opCtx, fromNss, toNss, options);
+            doLocalRenameIfOptionsAndIndexesHaveNotChanged(
+                opCtx, fromNss, toNss, options, std::move(indexList), collectionOptions);
         }
 
         NamespaceString ns() const override {
diff --git a/src/mongo/db/commands/killoperations_common.h b/src/mongo/db/commands/killoperations_common.h
index bae129a4bbe..b9dd9eb79a9 100644
--- a/src/mongo/db/commands/killoperations_common.h
+++ b/src/mongo/db/commands/killoperations_common.h
@@ -64,8 +64,7 @@ public:
             auto opKeys = Base::request().getOperationKeys();
 
             for (auto& opKey : opKeys) {
-                LOGV2_DEBUG(
-                    4615602, 2, "Attempting to kill operation", "operationKey"_attr = opKey);
+                LOGV2(4615602, "Attempting to kill operation", "operationKey"_attr = opKey);
                 opKiller.killOperation(OperationKey(opKey));
             }
             Derived::killCursors(opCtx, opKeys);
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index ec45ee6356b..f78cfda6308 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -167,7 +167,9 @@ BSONObj buildViewBson(const ViewDefinition& view, bool nameOnly) {
     return b.obj();
 }
 
-BSONObj buildTimeseriesBson(const CollectionPtr& collection, bool nameOnly) {
+BSONObj buildTimeseriesBson(OperationContext* opCtx,
+                            const CollectionPtr& collection,
+                            bool nameOnly) {
     invariant(collection);
 
     BSONObjBuilder builder;
@@ -375,7 +377,8 @@ public:
                                     if (auto bucketsCollection = CollectionCatalog::get(opCtx)
                                                                      ->lookupCollectionByNamespace(
                                                                          opCtx, view->viewOn())) {
-                                        return buildTimeseriesBson(bucketsCollection, nameOnly);
+                                        return buildTimeseriesBson(
+                                            opCtx, bucketsCollection, nameOnly);
                                     } else {
                                         // The buckets collection does not exist, so the time-series
                                         // view will be appended when we iterate through the view
@@ -394,22 +397,21 @@ public:
                     } else {
                         auto perCollectionWork = [&](const CollectionPtr& collection) {
                             if (collection && collection->getTimeseriesOptions() &&
-                                !collection->ns().isDropPendingNamespace()) {
-                                auto viewNss = collection->ns().getTimeseriesViewNamespace();
-                                auto view =
-                                    catalog->lookupViewWithoutValidatingDurable(opCtx, viewNss);
-                                if (view && view->timeseries() &&
-                                    (!authorizedCollections ||
-                                     as->isAuthorizedForAnyActionOnResource(
-                                         ResourcePattern::forExactNamespace(viewNss)))) {
-                                    // The time-series view for this buckets namespace exists, so
-                                    // add it here while we have the collection options.
-                                    _addWorkingSetMember(opCtx,
-                                                         buildTimeseriesBson(collection, nameOnly),
-                                                         matcher.get(),
-                                                         ws.get(),
-                                                         root.get());
-                                }
+                                !collection->ns().isDropPendingNamespace() &&
+                                catalog->lookupViewWithoutValidatingDurable(
+                                    opCtx, collection->ns().getTimeseriesViewNamespace()) &&
+                                (!authorizedCollections ||
+                                 as->isAuthorizedForAnyActionOnResource(
+                                     ResourcePattern::forExactNamespace(
+                                         collection->ns().getTimeseriesViewNamespace())))) {
+                                // The time-series view for this buckets namespace exists, so add it
+                                // here while we have the collection options.
+                                _addWorkingSetMember(
+                                    opCtx,
+                                    buildTimeseriesBson(opCtx, collection, nameOnly),
+                                    matcher.get(),
+                                    ws.get(),
+                                    root.get());
                             }
 
                             if (authorizedCollections &&
@@ -433,8 +435,10 @@ public:
                         // needing to yield as we don't take any locks.
                         if (opCtx->isLockFreeReadsOp()) {
                             auto collectionCatalog = CollectionCatalog::get(opCtx);
-                            for (auto&& coll : collectionCatalog->range(tenantDbName)) {
-                                perCollectionWork(coll);
+                            for (auto it = collectionCatalog->begin(opCtx, tenantDbName);
+                                 it != collectionCatalog->end(opCtx);
+                                 ++it) {
+                                perCollectionWork(*it);
                             }
                         } else {
                             mongo::catalog::forEachCollectionFromDb(
@@ -497,7 +501,7 @@ public:
                     batchSize = *listCollRequest.getCursor()->getBatchSize();
                 }
 
-                FindCommon::BSONArrayResponseSizeTracker responseSizeTracker;
+                size_t bytesBuffered = 0;
                 for (long long objCount = 0; objCount < batchSize; objCount++) {
                     BSONObj nextDoc;
                     PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr);
@@ -508,7 +512,7 @@ public:
 
                     // If we can't fit this result inside the current batch, then we stash it for
                     // later.
-                    if (!responseSizeTracker.haveSpaceForNext(nextDoc)) {
+                    if (!FindCommon::haveSpaceForNext(nextDoc, objCount, bytesBuffered)) {
                         exec->stashResult(nextDoc);
                         break;
                     }
@@ -524,7 +528,7 @@ public:
                             "error"_attr = exc);
                         fassertFailed(5254301);
                     }
-                    responseSizeTracker.add(nextDoc);
+                    bytesBuffered += nextDoc.objsize();
                 }
                 if (exec->isEOF()) {
                     return createListCollectionsCursorReply(
diff --git a/src/mongo/db/commands/list_databases.cpp b/src/mongo/db/commands/list_databases.cpp
index fd60d2b26b5..4b59a41f64e 100644
--- a/src/mongo/db/commands/list_databases.cpp
+++ b/src/mongo/db/commands/list_databases.cpp
@@ -35,7 +35,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/list_databases_gen.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/matcher/expression.h"
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 22cbb4b11df..6f4f4f461b9 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/catalog/list_indexes.h"
 #include "mongo/db/clientcursor.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/cursor_manager.h"
@@ -64,40 +65,38 @@ namespace mongo {
 namespace {
 
 // The allowed fields have to be in sync with those defined in 'src/mongo/db/list_indexes.idl'.
-static std::map<StringData, std::set<IndexType>> allowedFieldNames = {
-    {ListIndexesReplyItem::k2dsphereIndexVersionFieldName,
-     {IndexType::INDEX_2DSPHERE, IndexType::INDEX_2DSPHERE_BUCKET}},
-    {ListIndexesReplyItem::kBackgroundFieldName, {}},
-    {ListIndexesReplyItem::kBitsFieldName, {IndexType::INDEX_2D}},
-    {ListIndexesReplyItem::kBucketSizeFieldName, {}},
-    {ListIndexesReplyItem::kBuildUUIDFieldName, {}},
-    {ListIndexesReplyItem::kClusteredFieldName, {}},
-    {ListIndexesReplyItem::kCoarsestIndexedLevelFieldName, {IndexType::INDEX_2DSPHERE}},
-    {ListIndexesReplyItem::kCollationFieldName, {}},
-    {ListIndexesReplyItem::kDefault_languageFieldName, {}},
-    {ListIndexesReplyItem::kDropDupsFieldName, {}},
-    {ListIndexesReplyItem::kExpireAfterSecondsFieldName, {}},
-    {ListIndexesReplyItem::kFinestIndexedLevelFieldName, {IndexType::INDEX_2DSPHERE}},
-    {ListIndexesReplyItem::kHiddenFieldName, {}},
-    {ListIndexesReplyItem::kIndexBuildInfoFieldName, {}},
-    {ListIndexesReplyItem::kKeyFieldName, {}},
-    {ListIndexesReplyItem::kLanguage_overrideFieldName, {}},
-    {ListIndexesReplyItem::kMaxFieldName, {IndexType::INDEX_2D}},
-    {ListIndexesReplyItem::kMinFieldName, {IndexType::INDEX_2D}},
-    {ListIndexesReplyItem::kNameFieldName, {}},
-    {ListIndexesReplyItem::kNsFieldName, {}},
-    {ListIndexesReplyItem::kOriginalSpecFieldName, {}},
-    {ListIndexesReplyItem::kPartialFilterExpressionFieldName, {}},
-    {ListIndexesReplyItem::kPrepareUniqueFieldName, {}},
-    {ListIndexesReplyItem::kSparseFieldName, {}},
-    {ListIndexesReplyItem::kSpecFieldName, {}},
-    {ListIndexesReplyItem::kStorageEngineFieldName, {}},
-    {ListIndexesReplyItem::kTextIndexVersionFieldName, {IndexType::INDEX_TEXT}},
-    {ListIndexesReplyItem::kUniqueFieldName, {}},
-    {ListIndexesReplyItem::kVFieldName, {}},
-    {ListIndexesReplyItem::kWeightsFieldName, {IndexType::INDEX_TEXT}},
-    {ListIndexesReplyItem::kWildcardProjectionFieldName, {IndexType::INDEX_WILDCARD}},
-};
+static std::set<StringData> allowedFieldNames = {
+    ListIndexesReplyItem::k2dsphereIndexVersionFieldName,
+    ListIndexesReplyItem::kBackgroundFieldName,
+    ListIndexesReplyItem::kBitsFieldName,
+    ListIndexesReplyItem::kBucketSizeFieldName,
+    ListIndexesReplyItem::kBuildUUIDFieldName,
+    ListIndexesReplyItem::kClusteredFieldName,
+    ListIndexesReplyItem::kCoarsestIndexedLevelFieldName,
+    ListIndexesReplyItem::kCollationFieldName,
+    ListIndexesReplyItem::kDefault_languageFieldName,
+    ListIndexesReplyItem::kDropDupsFieldName,
+    ListIndexesReplyItem::kExpireAfterSecondsFieldName,
+    ListIndexesReplyItem::kFinestIndexedLevelFieldName,
+    ListIndexesReplyItem::kHiddenFieldName,
+    ListIndexesReplyItem::kIndexBuildInfoFieldName,
+    ListIndexesReplyItem::kKeyFieldName,
+    ListIndexesReplyItem::kLanguage_overrideFieldName,
+    ListIndexesReplyItem::kMaxFieldName,
+    ListIndexesReplyItem::kMinFieldName,
+    ListIndexesReplyItem::kNameFieldName,
+    ListIndexesReplyItem::kNsFieldName,
+    ListIndexesReplyItem::kOriginalSpecFieldName,
+    ListIndexesReplyItem::kPartialFilterExpressionFieldName,
+    ListIndexesReplyItem::kPrepareUniqueFieldName,
+    ListIndexesReplyItem::kSparseFieldName,
+    ListIndexesReplyItem::kSpecFieldName,
+    ListIndexesReplyItem::kStorageEngineFieldName,
+    ListIndexesReplyItem::kTextIndexVersionFieldName,
+    ListIndexesReplyItem::kUniqueFieldName,
+    ListIndexesReplyItem::kVFieldName,
+    ListIndexesReplyItem::kWeightsFieldName,
+    ListIndexesReplyItem::kWildcardProjectionFieldName};
 
 /**
  * Returns index specs, with resolved namespace, from the catalog for this listIndexes request.
@@ -307,7 +306,7 @@ public:
                                             nss));
 
             std::vector<mongo::ListIndexesReplyItem> firstBatch;
-            FindCommon::BSONArrayResponseSizeTracker responseSizeTracker;
+            size_t bytesBuffered = 0;
             for (long long objCount = 0; objCount < batchSize; objCount++) {
                 BSONObj nextDoc;
                 PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr);
@@ -319,7 +318,7 @@ public:
 
                 // If we can't fit this result inside the current batch, then we stash it for
                 // later.
-                if (!responseSizeTracker.haveSpaceForNext(nextDoc)) {
+                if (!FindCommon::haveSpaceForNext(nextDoc, objCount, bytesBuffered)) {
                     exec->stashResult(nextDoc);
                     break;
                 }
@@ -338,7 +337,7 @@ public:
                                           nextDoc.toString(),
                                           exc.toString()));
                 }
-                responseSizeTracker.add(nextDoc);
+                bytesBuffered += nextDoc.objsize();
             }
 
             if (exec->isEOF()) {
diff --git a/src/mongo/db/commands/map_reduce_agg.cpp b/src/mongo/db/commands/map_reduce_agg.cpp
index 91a27c93bf4..eb1d432f5b5 100644
--- a/src/mongo/db/commands/map_reduce_agg.cpp
+++ b/src/mongo/db/commands/map_reduce_agg.cpp
@@ -138,8 +138,6 @@ bool runAggregationMapReduce(OperationContext* opCtx,
     Timer cmdTimer;
 
     auto parsedMr = MapReduceCommandRequest::parse(IDLParserErrorContext("mapReduce"), cmd);
-    auto curop = CurOp::get(opCtx);
-    curop->beginQueryPlanningTimer();
     auto expCtx = makeExpressionContext(opCtx, parsedMr, verbosity);
     auto runnablePipeline = [&]() {
         auto pipeline = map_reduce_common::translateFromMR(parsedMr, expCtx);
@@ -148,10 +146,10 @@ bool runAggregationMapReduce(OperationContext* opCtx,
     }();
     auto exec = plan_executor_factory::make(expCtx, std::move(runnablePipeline));
     auto&& explainer = exec->getPlanExplainer();
-    // Store the plan summary string in CurOp.
+
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
-        curop->setPlanSummary_inlock(explainer.getPlanSummary());
+        CurOp::get(opCtx)->setPlanSummary_inlock(explainer.getPlanSummary());
     }
 
     try {
diff --git a/src/mongo/db/commands/map_reduce_agg_test.cpp b/src/mongo/db/commands/map_reduce_agg_test.cpp
index b47b2bfa0e4..52ee67c416b 100644
--- a/src/mongo/db/commands/map_reduce_agg_test.cpp
+++ b/src/mongo/db/commands/map_reduce_agg_test.cpp
@@ -46,6 +46,15 @@
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/unittest/unittest.h"
 
+#define ASSERT_DOES_NOT_THROW(EXPRESSION)                                          \
+    try {                                                                          \
+        EXPRESSION;                                                                \
+    } catch (const AssertionException& e) {                                        \
+        str::stream err;                                                           \
+        err << "Threw an exception incorrectly: " << e.toString();                 \
+        ::mongo::unittest::TestAssertionFailure(__FILE__, __LINE__, err).stream(); \
+    }
+
 namespace mongo {
 namespace {
 
diff --git a/src/mongo/db/commands/map_reduce_command_base.h b/src/mongo/db/commands/map_reduce_command_base.h
index c0d19d6372a..f3806c40766 100644
--- a/src/mongo/db/commands/map_reduce_command_base.h
+++ b/src/mongo/db/commands/map_reduce_command_base.h
@@ -63,7 +63,7 @@ public:
                 {kDefaultReadConcernNotPermitted}};
     }
 
-    bool shouldAffectReadOptionCounters() const override {
+    bool shouldAffectReadConcernCounter() const override {
         return true;
     }
 
diff --git a/src/mongo/db/commands/mr_test.cpp b/src/mongo/db/commands/mr_test.cpp
index 95fe13b68ee..745d883c8e1 100644
--- a/src/mongo/db/commands/mr_test.cpp
+++ b/src/mongo/db/commands/mr_test.cpp
@@ -42,7 +42,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/map_reduce_gen.h"
 #include "mongo/db/commands/mr_common.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/json.h"
 #include "mongo/db/op_observer_noop.h"
diff --git a/src/mongo/db/commands/oplog_note.cpp b/src/mongo/db/commands/oplog_note.cpp
index 9f93f116815..7af5c111ca3 100644
--- a/src/mongo/db/commands/oplog_note.cpp
+++ b/src/mongo/db/commands/oplog_note.cpp
@@ -39,7 +39,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/resource_pattern.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/exception_util.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 2042914e16b..eb5f74d4387 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -89,7 +89,7 @@ public:
             this, opMsgRequest, std::move(aggregationRequest), std::move(privileges));
     }
 
-    bool shouldAffectReadOptionCounters() const override {
+    bool shouldAffectReadConcernCounter() const override {
         return true;
     }
 
@@ -142,6 +142,7 @@ public:
                 opCtx, !Pipeline::aggHasWriteStage(_request.body));
 
             uassertStatusOK(runAggregate(opCtx,
+                                         _aggregationRequest.getNamespace(),
                                          _aggregationRequest,
                                          _liteParsedPipeline,
                                          _request.body,
@@ -164,6 +165,7 @@ public:
                      rpc::ReplyBuilderInterface* result) override {
 
             uassertStatusOK(runAggregate(opCtx,
+                                         _aggregationRequest.getNamespace(),
                                          _aggregationRequest,
                                          _liteParsedPipeline,
                                          _request.body,
diff --git a/src/mongo/db/commands/profile.idl b/src/mongo/db/commands/profile.idl
index f1c56a88a05..cd4a295a950 100644
--- a/src/mongo/db/commands/profile.idl
+++ b/src/mongo/db/commands/profile.idl
@@ -66,16 +66,3 @@ commands:
                       an alternative to slowms and sampleRate. The special value 'unset' removes
                       the filter."
         optional: true
-
-  setProfilingFilterGlobally:
-    description: "Parser for the 'setProfilingFilterGlobally' command."
-    command_name: "setProfilingFilterGlobally"
-    cpp_name: SetProfilingFilterGloballyCmdRequest
-    strict: true
-    namespace: ignored
-    api_version: ""
-    fields:
-      filter:
-        type: ObjectOrUnset
-        description: "A query predicate that determines which ops are logged/profiled on a global
-                      level. The special value 'unset' removes the filter."
diff --git a/src/mongo/db/commands/resize_oplog.cpp b/src/mongo/db/commands/resize_oplog.cpp
index 4433edae662..48dc95ade4e 100644
--- a/src/mongo/db/commands/resize_oplog.cpp
+++ b/src/mongo/db/commands/resize_oplog.cpp
@@ -37,7 +37,7 @@
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/resize_oplog_gen.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/jsobj.h"
 #include "mongo/db/operation_context.h"
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 66d68f4413c..9172103b493 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -74,9 +74,6 @@
 #include "mongo/db/query/query_feature_flags_gen.h"
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/query/query_planner_common.h"
-#include "mongo/db/query/query_stats/agg_key.h"
-#include "mongo/db/query/query_stats/key.h"
-#include "mongo/db/query/query_stats/query_stats.h"
 #include "mongo/db/read_concern.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/read_concern_args.h"
@@ -107,8 +104,6 @@ namespace {
 ServerStatusMetricField<Counter64> allowDiskUseMetric{"query.allowDiskUseFalse",
                                                       &allowDiskUseFalseCounter};
 
-MONGO_FAIL_POINT_DEFINE(hangAfterCreatingAggregationPlan);
-
 /**
  * If a pipeline is empty (assuming that a $cursor stage hasn't been created yet), it could mean
  * that we were able to absorb all pipeline stages and pull them into a single PlanExecutor. So,
@@ -226,6 +221,7 @@ bool handleCursorCommand(OperationContext* opCtx,
             auto&& [stats, _] =
                 explainer.getWinningPlanStats(ExplainOptions::Verbosity::kExecStats);
             LOGV2_WARNING(23799,
+                          "Aggregate command executor error: {error}, stats: {stats}, cmd: {cmd}",
                           "Aggregate command executor error",
                           "error"_attr = exception.toStatus(),
                           "stats"_attr = redact(stats),
@@ -630,6 +626,7 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx
 
         getSearchHelpers(expCtx->opCtx->getServiceContext())
             ->injectSearchShardFiltererIfNeeded(pipeline.get());
+
         // Complete creation of the initial $cursor stage, if needed.
         PipelineD::attachInnerQueryExecutorToPipeline(collections,
                                                       attachExecutorCallback.first,
@@ -642,6 +639,7 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx
             // There are separate ExpressionContexts for each exchange pipeline, so make sure to
             // pass the pipeline's ExpressionContext to the plan executor factory.
             auto pipelineExpCtx = pipelineIt->getContext();
+
             execs.emplace_back(
                 plan_executor_factory::make(std::move(pipelineExpCtx),
                                             std::move(pipelineIt),
@@ -659,222 +657,25 @@ std::vector<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createLegacyEx
     return execs;
 }
 
-Status runAggregateOnView(OperationContext* opCtx,
-                          const NamespaceString& origNss,
-                          const AggregateCommandRequest& request,
-                          const MultipleCollectionAccessor& collections,
-                          boost::optional<std::unique_ptr<CollatorInterface>> collatorToUse,
-                          const ViewDefinition* view,
-                          std::shared_ptr<const CollectionCatalog> catalog,
-                          const PrivilegeVector& privileges,
-                          rpc::ReplyBuilderInterface* result,
-                          const std::function<void(void)>& resetContextFn) {
-    auto nss = request.getNamespace();
-
-    uassert(ErrorCodes::CommandNotSupportedOnView,
-            "mapReduce on a view is not supported",
-            !request.getIsMapReduceCommand());
-
-    // Check that the default collation of 'view' is compatible with the operation's
-    // collation. The check is skipped if the request did not specify a collation.
-    if (!request.getCollation().get_value_or(BSONObj()).isEmpty()) {
-        invariant(collatorToUse);  // Should already be resolved at this point.
-        if (!CollatorInterface::collatorsMatch(view->defaultCollator(), collatorToUse->get()) &&
-            !view->timeseries()) {
-
-            return {ErrorCodes::OptionNotSupportedOnView,
-                    "Cannot override a view's default collation"};
-        }
-    }
-
-    // Queries on timeseries views may specify non-default collation whereas queries
-    // on all other types of views must match the default collator (the collation use
-    // to originally create that collections). Thus in the case of operations on TS
-    // views, we use the request's collation.
-    auto timeSeriesCollator = view->timeseries() ? request.getCollation() : boost::none;
-
-    auto resolvedView =
-        uassertStatusOK(view_catalog_helpers::resolveView(opCtx, catalog, nss, timeSeriesCollator));
-
-    // With the view & collation resolved, we can relinquish locks.
-    resetContextFn();
-
-    // Set this operation's shard version for the underlying collection to unsharded.
-    // This is prerequisite for future shard versioning checks.
-    ScopedSetShardRole scopedSetShardRole(opCtx,
-                                          resolvedView.getNamespace(),
-                                          ChunkVersion::UNSHARDED() /* shardVersion */,
-                                          boost::none /* databaseVersion */);
-
-    uassert(std::move(resolvedView),
-            "Explain of a resolved view must be executed by mongos",
-            !ShardingState::get(opCtx)->enabled() || !request.getExplain());
-
-    // Parse the resolved view into a new aggregation request.
-    auto newRequest = resolvedView.asExpandedViewAggregation(request);
-    auto newCmd = aggregation_request_helper::serializeToCommandObj(newRequest);
-
-    auto status{Status::OK()};
-    try {
-        status = runAggregate(opCtx, newRequest, newCmd, privileges, result, resolvedView, request);
-    } catch (const ExceptionForCat<ErrorCategory::StaleShardVersionError>& ex) {
-        // Since we expect the view to be UNSHARDED, if we reached to this point there are
-        // two possibilities:
-        //   1. The shard doesn't know what its shard version/state is and needs to recover
-        //      it (in which case we throw so that the shard can run recovery)
-        //   2. The collection references by the view is actually SHARDED, in which case the
-        //      router must execute it
-        if (const auto staleInfo{ex.extraInfo<StaleConfigInfo>()}) {
-            uassert(std::move(resolvedView),
-                    "Resolved views on sharded collections must be executed by mongos",
-                    !staleInfo->getVersionWanted());
-        }
-        throw;
-    }
-
-    {
-        // Set the namespace of the curop back to the view namespace so ctx records
-        // stats on this view namespace on destruction.
-        stdx::lock_guard<Client> lk(*opCtx->getClient());
-        CurOp::get(opCtx)->setNS_inlock(nss.ns());
-    }
-
-    return status;
-}
-
-/**
- * Determines the collection type of the query by precedence of various configurations. The order
- * of these checks is critical since there may be overlap (e.g., a view over a virtual collection
- * is classified as a view).
- */
-query_shape::CollectionType determineCollectionType(
-    const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx,
-    boost::optional<const ResolvedView&> resolvedView,
-    bool hasChangeStream,
-    bool isCollectionless) {
-    if (resolvedView.has_value()) {
-        if (resolvedView->timeseries()) {
-            return query_shape::CollectionType::kTimeseries;
-        }
-        return query_shape::CollectionType::kView;
-    }
-    if (isCollectionless) {
-        return query_shape::CollectionType::kVirtual;
-    }
-    if (hasChangeStream) {
-        return query_shape::CollectionType::kChangeStream;
-    }
-    return ctx ? ctx->getCollectionType() : query_shape::CollectionType::kUnknown;
-}
-
-std::unique_ptr<Pipeline, PipelineDeleter> parsePipelineAndRegisterQueryStats(
-    OperationContext* opCtx,
-    const NamespaceString& origNss,
-    const AggregateCommandRequest& request,
-    const boost::optional<AutoGetCollectionForReadCommandMaybeLockFree>& ctx,
-    std::unique_ptr<CollatorInterface> collator,
-    boost::optional<UUID> uuid,
-    ExpressionContext::CollationMatchesDefault collationMatchesDefault,
-    const MultipleCollectionAccessor& collections,
-    stdx::unordered_set<NamespaceString> pipelineInvolvedNamespaces,
-    const LiteParsedPipeline& liteParsedPipeline,
-    bool isCollectionless,
-    boost::optional<const ResolvedView&> resolvedView,
-    boost::optional<const AggregateCommandRequest&> origRequest) {
-    // If we're operating over a view, we first parse just the original user-given request
-    // for the sake of registering query stats. Then, we'll parse the view pipeline and stitch
-    // the two pipelines together below.
-    auto expCtx =
-        makeExpressionContext(opCtx, request, std::move(collator), uuid, collationMatchesDefault);
-    // If any involved collection contains extended-range data, set a flag which individual
-    // DocumentSource parsers can check.
-    collections.forEach([&](const CollectionPtr& coll) {
-        if (coll->getRequiresTimeseriesExtendedRangeSupport())
-            expCtx->setRequiresTimeseriesExtendedRangeSupport(true);
-    });
-
-    const bool hasChangeStream = liteParsedPipeline.hasChangeStream();
-    // A pipeline with $changeStreamSplitLargeEvent requires the use of resume token format
-    // v2, since the 'fragmentNum' field only exists in this version and later.
-    if (hasChangeStream && liteParsedPipeline.endsWithChangeStreamSplitLargeEvent()) {
-        expCtx->changeStreamTokenVersion = 2;
-    }
-
-    auto requestForQueryStats = origRequest.has_value() ? *origRequest : request;
-    expCtx->startExpressionCounters();
-    auto pipeline = Pipeline::parse(requestForQueryStats.getPipeline(), expCtx);
-    expCtx->stopExpressionCounters();
-
-    // Register query stats with the pre-optimized pipeline. Exclude queries against collections
-    // with encrypted fields. We still collect query stats on collection-less aggregations.
-    bool hasEncryptedFields = ctx && ctx->getCollection() &&
-        ctx->getCollection()->getCollectionOptions().encryptedFieldConfig;
-    if (!hasEncryptedFields) {
-        // If this is a query over a resolved view, we want to register query stats with the
-        // original user-given request and pipeline, rather than the new request generated when
-        // resolving the view.
-        auto collectionType =
-            determineCollectionType(ctx, resolvedView, hasChangeStream, isCollectionless);
-
-        query_stats::registerRequest(opCtx,
-                                     origNss,
-                                     [&]() {
-                                         return std::make_unique<query_stats::AggKey>(
-                                             requestForQueryStats,
-                                             *pipeline,
-                                             expCtx,
-                                             pipelineInvolvedNamespaces,
-                                             origNss,
-                                             collectionType);
-                                     },
-                                     hasChangeStream);
-    }
-
-    if (resolvedView.has_value()) {
-        expCtx->startExpressionCounters();
-
-        if (resolvedView->timeseries()) {
-            // For timeseries, there may have been rewrites done on the raw BSON pipeline
-            // during view resolution. We must parse the request's full resolved pipeline
-            // which will account for those rewrites.
-            // TODO SERVER-82101 Re-organize timeseries rewrites so timeseries can follow the
-            // same pattern here as other views
-            pipeline = Pipeline::parse(request.getPipeline(), expCtx);
-        } else {
-            // Parse the view pipeline, then stitch the user pipeline and view pipeline together
-            // to build the total aggregation pipeline.
-            auto userPipeline = std::move(pipeline);
-            pipeline = Pipeline::parse(resolvedView->getPipeline(), expCtx);
-            pipeline->appendPipeline(std::move(userPipeline));
-        }
-
-        expCtx->stopExpressionCounters();
-    }
-
-    return pipeline;
-}
 }  // namespace
 
 Status runAggregate(OperationContext* opCtx,
+                    const NamespaceString& nss,
                     AggregateCommandRequest& request,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result,
-                    boost::optional<const ResolvedView&> resolvedView,
-                    boost::optional<const AggregateCommandRequest&> origRequest) {
-    return runAggregate(
-        opCtx, request, {request}, cmdObj, privileges, result, resolvedView, origRequest);
+                    rpc::ReplyBuilderInterface* result) {
+    return runAggregate(opCtx, nss, request, {request}, cmdObj, privileges, result);
 }
 
 Status runAggregate(OperationContext* opCtx,
+                    const NamespaceString& origNss,
                     AggregateCommandRequest& request,
                     const LiteParsedPipeline& liteParsedPipeline,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result,
-                    boost::optional<const ResolvedView&> resolvedView,
-                    boost::optional<const AggregateCommandRequest&> origRequest) {
-    auto origNss = origRequest.has_value() ? origRequest->getNamespace() : request.getNamespace();
+                    rpc::ReplyBuilderInterface* result) {
+
     // Perform some validations on the LiteParsedPipeline and request before continuing with the
     // aggregation command.
     performValidationChecks(opCtx, request, liteParsedPipeline);
@@ -944,8 +745,7 @@ Status runAggregate(OperationContext* opCtx,
         boost::optional<AutoStatsTracker> statsTracker;
 
         // If this is a change stream, perform special checks and change the execution namespace.
-        const auto hasChangeStream = liteParsedPipeline.hasChangeStream();
-        if (hasChangeStream) {
+        if (liteParsedPipeline.hasChangeStream()) {
             uassert(4928900,
                     str::stream() << AggregateCommandRequest::kCollectionUUIDFieldName
                                   << " is not supported for a change stream",
@@ -959,6 +759,7 @@ Status runAggregate(OperationContext* opCtx,
 
             // Raise an error if 'origNss' is a view. We do not need to check this if we are opening
             // a stream on an entire db or across the cluster.
+            const TenantDatabaseName origTenantDbName(boost::none, origNss.db());
             if (!origNss.isCollectionlessAggregateNS()) {
                 auto view = catalog->lookupView(opCtx, origNss);
                 uassert(ErrorCodes::CommandNotSupportedOnView,
@@ -992,7 +793,7 @@ Status runAggregate(OperationContext* opCtx,
                                  nss,
                                  Top::LockType::NotLocked,
                                  AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                 catalog->getDatabaseProfileLevel(nss.db()));
+                                 0);
             auto [collator, match] = PipelineD::resolveCollator(
                 opCtx, request.getCollation().get_value_or(BSONObj()), nullptr);
             collatorToUse.emplace(std::move(collator));
@@ -1014,53 +815,108 @@ Status runAggregate(OperationContext* opCtx,
             }
         }
 
-        // If collectionUUID was provided, verify the collection exists and has the expected UUID.
-        checkCollectionUUIDMismatch(opCtx,
-                                    nss,
-                                    collections.getMainCollection(),
-                                    request.getCollectionUUID(),
-                                    false /* checkFeatureFlag */);
-
         // If this is a view, resolve it by finding the underlying collection and stitching view
         // pipelines and this request's pipeline together. We then release our locks before
         // recursively calling runAggregate(), which will re-acquire locks on the underlying
         // collection.  (The lock must be released because recursively acquiring locks on the
         // database will prohibit yielding.)
-        // We do not need to expand the view pipeline when there is a $collStats stage, as
-        // $collStats is supported on a view namespace. For a time-series collection, however, the
-        // view is abstracted out for the users, so we needed to resolve the namespace to get the
-        // underlying bucket collection.
-        if (ctx && ctx->getView() &&
-            (!liteParsedPipeline.startsWithCollStats() || ctx->getView()->timeseries())) {
-            return runAggregateOnView(opCtx,
-                                      origNss,
-                                      request,
-                                      collections,
-                                      std::move(collatorToUse),
-                                      ctx->getView(),
-                                      catalog,
-                                      privileges,
-                                      result,
-                                      resetContext);
+        if (ctx && ctx->getView() && !liteParsedPipeline.startsWithCollStats()) {
+            invariant(nss != NamespaceString::kRsOplogNamespace);
+            invariant(!nss.isCollectionlessAggregateNS());
+
+            checkCollectionUUIDMismatch(opCtx,
+                                        nss,
+                                        collections.getMainCollection(),
+                                        request.getCollectionUUID(),
+                                        false /* checkFeatureFlag */);
+
+            uassert(ErrorCodes::CommandNotSupportedOnView,
+                    "mapReduce on a view is not supported",
+                    !request.getIsMapReduceCommand());
+
+            // Check that the default collation of 'view' is compatible with the operation's
+            // collation. The check is skipped if the request did not specify a collation.
+            if (!request.getCollation().get_value_or(BSONObj()).isEmpty()) {
+                invariant(collatorToUse);  // Should already be resolved at this point.
+                if (!CollatorInterface::collatorsMatch(ctx->getView()->defaultCollator(),
+                                                       collatorToUse->get()) &&
+                    !ctx->getView()->timeseries()) {
+
+                    return {ErrorCodes::OptionNotSupportedOnView,
+                            "Cannot override a view's default collation"};
+                }
+            }
+
+            // Queries on timeseries views may specify non-default collation whereas queries
+            // on all other types of views must match the default collator (the collation use
+            // to originally create that collections). Thus in the case of operations on TS
+            // views, we use the request's collation.
+            auto timeSeriesCollator =
+                ctx->getView()->timeseries() ? request.getCollation() : boost::none;
+
+            auto resolvedView = uassertStatusOK(
+                view_catalog_helpers::resolveView(opCtx, catalog, nss, timeSeriesCollator));
+
+            // With the view & collation resolved, we can relinquish locks.
+            resetContext();
+
+            // Set this operation's shard version for the underlying collection to unsharded.
+            // This is prerequisite for future shard versioning checks.
+            ScopedSetShardRole scopedSetShardRole(opCtx,
+                                                  resolvedView.getNamespace(),
+                                                  ChunkVersion::UNSHARDED() /* shardVersion */,
+                                                  boost::none /* databaseVersion */);
+
+            uassert(std::move(resolvedView),
+                    "Explain of a resolved view must be executed by mongos",
+                    !ShardingState::get(opCtx)->enabled() || !request.getExplain());
+
+            // Parse the resolved view into a new aggregation request.
+            auto newRequest = resolvedView.asExpandedViewAggregation(request);
+            auto newCmd = aggregation_request_helper::serializeToCommandObj(newRequest);
+
+            auto status{Status::OK()};
+            try {
+                status = runAggregate(opCtx, origNss, newRequest, newCmd, privileges, result);
+            } catch (const ExceptionForCat<ErrorCategory::StaleShardVersionError>& ex) {
+                // Since we expect the view to be UNSHARDED, if we reached to this point there are
+                // two possibilities:
+                //   1. The shard doesn't know what its shard version/state is and needs to recover
+                //      it (in which case we throw so that the shard can run recovery)
+                //   2. The collection references by the view is actually SHARDED, in which case the
+                //      router must execute it
+                if (const auto staleInfo{ex.extraInfo<StaleConfigInfo>()}) {
+                    uassert(std::move(resolvedView),
+                            "Resolved views on sharded collections must be executed by mongos",
+                            !staleInfo->getVersionWanted());
+                }
+                throw;
+            }
+
+            {
+                // Set the namespace of the curop back to the view namespace so ctx records
+                // stats on this view namespace on destruction.
+                stdx::lock_guard<Client> lk(*opCtx->getClient());
+                curOp->setNS_inlock(nss.ns());
+            }
+
+            return status;
         }
 
+        // If collectionUUID was provided, verify the collection exists and has the expected UUID.
+        checkCollectionUUIDMismatch(opCtx,
+                                    nss,
+                                    collections.getMainCollection(),
+                                    request.getCollectionUUID(),
+                                    false /* checkFeatureFlag */);
+
         invariant(collatorToUse);
-        auto pipeline = parsePipelineAndRegisterQueryStats(opCtx,
-                                                           origNss,
-                                                           request,
-                                                           ctx,
-                                                           std::move(*collatorToUse),
-                                                           uuid,
-                                                           collatorToUseMatchesDefault,
-                                                           collections,
-                                                           pipelineInvolvedNamespaces,
-                                                           liteParsedPipeline,
-                                                           nss.isCollectionlessAggregateNS(),
-                                                           resolvedView,
-                                                           origRequest);
-        expCtx = pipeline->getContext();
-
-        CurOp::get(opCtx)->beginQueryPlanningTimer();
+        expCtx = makeExpressionContext(
+            opCtx, request, std::move(*collatorToUse), uuid, collatorToUseMatchesDefault);
+
+        expCtx->startExpressionCounters();
+        auto pipeline = Pipeline::parse(request.getPipeline(), expCtx);
+        expCtx->stopExpressionCounters();
 
         if (!request.getAllowDiskUse().value_or(true)) {
             allowDiskUseFalseCounter.increment();
@@ -1126,9 +982,6 @@ Status runAggregate(OperationContext* opCtx,
     // cursor manager. The global cursor manager does not deliver invalidations or kill
     // notifications; the underlying PlanExecutor(s) used by the pipeline will be receiving
     // invalidations and kill notifications themselves, not the cursor we create here.
-    hangAfterCreatingAggregationPlan.executeIf(
-        [](const auto&) { hangAfterCreatingAggregationPlan.pauseWhileSet(); },
-        [&](const BSONObj& data) { return uuid && UUID::parse(data["uuid"]) == *uuid; });
 
     std::vector<ClientCursorPin> pins;
     std::vector<ClientCursor*> cursors;
@@ -1139,7 +992,6 @@ Status runAggregate(OperationContext* opCtx,
         }
     });
     for (auto&& exec : execs) {
-        // TODO SERVER-79373: Do not create a cursor if results can fit in a single batch.
         ClientCursorParams cursorParams(
             std::move(exec),
             origNss,
@@ -1183,7 +1035,6 @@ Status runAggregate(OperationContext* opCtx,
                                    cmdObj,
                                    &bodyBuilder);
         }
-        collectQueryStatsMongod(opCtx, std::move(curOp->debug().queryStatsInfo.key));
     } else {
         // Cursor must be specified, if explain is not.
         const bool keepCursor = handleCursorCommand(
@@ -1196,15 +1047,13 @@ Status runAggregate(OperationContext* opCtx,
         PlanSummaryStats stats;
         planExplainer.getSummaryStats(&stats);
         curOp->debug().setPlanSummaryMetrics(stats);
-        curOp->setEndOfOpMetrics(stats.nReturned);
+        curOp->debug().nreturned = stats.nReturned;
 
-        collectQueryStatsMongod(opCtx, pins[0]);
-
-        // For an optimized away pipeline, signal the cache that a query operation has
-        // completed. For normal pipelines this is done in DocumentSourceCursor.
+        // For an optimized away pipeline, signal the cache that a query operation has completed.
+        // For normal pipelines this is done in DocumentSourceCursor.
         if (ctx) {
-            // Due to yielding, the collection pointers saved in MultipleCollectionAccessor
-            // might have become invalid. We will need to refresh them here.
+            // Due to yielding, the collection pointers saved in MultipleCollectionAccessor might
+            // have become invalid. We will need to refresh them here.
             collections = MultipleCollectionAccessor(opCtx,
                                                      &ctx->getCollection(),
                                                      ctx->getNss(),
@@ -1227,11 +1076,10 @@ Status runAggregate(OperationContext* opCtx,
         }
     }
 
-    // The aggregation pipeline may change the namespace of the curop and we need to set it back
-    // to the original namespace to correctly report command stats. One example when the
-    // namespace can be changed is when the pipeline contains an $out stage, which executes an
-    // internal command to create a temp collection, changing the curop namespace to the name of
-    // this temp collection.
+    // The aggregation pipeline may change the namespace of the curop and we need to set it back to
+    // the original namespace to correctly report command stats. One example when the namespace can
+    // be changed is when the pipeline contains an $out stage, which executes an internal command to
+    // create a temp collection, changing the curop namespace to the name of this temp collection.
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
         curOp->setNS_inlock(origNss.ns());
@@ -1239,4 +1087,5 @@ Status runAggregate(OperationContext* opCtx,
 
     return Status::OK();
 }
+
 }  // namespace mongo
diff --git a/src/mongo/db/commands/run_aggregate.h b/src/mongo/db/commands/run_aggregate.h
index ba73245bdbf..b61538fb93d 100644
--- a/src/mongo/db/commands/run_aggregate.h
+++ b/src/mongo/db/commands/run_aggregate.h
@@ -49,30 +49,25 @@ namespace mongo {
  * 'privileges' contains the privileges that were required to run this aggregation, to be used later
  * for re-checking privileges for getMore commands.
  *
- * If the query over a view that's already been resolved, the resolved view and the original
- * user-provided request both must be provided.
- *
  * On success, fills out 'result' with the command response.
  */
 Status runAggregate(OperationContext* opCtx,
+                    const NamespaceString& nss,
                     AggregateCommandRequest& request,
                     const LiteParsedPipeline& liteParsedPipeline,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result,
-                    boost::optional<const ResolvedView&> resolvedView = boost::none,
-                    boost::optional<const AggregateCommandRequest&> origRequest = boost::none);
+                    rpc::ReplyBuilderInterface* result);
 
 /**
  * Convenience version that internally constructs the LiteParsedPipeline.
  */
 Status runAggregate(OperationContext* opCtx,
+                    const NamespaceString& nss,
                     AggregateCommandRequest& request,
                     const BSONObj& cmdObj,
                     const PrivilegeVector& privileges,
-                    rpc::ReplyBuilderInterface* result,
-                    boost::optional<const ResolvedView&> resolvedView = boost::none,
-                    boost::optional<const AggregateCommandRequest&> origRequest = boost::none);
+                    rpc::ReplyBuilderInterface* result);
 
 /**
  * Tracks explicit use of allowDiskUse:false with find and aggregate commands.
diff --git a/src/mongo/db/commands/server_status_metric.h b/src/mongo/db/commands/server_status_metric.h
index 423c49e2f5f..ff546527a8d 100644
--- a/src/mongo/db/commands/server_status_metric.h
+++ b/src/mongo/db/commands/server_status_metric.h
@@ -32,13 +32,8 @@
 #include <string>
 
 #include "mongo/db/jsobj.h"
-#include "mongo/platform/atomic_word.h"
 
 namespace mongo {
-class Atomic64Metric;
-
-template <>
-struct BSONObjAppendFormat<Atomic64Metric> : FormatKind<NumberLong> {};
 
 class ServerStatusMetric {
 public:
@@ -89,36 +84,4 @@ public:
 private:
     const T* _t;
 };
-
-/**
- * Atomic wrapper for long long type for Metrics.  This is for values which are set rather than
- * just incremented or decremented; if you want a counter, use Counter64.
- */
-class Atomic64Metric {
-public:
-    /** Set _value to the max of the current or newMax. */
-    void setIfMax(long long newMax) {
-        /*  Note: compareAndSwap will load into val most recent value. */
-        for (long long val = _value.load(); val < newMax && !_value.compareAndSwap(&val, newMax);) {
-        }
-    }
-
-    /** store val into value. */
-    void set(long long val) {
-        _value.storeRelaxed(val);
-    }
-
-    /** Return the current value. */
-    long long get() const {
-        return _value.loadRelaxed();
-    }
-
-    /** TODO: SERVER-73806 Avoid implicit conversion to long long */
-    operator long long() const {
-        return get();
-    }
-
-private:
-    mongo::AtomicWord<long long> _value;
-};
 }  // namespace mongo
diff --git a/src/mongo/db/commands/set_cluster_parameter_command.cpp b/src/mongo/db/commands/set_cluster_parameter_command.cpp
index 2049b3043e5..005f17d0c06 100644
--- a/src/mongo/db/commands/set_cluster_parameter_command.cpp
+++ b/src/mongo/db/commands/set_cluster_parameter_command.cpp
@@ -74,11 +74,6 @@ public:
                     (serverGlobalParams.clusterRole == ClusterRole::None));
 
             FixedFCVRegion fcvRegion(opCtx);
-            uassert(ErrorCodes::UnknownFeatureCompatibilityVersion,
-                    "FCV is not yet initialized, retry the command after FCV initialization has "
-                    "completed",
-                    serverGlobalParams.featureCompatibility.isVersionInitialized());
-
             uassert(
                 ErrorCodes::IllegalOperation,
                 "Cannot set cluster parameter, gFeatureFlagClusterWideConfig is not enabled",
diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
index 7462b287ed8..73933d1abe2 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -84,7 +84,6 @@
 #include "mongo/db/session_catalog.h"
 #include "mongo/db/session_catalog_mongod.h"
 #include "mongo/db/session_txn_record_gen.h"
-#include "mongo/db/storage/storage_parameters_gen.h"
 #include "mongo/db/timeseries/timeseries_index_schema_conversion_functions.h"
 #include "mongo/db/vector_clock.h"
 #include "mongo/idl/cluster_server_parameter_gen.h"
@@ -466,7 +465,7 @@ public:
                 if (actualVersion > requestedVersion &&
                     !feature_flags::gOrphanTracking.isEnabledOnVersion(requestedVersion)) {
                     BalancerStatsRegistry::get(opCtx)->terminate();
-                    ScopedRangeDeleterLock rangeDeleterLock(opCtx, LockMode::MODE_X);
+                    ScopedRangeDeleterLock rangeDeleterLock(opCtx);
                     clearOrphanCountersFromRangeDeletionTasks(opCtx);
                 }
 
@@ -631,6 +630,14 @@ private:
                     tenantDbName,
                     MODE_X,
                     [&](const CollectionPtr& collection) {
+                        if (collection->getTimeseriesBucketsMayHaveMixedSchemaData()) {
+                            // The catalog entry flag has already been added. This can happen if the
+                            // upgrade process was interrupted and is being run again, or if there
+                            // was a time-series collection created during the upgrade. The upgrade
+                            // process cannot be aborted at this point.
+                            return true;
+                        }
+
                         NamespaceStringOrUUID nsOrUUID(dbName, collection->uuid());
                         CollMod collModCmd(collection->ns());
                         BSONObjBuilder unusedBuilder;
@@ -743,46 +750,16 @@ private:
                 const auto& dbName = tenantDbName.dbName();
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
-                    opCtx, tenantDbName, MODE_X, [&](const CollectionPtr& collection) {
-                        const auto collNs = collection->getTimeseriesOptions()
-                            ? collection->ns().getTimeseriesViewNamespace()
-                            : collection->ns();
-                        auto indexCatalog = collection->getIndexCatalog();
+                    opCtx,
+                    tenantDbName,
+                    MODE_X,
+                    [&](const CollectionPtr& collection) {
+                        invariant(collection->getTimeseriesOptions());
 
+                        auto indexCatalog = collection->getIndexCatalog();
                         auto indexIt = indexCatalog->getIndexIterator(
-                            opCtx,
-                            IndexCatalog::InclusionPolicy::kReady |
-                                IndexCatalog::InclusionPolicy::kUnfinished |
-                                IndexCatalog::InclusionPolicy::kFrozen);
-                        while (indexIt->more()) {
-                            auto indexEntry = indexIt->next();
-                            if (auto filter = indexEntry->getFilterExpression()) {
-                                auto status = IndexCatalogImpl::checkValidFilterExpressions(
-                                    filter,
-                                    /*timeseriesMetricIndexesFeatureFlagEnabled*/ false);
-                                uassert(ErrorCodes::CannotDowngrade,
-                                        str::stream()
-                                            << "Cannot downgrade the cluster when there are "
-                                               "secondary indexes with partial filter expressions "
-                                               "that contain $in/$or/$geoWithin or an $and that is "
-                                               "not top level. Drop all indexes containing these "
-                                               "partial filter elements before downgrading. First "
-                                               "detected incompatible index name: '"
-                                            << indexEntry->descriptor()->indexName()
-                                            << "' on collection: '" << collNs << "'",
-                                        status.isOK());
-                            }
-                        }
+                            opCtx, /*includeUnfinishedIndexes=*/true);
 
-                        if (!collection->getTimeseriesOptions()) {
-                            return true;
-                        }
-
-                        indexIt = indexCatalog->getIndexIterator(
-                            opCtx,
-                            IndexCatalog::InclusionPolicy::kReady |
-                                IndexCatalog::InclusionPolicy::kUnfinished |
-                                IndexCatalog::InclusionPolicy::kFrozen);
                         while (indexIt->more()) {
                             auto indexEntry = indexIt->next();
                             // Secondary indexes on time-series measurements are only supported
@@ -793,17 +770,34 @@ private:
                                 ErrorCodes::CannotDowngrade,
                                 str::stream()
                                     << "Cannot downgrade the cluster when there are secondary "
-                                       "indexes on time-series measurements present, or when "
-                                       "there are partial indexes on a time-series collection. "
-                                       "Drop "
-                                       "all secondary indexes on time-series measurements, and all "
+                                       "indexes on time-series measurements present, or when there "
+                                       "are partial indexes on a time-series collection. Drop all "
+                                       "secondary indexes on time-series measurements, and all "
                                        "partial indexes on time-series collections, before "
                                        "downgrading. First detected incompatible index name: '"
                                     << indexEntry->descriptor()->indexName() << "' on collection: '"
-                                    << collNs << "'",
+                                    << collection->ns().getTimeseriesViewNamespace() << "'",
                                 timeseries::isBucketsIndexSpecCompatibleForDowngrade(
                                     *collection->getTimeseriesOptions(),
                                     indexEntry->descriptor()->infoObj()));
+
+                            if (auto filter = indexEntry->getFilterExpression()) {
+                                auto status = IndexCatalogImpl::checkValidFilterExpressions(
+                                    filter,
+                                    /*timeseriesMetricIndexesFeatureFlagEnabled*/ false);
+                                uassert(ErrorCodes::CannotDowngrade,
+                                        str::stream()
+                                            << "Cannot downgrade the cluster when there are "
+                                               "secondary indexes with partial filter expressions "
+                                               "that contain $in/$or/$geoWithin or an $and that is "
+                                               "not top level. Drop all indexes containing these "
+                                               "partial filter elements before downgrading. First "
+                                               "detected incompatible index name: '"
+                                            << indexEntry->descriptor()->indexName()
+                                            << "' on collection: '"
+                                            << collection->ns().getTimeseriesViewNamespace() << "'",
+                                        status.isOK());
+                            }
                         }
 
                         if (!collection->getTimeseriesBucketsMayHaveMixedSchemaData()) {
@@ -830,6 +824,9 @@ private:
                         }
 
                         return true;
+                    },
+                    [&](const CollectionPtr& collection) {
+                        return collection->getTimeseriesOptions() != boost::none;
                     });
             }
         }
@@ -871,10 +868,7 @@ private:
                     opCtx, tenantDbName, MODE_X, [&](const CollectionPtr& collection) {
                         auto indexCatalog = collection->getIndexCatalog();
                         auto indexIt = indexCatalog->getIndexIterator(
-                            opCtx,
-                            IndexCatalog::InclusionPolicy::kReady |
-                                IndexCatalog::InclusionPolicy::kUnfinished |
-                                IndexCatalog::InclusionPolicy::kFrozen);
+                            opCtx, true /* includeUnfinishedIndexes */);
                         while (indexIt->more()) {
                             auto indexEntry = indexIt->next();
                             uassert(
diff --git a/src/mongo/db/commands/set_profiling_filter_globally_cmd.cpp b/src/mongo/db/commands/set_profiling_filter_globally_cmd.cpp
deleted file mode 100644
index 287b4331831..00000000000
--- a/src/mongo/db/commands/set_profiling_filter_globally_cmd.cpp
+++ /dev/null
@@ -1,112 +0,0 @@
-/**
- *    Copyright (C) 2023-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the Server Side Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
-
-#include "mongo/db/commands/set_profiling_filter_globally_cmd.h"
-#include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/commands/profile_gen.h"
-#include "mongo/db/profile_filter_impl.h"
-#include "mongo/logv2/log.h"
-
-namespace mongo {
-
-Status SetProfilingFilterGloballyCmd::checkAuthForCommand(Client* client,
-                                                          const std::string& dbName,
-                                                          const BSONObj& cmdObj) const {
-    AuthorizationSession* authSession = AuthorizationSession::get(client);
-    return authSession->isAuthorizedForActionsOnResource(ResourcePattern::forAnyNormalResource(),
-                                                         ActionType::enableProfiler)
-        ? Status::OK()
-        : Status(ErrorCodes::Unauthorized, "unauthorized");
-}
-
-bool SetProfilingFilterGloballyCmd::run(OperationContext* opCtx,
-                                        const std::string& dbName,
-                                        const BSONObj& cmdObj,
-                                        BSONObjBuilder& result) {
-    uassert(7283301,
-            str::stream() << "setProfilingFilterGlobally command requires query knob to be enabled",
-            internalQueryGlobalProfilingFilter.load());
-
-    auto request = SetProfilingFilterGloballyCmdRequest::parse(
-        IDLParserErrorContext("setProfilingFilterGlobally"), cmdObj);
-
-    // Save off the old global default setting so that we can log it and return in the result.
-    auto oldDefault = ProfileFilter::getDefault();
-    auto newDefault = [&request] {
-        const auto& filterOrUnset = request.getFilter();
-        if (auto filter = filterOrUnset.obj) {
-            return std::make_shared<ProfileFilterImpl>(*filter);
-        }
-        return std::shared_ptr<ProfileFilterImpl>(nullptr);
-    }();
-
-    // Update the global default.
-    // Note that since this is not done atomically with the collection catalog write, there is a
-    // minor race condition where queries on some databases see the new global default while queries
-    // on other databases see old database-specific settings. This is a temporary state and
-    // shouldn't impact much in practice. We also don't have to worry about races with database
-    // creation, since the global default gets picked up dynamically by queries instead of being
-    // explicitly stored for new databases.
-    ProfileFilter::setDefault(newDefault);
-
-    // Writing to the CollectionCatalog requires holding the Global lock to avoid concurrent races
-    // with BatchedCollectionCatalogWriter.
-    boost::optional<Lock::GlobalLock> lk;
-    if (!opCtx->lockState()->isNoop()) {
-        // Taking the lock is only meaningful if we're in a mongod. Other mongos do not have a
-        // notion of collections.
-        lk.emplace(opCtx, MODE_IX);
-    }
-
-    // Update all existing database settings.
-    CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-        catalog.setAllDatabaseProfileFilters(newDefault);
-    });
-
-    // Capture the old setting in the result object.
-    if (oldDefault) {
-        result.append("was", oldDefault->serialize());
-    } else {
-        result.append("was", "none");
-    }
-
-    // Log the change made to server's global profiling settings.
-    LOGV2(72832,
-          "Profiler settings changed globally",
-          "from"_attr = oldDefault ? BSON("filter" << oldDefault->serialize())
-                                   : BSON("filter"
-                                          << "none"),
-          "to"_attr = newDefault ? BSON("filter" << newDefault->serialize())
-                                 : BSON("filter"
-                                        << "none"));
-    return true;
-}
-}  // namespace mongo
diff --git a/src/mongo/db/commands/set_profiling_filter_globally_cmd.h b/src/mongo/db/commands/set_profiling_filter_globally_cmd.h
deleted file mode 100644
index 15c36f0a9cc..00000000000
--- a/src/mongo/db/commands/set_profiling_filter_globally_cmd.h
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- *    Copyright (C) 2023-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the Server Side Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#pragma once
-
-#include "mongo/base/status.h"
-#include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/commands.h"
-
-namespace mongo {
-
-class SetProfilingFilterGloballyCmdRequest;
-
-/**
- * Command class implementing functionality for both the mongoD and mongoS
- * 'setProfilingFilterGlobally' command.
- */
-class SetProfilingFilterGloballyCmd : public BasicCommand {
-public:
-    SetProfilingFilterGloballyCmd() : BasicCommand("setProfilingFilterGlobally") {}
-
-    AllowedOnSecondary secondaryAllowed(ServiceContext*) const final {
-        return AllowedOnSecondary::kAlways;
-    }
-
-    std::string help() const final {
-        return "updates a global filter that determines which operations are eligible for "
-               "logging/profiling";
-    }
-
-    bool supportsWriteConcern(const BSONObj& cmd) const final {
-        return false;
-    }
-
-    Status checkAuthForCommand(Client* client,
-                               const std::string& dbname,
-                               const BSONObj& cmdObj) const final;
-
-    bool run(OperationContext* opCtx,
-             const std::string& dbName,
-             const BSONObj& cmdObj,
-             BSONObjBuilder& result) final;
-};
-}  // namespace mongo
diff --git a/src/mongo/db/commands/tenant_migration_donor_cmds.cpp b/src/mongo/db/commands/tenant_migration_donor_cmds.cpp
index 14fa2dcf36b..e94f865d1cb 100644
--- a/src/mongo/db/commands/tenant_migration_donor_cmds.cpp
+++ b/src/mongo/db/commands/tenant_migration_donor_cmds.cpp
@@ -187,7 +187,7 @@ public:
             auto donorService =
                 repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext())
                     ->lookupServiceByName(TenantMigrationDonorService::kServiceName);
-            auto [optionalDonor, _] = TenantMigrationDonorService::Instance::lookup(
+            auto optionalDonor = TenantMigrationDonorService::Instance::lookup(
                 opCtx, donorService, BSON("_id" << cmd.getMigrationId()));
             uassert(ErrorCodes::NoSuchTenantMigration,
                     str::stream() << "Could not find tenant migration with id "
@@ -261,7 +261,7 @@ public:
             auto donorService =
                 repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext())
                     ->lookupServiceByName(TenantMigrationDonorService::kServiceName);
-            auto [optionalDonor, _] = TenantMigrationDonorService::Instance::lookup(
+            auto optionalDonor = TenantMigrationDonorService::Instance::lookup(
                 opCtx, donorService, BSON("_id" << cmd.getMigrationId()));
 
             // If there is NoSuchTenantMigration, perform a noop write and wait for it to be
diff --git a/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp b/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp
index 505a5f73f68..e826e51dacc 100644
--- a/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp
+++ b/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp
@@ -201,7 +201,7 @@ public:
                 repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext())
                     ->lookupServiceByName(repl::TenantMigrationRecipientService::
                                               kTenantMigrationRecipientServiceName);
-            auto [instance, _] = repl::TenantMigrationRecipientService::Instance::lookup(
+            auto instance = repl::TenantMigrationRecipientService::Instance::lookup(
                 opCtx, recipientService, BSON("_id" << cmd.getMigrationId()));
             uassert(8423340, "Unknown migrationId", instance);
             (*instance)->onMemberImportedFiles(cmd.getFrom(), cmd.getSuccess(), cmd.getReason());
diff --git a/src/mongo/db/commands/user_management_commands.cpp b/src/mongo/db/commands/user_management_commands.cpp
index ff0af0d0f0d..13c5650bc98 100644
--- a/src/mongo/db/commands/user_management_commands.cpp
+++ b/src/mongo/db/commands/user_management_commands.cpp
@@ -1450,6 +1450,7 @@ UsersInfoReply CmdUMCTyped<UsersInfoCommand, UMCInfoParams>::Invocation::typedRu
                                            std::move(pipeline));
         // Impose no cursor privilege requirements, as cursor is drained internally
         uassertStatusOK(runAggregate(opCtx,
+                                     AuthorizationManager::usersCollectionNamespace,
                                      aggRequest,
                                      aggregation_request_helper::serializeToCommandObj(aggRequest),
                                      PrivilegeVector(),
diff --git a/src/mongo/db/commands/validate.cpp b/src/mongo/db/commands/validate.cpp
index 1ca15aca557..c6724076b9f 100644
--- a/src/mongo/db/commands/validate.cpp
+++ b/src/mongo/db/commands/validate.cpp
@@ -31,29 +31,15 @@
 
 #include "mongo/platform/basic.h"
 
-
-#include "mongo/base/error_codes.h"
-#include "mongo/base/status.h"
-#include "mongo/base/string_data.h"
-#include "mongo/bson/bson_validate_gen.h"
-#include "mongo/bson/bsonelement.h"
-#include "mongo/bson/bsonmisc.h"
-#include "mongo/bson/bsonobj.h"
-#include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/db/auth/action_type.h"
-#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_validation.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
-#include "mongo/db/commands/test_commands_enabled.h"
-#include "mongo/db/dbdirectclient.h"
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/scopeguard.h"
-#include "mongo/util/testing_proctor.h"
 
 namespace mongo {
 
@@ -75,79 +61,6 @@ std::set<std::string> _validationsInProgress;
 // finishes on any namespace.
 stdx::condition_variable _validationNotifier;
 
-/**
- * Creates an aggregation command with a $collStats pipeline that fetches 'storageStats' and
- * 'count'.
- */
-BSONObj makeCollStatsCommand(StringData collectionNameOnly) {
-    BSONArrayBuilder pipelineBuilder;
-    pipelineBuilder << BSON("$collStats"
-                            << BSON("storageStats" << BSONObj() << "count" << BSONObj()));
-    return BSON("aggregate" << collectionNameOnly << "pipeline" << pipelineBuilder.arr() << "cursor"
-                            << BSONObj());
-}
-
-/**
- * $collStats never returns more than a single document. If that ever changes in future, validate
- * must invariant so that the handling can be updated, but only invariant in testing environments,
- * never invariant because of debug logging in production situations.
- */
-void verifyCommandResponse(const BSONObj& collStatsResult) {
-    if (TestingProctor::instance().isEnabled()) {
-        invariant(
-            !collStatsResult.getObjectField("cursor").isEmpty() &&
-                !collStatsResult.getObjectField("cursor").getObjectField("firstBatch").isEmpty(),
-            str::stream() << "Expected a cursor to be present in the $collStats results: "
-                          << collStatsResult.toString());
-        invariant(collStatsResult.getObjectField("cursor").getIntField("id") == 0,
-                  str::stream() << "Expected cursor ID to be 0: " << collStatsResult.toString());
-    } else {
-        uassert(
-            7463202,
-            str::stream() << "Expected a cursor to be present in the $collStats results: "
-                          << collStatsResult.toString(),
-            !collStatsResult.getObjectField("cursor").isEmpty() &&
-                !collStatsResult.getObjectField("cursor").getObjectField("firstBatch").isEmpty());
-        uassert(7463203,
-                str::stream() << "Expected cursor ID to be 0: " << collStatsResult.toString(),
-                collStatsResult.getObjectField("cursor").getIntField("id") == 0);
-    }
-}
-
-/**
- * Log the $collStats results for 'nss' to provide additional debug information for validation
- * failures.
- */
-void logCollStats(OperationContext* opCtx, const NamespaceString& nss) {
-    DBDirectClient client(opCtx);
-
-    BSONObj collStatsResult;
-    try {
-        // Run $collStats via aggregation.
-        client.runCommand(nss.db().toString(),
-                          makeCollStatsCommand(nss.coll()),
-                          collStatsResult /* command return results */);
-        // Logging $collStats information is best effort. If the collection doesn't exist, for
-        // example, then the $collStats query will fail and the failure reason will be logged.
-        uassertStatusOK(getStatusFromWriteCommandReply(collStatsResult));
-        verifyCommandResponse(collStatsResult);
-
-        LOGV2_OPTIONS(7463200,
-                      logv2::LogTruncation::Disabled,
-                      "Corrupt namespace $collStats results",
-                      "namespace"_attr = nss,
-                      "collStats"_attr =
-                          collStatsResult.getObjectField("cursor").getObjectField("firstBatch"));
-    } catch (const DBException& ex) {
-        // Catch the error so that the validate error does not get overwritten by the attempt to add
-        // debug logging.
-        LOGV2_WARNING(7463201,
-                      "Failed to fetch $collStats for validation error",
-                      "namespace"_attr = nss,
-                      "error"_attr = ex.toStatus());
-    }
-}
-
 }  // namespace
 
 /**
@@ -210,7 +123,6 @@ public:
 
         const NamespaceString nss(CommandHelpers::parseNsCollectionRequired(dbname, cmdObj));
         bool background = cmdObj["background"].trueValue();
-        bool logDiagnostics = cmdObj["logDiagnostics"].trueValue();
 
         // Background validation is not supported on the ephemeralForTest storage engine due to its
         // lack of support for timestamps. Switch the mode to foreground validation instead.
@@ -342,20 +254,9 @@ public:
                 PrepareConflictBehavior::kIgnoreConflictsAllowWrites);
         }
 
-        CollectionValidation::AdditionalOptions additionalOptions;
-        additionalOptions.validationVersion = getTestCommandsEnabled()
-            ? (ValidationVersion)bsonTestValidationVersion
-            : currentValidationVersion;
-
         ValidateResults validateResults;
-        Status status = CollectionValidation::validate(opCtx,
-                                                       nss,
-                                                       mode,
-                                                       repairMode,
-                                                       additionalOptions,
-                                                       &validateResults,
-                                                       &result,
-                                                       logDiagnostics);
+        Status status =
+            CollectionValidation::validate(opCtx, nss, mode, repairMode, &validateResults, &result);
         if (!status.isOK()) {
             return CommandHelpers::appendCommandStatusNoThrow(result, status);
         }
@@ -366,7 +267,6 @@ public:
             result.append("advice",
                           "A corrupt namespace has been detected. See "
                           "http://dochub.mongodb.org/core/data-recovery for recovery steps.");
-            logCollStats(opCtx, nss);
         }
 
         return true;
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index 737bec2a1e7..c3a00ff74e6 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -149,10 +149,12 @@ public:
                         return _validateView(opCtx, view);
                     });
 
-                for (auto&& coll : collectionCatalog->range(tenantDbName)) {
+                for (auto collIt = collectionCatalog->begin(opCtx, tenantDbName);
+                     collIt != collectionCatalog->end(opCtx);
+                     ++collIt) {
                     if (!_validateNamespace(
                             opCtx,
-                            collectionCatalog->lookupNSSByUUID(opCtx, coll->uuid()).value())) {
+                            collectionCatalog->lookupNSSByUUID(opCtx, collIt.uuid().get()).get())) {
                         return;
                     }
                 }
@@ -213,10 +215,8 @@ public:
 
             // Ensure there are no unstable indexes.
             const auto* indexCatalog = collection->getIndexCatalog();
-            auto ii = indexCatalog->getIndexIterator(
-                opCtx,
-                IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
-                    IndexCatalog::InclusionPolicy::kFrozen);
+            std::unique_ptr<IndexCatalog::IndexIterator> ii =
+                indexCatalog->getIndexIterator(opCtx, true /* includeUnfinishedIndexes */);
             while (ii->more()) {
                 // Check if the index is allowed in API version 1.
                 const IndexDescriptor* desc = ii->next()->descriptor();
diff --git a/src/mongo/db/commands/write_commands.cpp b/src/mongo/db/commands/write_commands.cpp
index 7196c905f1b..0254baca47d 100644
--- a/src/mongo/db/commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands.cpp
@@ -30,13 +30,11 @@
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kDefault
 
 #include "mongo/base/checked_cast.h"
-#include "mongo/base/error_codes.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/bson/mutable/document.h"
 #include "mongo/bson/mutable/element.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_operation_source.h"
-#include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/client.h"
@@ -75,7 +73,6 @@
 #include "mongo/db/timeseries/bucket_catalog.h"
 #include "mongo/db/timeseries/bucket_compression.h"
 #include "mongo/db/timeseries/timeseries_constants.h"
-#include "mongo/db/timeseries/timeseries_extended_range.h"
 #include "mongo/db/timeseries/timeseries_options.h"
 #include "mongo/db/timeseries/timeseries_stats.h"
 #include "mongo/db/transaction_participant.h"
@@ -139,26 +136,9 @@ bool isTimeseries(OperationContext* opCtx, const Request& request) {
     // collection does not yet exist, this check may return false unnecessarily. As a result, an
     // insert attempt into the time-series namespace will either succeed or fail, depending on who
     // wins the race.
-    // Hold reference to the catalog for collection lookup without locks to be safe.
-    auto catalog = CollectionCatalog::get(opCtx);
-    auto coll = catalog->lookupCollectionByNamespace(opCtx, bucketNss);
-    if (!coll) {
-        return false;
-    }
-
-    if (auto options = coll->getTimeseriesOptions()) {
-        uassert(ErrorCodes::InvalidOptions,
-                "Time-series buckets collection is not clustered",
-                coll->isClustered());
-
-        uassert(ErrorCodes::InvalidOptions,
-                "Time-series buckets collection is missing bucketMaxSpanSeconds",
-                options->getBucketMaxSpanSeconds());
-
-        return true;
-    }
-
-    return false;
+    return CollectionCatalog::get(opCtx)
+        ->lookupCollectionByNamespaceForRead(opCtx, bucketNss)
+        .get();
 }
 
 NamespaceString makeTimeseriesBucketsNamespace(const NamespaceString& nss) {
@@ -344,22 +324,6 @@ boost::optional<std::pair<Status, bool>> checkFailUnorderedTimeseriesInsertFailP
     return boost::none;
 }
 
-boost::optional<write_ops::WriteError> generateErrorNoTenantMigration(OperationContext* opCtx,
-                                                                      const Status& status,
-                                                                      int index,
-                                                                      size_t numErrors) {
-    constexpr size_t kMaxErrorReasonsToReport = 1;
-    constexpr size_t kMaxErrorSizeToReportAfterMaxReasonsReached = 1024 * 1024;
-
-    if (numErrors > kMaxErrorReasonsToReport) {
-        size_t errorSize = status.reason().size();
-        if (errorSize > kMaxErrorSizeToReportAfterMaxReasonsReached)
-            return write_ops::WriteError(index, status.withReason(""));
-    }
-
-    return write_ops::WriteError(index, status);
-}
-
 boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx,
                                                      const Status& status,
                                                      int index,
@@ -368,16 +332,18 @@ boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx,
         return boost::none;
     }
 
+    boost::optional<Status> overwrittenStatus;
+
     if (status == ErrorCodes::TenantMigrationConflict) {
         hangWriteBeforeWaitingForMigrationDecision.pauseWhileSet(opCtx);
 
-        Status overwrittenStatus =
-            tenant_migration_access_blocker::handleTenantMigrationConflict(opCtx, status);
+        overwrittenStatus.emplace(
+            tenant_migration_access_blocker::handleTenantMigrationConflict(opCtx, status));
 
         // Interruption errors encountered during batch execution fail the entire batch, so throw on
         // such errors here for consistency.
-        if (ErrorCodes::isInterruption(overwrittenStatus)) {
-            uassertStatusOK(overwrittenStatus);
+        if (ErrorCodes::isInterruption(*overwrittenStatus)) {
+            uassertStatusOK(*overwrittenStatus);
         }
 
         // Tenant migration errors, similarly to migration errors consume too much space in the
@@ -385,13 +351,25 @@ boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx,
         // 'handleTenantMigrationConflict' above replaces the original status, we need to manually
         // truncate the new reason if the original 'status' was also truncated.
         if (status.reason().empty()) {
-            overwrittenStatus = overwrittenStatus.withReason("");
+            overwrittenStatus = overwrittenStatus->withReason("");
         }
+    }
 
-        return generateErrorNoTenantMigration(opCtx, overwrittenStatus, index, numErrors);
+    constexpr size_t kMaxErrorReasonsToReport = 1;
+    constexpr size_t kMaxErrorSizeToReportAfterMaxReasonsReached = 1024 * 1024;
+
+    if (numErrors > kMaxErrorReasonsToReport) {
+        size_t errorSize =
+            overwrittenStatus ? overwrittenStatus->reason().size() : status.reason().size();
+        if (errorSize > kMaxErrorSizeToReportAfterMaxReasonsReached)
+            overwrittenStatus =
+                overwrittenStatus ? overwrittenStatus->withReason("") : status.withReason("");
     }
 
-    return generateErrorNoTenantMigration(opCtx, status, index, numErrors);
+    if (overwrittenStatus)
+        return write_ops::WriteError(index, std::move(*overwrittenStatus));
+    else
+        return write_ops::WriteError(index, status);
 }
 
 template <typename T>
@@ -435,7 +413,6 @@ void populateReply(OperationContext* opCtx,
         const auto& lastResult = result.results.back();
 
         if (lastResult == ErrorCodes::StaleDbVersion ||
-            lastResult == ErrorCodes::ShardCannotRefreshDueToLocksHeld ||
             ErrorCodes::isStaleShardVersionError(lastResult.getStatus()) ||
             ErrorCodes::isTenantMigrationError(lastResult.getStatus())) {
             // For ordered:false commands we need to duplicate these error results for all ops
@@ -550,11 +527,6 @@ public:
         }
 
         write_ops::InsertCommandReply typedRun(OperationContext* opCtx) final try {
-            // On debug builds, verify that the estimated size of the insert command is at least as
-            // large as the size of the actual, serialized insert command. This ensures that the
-            // logic which estimates the size of insert commands is correct.
-            dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest()));
-
             transactionChecks(opCtx, ns());
 
             if (request().getEncryptionInformation().has_value() &&
@@ -726,16 +698,16 @@ public:
                 OperationSource::kTimeseriesInsert));
         }
 
-        void _performTimeseriesBucketCompression(
+        TimeseriesSingleWriteResult _performTimeseriesBucketCompression(
             OperationContext* opCtx, const BucketCatalog::ClosedBucket& closedBucket) const {
             if (!feature_flags::gTimeseriesBucketCompression.isEnabled(
                     serverGlobalParams.featureCompatibility)) {
-                return;
+                return {SingleWriteResult(), true};
             }
 
             // Buckets with just a single measurement is not worth compressing.
             if (closedBucket.numMeasurements <= 1) {
-                return;
+                return {SingleWriteResult(), true};
             }
 
             bool validateCompression = gValidateTimeseriesCompression.load();
@@ -773,8 +745,8 @@ public:
 
             auto compressionOp =
                 _makeTimeseriesCompressionOp(opCtx, closedBucket.bucketId, bucketCompressionFunc);
-            auto result = _getTimeseriesSingleWriteResult(write_ops_exec::performUpdates(
-                opCtx, compressionOp, OperationSource::kTimeseriesBucketCompression));
+            auto result = _getTimeseriesSingleWriteResult(
+                write_ops_exec::performUpdates(opCtx, compressionOp, OperationSource::kStandard));
 
             // Report stats, if we fail before running the transform function then just skip
             // reporting.
@@ -789,6 +761,8 @@ public:
                     stats.onBucketClosed(*beforeSize, compressionStats);
                 }
             }
+
+            return result;
         }
 
         /**
@@ -802,8 +776,7 @@ public:
                                      std::vector<write_ops::WriteError>* errors,
                                      boost::optional<repl::OpTime>* opTime,
                                      boost::optional<OID>* electionId,
-                                     std::vector<size_t>* docsToRetry,
-                                     absl::flat_hash_map<int, int>& retryAttemptsForDup) const try {
+                                     std::vector<size_t>* docsToRetry) const try {
             auto& bucketCatalog = BucketCatalog::get(opCtx);
 
             auto metadata = bucketCatalog.getMetadata(batch->bucket());
@@ -823,18 +796,9 @@ public:
                     _performTimeseriesInsert(opCtx, batch, metadata, std::move(stmtIds));
                 if (auto error =
                         generateError(opCtx, output.result, start + index, errors->size())) {
-                    bool canContinue = output.canContinue;
-                    // Automatically attempts to retry on DuplicateKey error.
-                    if (error->getStatus().code() == ErrorCodes::DuplicateKey &&
-                        retryAttemptsForDup[index]++ <
-                            gTimeseriesInsertMaxRetriesOnDuplicates.load()) {
-                        docsToRetry->push_back(index);
-                        canContinue = true;
-                    } else {
-                        errors->emplace_back(std::move(*error));
-                    }
-                    BucketCatalog::get(opCtx).abort(batch, output.result.getStatus());
-                    return canContinue;
+                    errors->emplace_back(std::move(*error));
+                    bucketCatalog.abort(batch, output.result.getStatus());
+                    return output.canContinue;
                 }
 
                 invariant(output.result.getValue().getN() == 1,
@@ -864,7 +828,12 @@ public:
 
             if (closedBucket) {
                 // If this write closed a bucket, compress the bucket
-                _performTimeseriesBucketCompression(opCtx, *closedBucket);
+                auto output = _performTimeseriesBucketCompression(opCtx, *closedBucket);
+                if (auto error =
+                        generateError(opCtx, output.result, start + index, errors->size())) {
+                    errors->emplace_back(std::move(*error));
+                    return output.canContinue;
+                }
             }
             return true;
         } catch (const DBException& ex) {
@@ -872,12 +841,19 @@ public:
             throw;
         }
 
-        bool _commitTimeseriesBucketsAtomically(OperationContext* opCtx,
-                                                TimeseriesBatches* batches,
-                                                TimeseriesStmtIds&& stmtIds,
-                                                std::vector<write_ops::WriteError>* errors,
-                                                boost::optional<repl::OpTime>* opTime,
-                                                boost::optional<OID>* electionId) const {
+        enum struct TimeseriesAtomicWriteResult {
+            kSuccess,
+            kContinuableError,
+            kNonContinuableError,
+        };
+
+        TimeseriesAtomicWriteResult _commitTimeseriesBucketsAtomically(
+            OperationContext* opCtx,
+            TimeseriesBatches* batches,
+            TimeseriesStmtIds&& stmtIds,
+            std::vector<write_ops::WriteError>* errors,
+            boost::optional<repl::OpTime>* opTime,
+            boost::optional<OID>* electionId) const {
             auto& bucketCatalog = BucketCatalog::get(opCtx);
 
             std::vector<std::reference_wrapper<std::shared_ptr<BucketCatalog::WriteBatch>>>
@@ -890,7 +866,7 @@ public:
             }
 
             if (batchesToCommit.empty()) {
-                return true;
+                return TimeseriesAtomicWriteResult::kSuccess;
             }
 
             // Sort by bucket so that preparing the commit for each batch cannot deadlock.
@@ -916,7 +892,7 @@ public:
                     auto prepareCommitStatus = bucketCatalog.prepareCommit(batch);
                     if (!prepareCommitStatus.isOK()) {
                         abortStatus = prepareCommitStatus;
-                        return false;
+                        return TimeseriesAtomicWriteResult::kContinuableError;
                     }
 
                     if (batch.get()->numPreviouslyCommittedMeasurements() == 0) {
@@ -933,26 +909,33 @@ public:
                 auto result =
                     write_ops_exec::performAtomicTimeseriesWrites(opCtx, insertOps, updateOps);
                 if (!result.isOK()) {
-                    if (result.code() == ErrorCodes::DuplicateKey) {
-                        BucketCatalog::get(opCtx).resetBucketOIDCounter();
-                    }
                     abortStatus = result;
-                    return false;
+                    return TimeseriesAtomicWriteResult::kContinuableError;
                 }
 
                 getOpTimeAndElectionId(opCtx, opTime, electionId);
 
+                bool compressClosedBuckets = true;
                 for (auto batch : batchesToCommit) {
                     auto closedBucket = bucketCatalog.finish(
                         batch, BucketCatalog::CommitInfo{*opTime, *electionId});
                     batch.get().reset();
 
-                    if (!closedBucket) {
+                    if (!closedBucket || !compressClosedBuckets) {
                         continue;
                     }
 
                     // If this write closed a bucket, compress the bucket
-                    _performTimeseriesBucketCompression(opCtx, *closedBucket);
+                    auto ret = _performTimeseriesBucketCompression(opCtx, *closedBucket);
+                    if (!ret.result.isOK()) {
+                        // Don't try to compress any other buckets if we fail. We're not allowed to
+                        // do more write operations.
+                        compressClosedBuckets = false;
+                    }
+                    if (!ret.canContinue) {
+                        abortStatus = ret.result.getStatus();
+                        return TimeseriesAtomicWriteResult::kNonContinuableError;
+                    }
                 }
             } catch (const DBException& ex) {
                 abortStatus = ex.toStatus();
@@ -960,7 +943,7 @@ public:
             }
 
             batchGuard.dismiss();
-            return true;
+            return TimeseriesAtomicWriteResult::kSuccess;
         }
 
         // For sharded time-series collections, we need to use the granularity from the config
@@ -985,7 +968,10 @@ public:
             }
         }
 
-        std::tuple<TimeseriesBatches, TimeseriesStmtIds, size_t /* numInserted */>
+        std::tuple<TimeseriesBatches,
+                   TimeseriesStmtIds,
+                   size_t /* numInserted */,
+                   bool /* canContinue */>
         _insertIntoBucketCatalog(OperationContext* opCtx,
                                  size_t start,
                                  size_t numDocs,
@@ -1025,6 +1011,7 @@ public:
 
             TimeseriesBatches batches;
             TimeseriesStmtIds stmtIds;
+            bool canContinue = true;
 
             auto insert = [&](size_t index) {
                 invariant(start + index < request().getDocuments().size());
@@ -1070,63 +1057,58 @@ public:
                 // If this insert closed buckets, rewrite to be a compressed column. If we cannot
                 // perform write operations at this point the bucket will be left uncompressed.
                 for (const auto& closedBucket : result.getValue().closedBuckets) {
+                    if (!canContinue) {
+                        break;
+                    }
+
                     // If this write closed a bucket, compress the bucket
-                    _performTimeseriesBucketCompression(opCtx, closedBucket);
+                    auto ret = _performTimeseriesBucketCompression(opCtx, closedBucket);
+                    if (auto error =
+                            generateError(opCtx, ret.result, start + index, errors->size())) {
+                        // Bucket compression only fail when we may not try to perform any other
+                        // write operation. When handleError() inside write_ops_exec.cpp return
+                        // false.
+                        errors->emplace_back(std::move(*error));
+                        canContinue = false;
+                        return false;
+                    }
+                    canContinue = ret.canContinue;
                 }
 
                 return true;
             };
 
-            try {
-                if (!indices.empty()) {
-                    std::for_each(indices.begin(), indices.end(), insert);
-                } else {
-                    for (size_t i = 0; i < numDocs; i++) {
-                        if (!insert(i) && request().getOrdered()) {
-                            return {std::move(batches), std::move(stmtIds), i};
-                        }
+            if (!indices.empty()) {
+                std::for_each(indices.begin(), indices.end(), insert);
+            } else {
+                for (size_t i = 0; i < numDocs; i++) {
+                    if (!insert(i) && request().getOrdered()) {
+                        return {std::move(batches), std::move(stmtIds), i, canContinue};
                     }
                 }
-            } catch (const DBException& ex) {
-                // Exception insert into bucket catalog, append error and wait for all batches that
-                // we've already managed to write into to commit or abort. We need to wait here as
-                // pointers to memory owned by this command is stored in the WriteBatch(es). This
-                // ensures that no other thread may try to access this memory after this command has
-                // been torn down due to the exception.
-
-                boost::optional<repl::OpTime> opTime;
-                boost::optional<OID> electionId;
-                std::vector<size_t> docsToRetry;
-                errors->emplace_back(
-                    *generateErrorNoTenantMigration(opCtx, ex.toStatus(), 0, errors->size()));
-
-                _getTimeseriesBatchResultsNoTenantMigration(
-                    opCtx, batches, 0, -1, false, errors, &opTime, &electionId, &docsToRetry);
-                throw;
             }
 
-
-            return {std::move(batches), std::move(stmtIds), request().getDocuments().size()};
+            return {std::move(batches),
+                    std::move(stmtIds),
+                    request().getDocuments().size(),
+                    canContinue};
         }
 
-        template <typename ErrorGenerator>
-        void _getTimeseriesBatchResultsBase(ErrorGenerator&& errorGenerator,
-                                            OperationContext* opCtx,
-                                            const TimeseriesBatches& batches,
-                                            int64_t start,
-                                            int64_t indexOfLastProcessedBatch,
-                                            bool canContinue,
-                                            std::vector<write_ops::WriteError>* errors,
-                                            boost::optional<repl::OpTime>* opTime,
-                                            boost::optional<OID>* electionId,
-                                            std::vector<size_t>* docsToRetry = nullptr) const {
+        void _getTimeseriesBatchResults(OperationContext* opCtx,
+                                        const TimeseriesBatches& batches,
+                                        size_t start,
+                                        size_t indexOfLastProcessedBatch,
+                                        bool canContinue,
+                                        std::vector<write_ops::WriteError>* errors,
+                                        boost::optional<repl::OpTime>* opTime,
+                                        boost::optional<OID>* electionId,
+                                        std::vector<size_t>* docsToRetry = nullptr) const {
             boost::optional<write_ops::WriteError> lastError;
             if (!errors->empty()) {
                 lastError = errors->back();
             }
-            invariant(indexOfLastProcessedBatch == (int64_t)batches.size() || lastError);
 
-            for (int64_t itr = 0, size = batches.size(); itr < size; ++itr) {
+            for (size_t itr = 0; itr < batches.size(); ++itr) {
                 const auto& [batch, index] = batches[itr];
                 if (!batch) {
                     continue;
@@ -1142,11 +1124,11 @@ public:
 
                 auto swCommitInfo = batch->getResult();
                 if (swCommitInfo.getStatus() == ErrorCodes::TimeseriesBucketCleared) {
-                    invariant(docsToRetry, "the 'docsToRetry' cannot be null");
+                    tassert(6023102, "the 'docsToRetry' cannot be null", docsToRetry);
                     docsToRetry->push_back(index);
                     continue;
                 }
-                if (auto error = errorGenerator(
+                if (auto error = generateError(
                         opCtx, swCommitInfo.getStatus(), start + index, errors->size())) {
                     errors->emplace_back(std::move(*error));
                     continue;
@@ -1171,76 +1153,30 @@ public:
             }
         }
 
-        void _getTimeseriesBatchResults(OperationContext* opCtx,
-                                        const TimeseriesBatches& batches,
-                                        int64_t start,
-                                        int64_t indexOfLastProcessedBatch,
-                                        bool canContinue,
-                                        std::vector<write_ops::WriteError>* errors,
-                                        boost::optional<repl::OpTime>* opTime,
-                                        boost::optional<OID>* electionId,
-                                        std::vector<size_t>* docsToRetry = nullptr) const {
-            auto errorGenerator =
-                [](OperationContext* opCtx, const Status& status, int index, size_t numErrors) {
-                    return generateError(opCtx, status, index, numErrors);
-                };
-            _getTimeseriesBatchResultsBase(errorGenerator,
-                                           opCtx,
-                                           batches,
-                                           start,
-                                           indexOfLastProcessedBatch,
-                                           canContinue,
-                                           errors,
-                                           opTime,
-                                           electionId,
-                                           docsToRetry);
-        }
-
-        void _getTimeseriesBatchResultsNoTenantMigration(
+        TimeseriesAtomicWriteResult _performOrderedTimeseriesWritesAtomically(
             OperationContext* opCtx,
-            const TimeseriesBatches& batches,
-            int64_t start,
-            int64_t indexOfLastProcessedBatch,
-            bool canContinue,
             std::vector<write_ops::WriteError>* errors,
             boost::optional<repl::OpTime>* opTime,
             boost::optional<OID>* electionId,
-            std::vector<size_t>* docsToRetry = nullptr) const {
-            auto errorGenerator =
-                [](OperationContext* opCtx, const Status& status, int index, size_t numErrors) {
-                    return generateErrorNoTenantMigration(opCtx, status, index, numErrors);
-                };
-            _getTimeseriesBatchResultsBase(errorGenerator,
-                                           opCtx,
-                                           batches,
-                                           start,
-                                           indexOfLastProcessedBatch,
-                                           canContinue,
-                                           errors,
-                                           opTime,
-                                           electionId,
-                                           docsToRetry);
-        }
-
-        bool _performOrderedTimeseriesWritesAtomically(OperationContext* opCtx,
-                                                       std::vector<write_ops::WriteError>* errors,
-                                                       boost::optional<repl::OpTime>* opTime,
-                                                       boost::optional<OID>* electionId,
-                                                       bool* containsRetry) const {
-            auto [batches, stmtIds, numInserted] = _insertIntoBucketCatalog(
+            bool* containsRetry) const {
+            auto [batches, stmtIds, numInserted, canContinue] = _insertIntoBucketCatalog(
                 opCtx, 0, request().getDocuments().size(), {}, errors, containsRetry);
+            if (!canContinue) {
+                return TimeseriesAtomicWriteResult::kNonContinuableError;
+            }
 
             hangTimeseriesInsertBeforeCommit.pauseWhileSet();
 
-            if (!_commitTimeseriesBucketsAtomically(
-                    opCtx, &batches, std::move(stmtIds), errors, opTime, electionId)) {
-                return false;
+            auto result = _commitTimeseriesBucketsAtomically(
+                opCtx, &batches, std::move(stmtIds), errors, opTime, electionId);
+            if (result != TimeseriesAtomicWriteResult::kSuccess) {
+                return result;
             }
 
             _getTimeseriesBatchResults(
                 opCtx, batches, 0, batches.size(), true, errors, opTime, electionId);
 
-            return true;
+            return TimeseriesAtomicWriteResult::kSuccess;
         }
 
         /**
@@ -1251,9 +1187,19 @@ public:
                                                boost::optional<repl::OpTime>* opTime,
                                                boost::optional<OID>* electionId,
                                                bool* containsRetry) const {
-            if (_performOrderedTimeseriesWritesAtomically(
-                    opCtx, errors, opTime, electionId, containsRetry)) {
-                return request().getDocuments().size();
+            auto result = _performOrderedTimeseriesWritesAtomically(
+                opCtx, errors, opTime, electionId, containsRetry);
+            switch (result) {
+                case TimeseriesAtomicWriteResult::kSuccess:
+                    return request().getDocuments().size();
+                case TimeseriesAtomicWriteResult::kNonContinuableError:
+                    // If we can't continue, we know that 0 were inserted since this function should
+                    // guarantee that the inserts are atomic.
+                    return 0;
+                case TimeseriesAtomicWriteResult::kContinuableError:
+                    break;
+                default:
+                    MONGO_UNREACHABLE;
             }
 
             for (size_t i = 0; i < request().getDocuments().size(); ++i) {
@@ -1272,8 +1218,6 @@ public:
          * which were attempted in an update operation, but found no bucket to update. These indices
          * can be passed as the 'indices' parameter in a subsequent call to this function, in order
          * to to be retried.
-         * In rare cases due to collision from OID generation, we will also retry inserting those
-         * bucket * documents for a limited number of times.
          */
         std::vector<size_t> _performUnorderedTimeseriesWrites(
             OperationContext* opCtx,
@@ -1283,73 +1227,39 @@ public:
             std::vector<write_ops::WriteError>* errors,
             boost::optional<repl::OpTime>* opTime,
             boost::optional<OID>* electionId,
-            bool* containsRetry,
-            absl::flat_hash_map<int, int>& retryAttemptsForDup) const {
-            auto [batches, bucketStmtIds, _] =
+            bool* containsRetry) const {
+            auto [batches, bucketStmtIds, _, canContinue] =
                 _insertIntoBucketCatalog(opCtx, start, numDocs, indices, errors, containsRetry);
 
             hangTimeseriesInsertBeforeCommit.pauseWhileSet();
 
             std::vector<size_t> docsToRetry;
 
-            bool canContinue = true;
-
-
-            stdx::unordered_set<BucketCatalog::WriteBatch*> handledHere;
-            int64_t handledElsewhere = 0;
-            auto guard = ScopeGuard([this, &handledElsewhere, opCtx]() {
-                if (handledElsewhere > 0) {
-                    auto& bucketCatalog = BucketCatalog::get(opCtx);
-                    bucketCatalog.reportMeasurementsGroupCommitted(request().getNamespace(),
-                                                                   handledElsewhere);
-                }
-            });
+            if (!canContinue) {
+                return docsToRetry;
+            }
 
             size_t itr = 0;
             for (; itr < batches.size(); ++itr) {
                 auto& [batch, index] = batches[itr];
                 if (batch->claimCommitRights()) {
-                    handledHere.insert(batch.get());
                     auto stmtIds = isTimeseriesWriteRetryable(opCtx)
                         ? std::move(bucketStmtIds[batch->bucket().id])
                         : std::vector<StmtId>{};
-                    try {
-                        canContinue = _commitTimeseriesBucket(opCtx,
-                                                              batch,
-                                                              start,
-                                                              index,
-                                                              std::move(stmtIds),
-                                                              errors,
-                                                              opTime,
-                                                              electionId,
-                                                              &docsToRetry,
-                                                              retryAttemptsForDup);
-                    } catch (const DBException& ex) {
-                        // Exception during commit, append error and wait for all our batches to
-                        // commit or
-                        // abort. We need to wait here as pointers to memory owned by this command
-                        // is stored in the WriteBatch(es). This ensures that no other thread may
-                        // try to access this memory after this command has been torn down due to
-                        // the exception.
-                        errors->emplace_back(*generateErrorNoTenantMigration(
-                            opCtx, ex.toStatus(), start + index, errors->size()));
-                        _getTimeseriesBatchResultsNoTenantMigration(opCtx,
-                                                                    batches,
-                                                                    0,
-                                                                    itr,
-                                                                    canContinue,
-                                                                    errors,
-                                                                    opTime,
-                                                                    electionId,
-                                                                    &docsToRetry);
-                        throw;
-                    }
+
+                    canContinue = _commitTimeseriesBucket(opCtx,
+                                                          batch,
+                                                          start,
+                                                          index,
+                                                          std::move(stmtIds),
+                                                          errors,
+                                                          opTime,
+                                                          electionId,
+                                                          &docsToRetry);
                     batch.reset();
                     if (!canContinue) {
                         break;
                     }
-                } else if (!handledHere.contains(batch.get())) {
-                    ++handledElsewhere;
                 }
             }
 
@@ -1370,20 +1280,9 @@ public:
             boost::optional<OID>* electionId,
             bool* containsRetry) const {
             std::vector<size_t> docsToRetry;
-            absl::flat_hash_map<int, int> retryAttemptsForDup;
             do {
-                docsToRetry = _performUnorderedTimeseriesWrites(opCtx,
-                                                                start,
-                                                                numDocs,
-                                                                docsToRetry,
-                                                                errors,
-                                                                opTime,
-                                                                electionId,
-                                                                containsRetry,
-                                                                retryAttemptsForDup);
-                if (!retryAttemptsForDup.empty()) {
-                    BucketCatalog::get(opCtx).resetBucketOIDCounter();
-                }
+                docsToRetry = _performUnorderedTimeseriesWrites(
+                    opCtx, start, numDocs, docsToRetry, errors, opTime, electionId, containsRetry);
             } while (!docsToRetry.empty());
         }
 
@@ -1403,11 +1302,6 @@ public:
                             curOp.getReadWriteType());
             });
 
-            // If an expected collection UUID is provided, always fail because the user-facing
-            // time-series namespace does not have a UUID.
-            checkCollectionUUIDMismatch(
-                opCtx, request().getNamespace(), nullptr, request().getCollectionUUID());
-
             uassert(
                 ErrorCodes::OperationNotSupportedInTransaction,
                 str::stream() << "Cannot insert into a time-series collection in a multi-document "
@@ -1547,29 +1441,19 @@ public:
 
             invariant(!_commandObj.isEmpty());
 
+            if (const auto& shardVersion = _commandObj.getField("shardVersion");
+                !shardVersion.eoo()) {
+                bob->append(shardVersion);
+            }
 
             bob->append("find", _commandObj["update"].String());
             extractQueryDetails(_updateOpObj, bob);
             bob->append("batchSize", 1);
             bob->append("singleBatch", true);
-
-            if (const auto& shardVersion = _commandObj.getField("shardVersion");
-                !shardVersion.eoo()) {
-                bob->append(shardVersion);
-            }
-            if (const auto& databaseVersion = _commandObj.getField("databaseVersion");
-                !databaseVersion.eoo()) {
-                bob->append(databaseVersion);
-            }
         }
 
         write_ops::UpdateCommandReply typedRun(OperationContext* opCtx) final try {
-            // On debug builds, verify that the estimated size of the update command is at least as
-            // large as the size of the actual, serialized update command. This ensures that the
-            // logic which estimates the size of update commands is correct.
-            dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest()));
             transactionChecks(opCtx, ns());
-
             write_ops::UpdateCommandReply updateReply;
             OperationSource source = OperationSource::kStandard;
 
@@ -1676,7 +1560,6 @@ public:
             updateRequest.setLegacyRuntimeConstants(request().getLegacyRuntimeConstants().value_or(
                 Variables::generateRuntimeConstants(opCtx)));
             updateRequest.setLetParameters(request().getLet());
-            updateRequest.setBypassEmptyTsReplacement(request().getBypassEmptyTsReplacement());
             updateRequest.setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
             updateRequest.setExplain(verbosity);
 
@@ -1758,11 +1641,6 @@ public:
         }
 
         write_ops::DeleteCommandReply typedRun(OperationContext* opCtx) final try {
-            // On debug builds, verify that the estimated size of the deletes are at least as large
-            // as the actual, serialized size. This ensures that the logic that estimates the size
-            // of deletes for batch writes is correct.
-            dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest()));
-
             transactionChecks(opCtx, ns());
             write_ops::DeleteCommandReply deleteReply;
             OperationSource source = OperationSource::kStandard;