Update upstream source from tag 'upstream/6.0.10'debian/6.0.10-1

Update to upstream version '6.0.10'
with Debian dir 2d176fa254eee97b139f712fec5709641335a8c3

35 files changed, 721 insertions, 400 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 143c5eae3ef..5a53af0ebf0 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -231,7 +231,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/commands',
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/storage/backup_cursor_hooks',
         'fsync_locked',
@@ -372,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',
@@ -401,6 +401,7 @@ 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',
@@ -575,6 +576,7 @@ 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',
@@ -661,6 +663,7 @@ 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',
@@ -736,7 +739,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/commands/servers',
         '$BUILD_DIR/mongo/db/db_raii',
-        '$BUILD_DIR/mongo/db/index/index_access_methods',
+        '$BUILD_DIR/mongo/db/index/index_access_method',
         '$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',
@@ -809,6 +812,7 @@ 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 2b379ee085e..1a31cb0d8f8 100644
--- a/src/mongo/db/commands/apply_ops_cmd.cpp
+++ b/src/mongo/db/commands/apply_ops_cmd.cpp
@@ -39,7 +39,6 @@
 #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 0fef67553d0..8f735d5b948 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);
+            session->updateUserName(user, true /* isMechX509 */);
             return user;
         } else {
             uassert(ErrorCodes::AuthenticationFailed,
@@ -341,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);
+        session->updateUserName(internalSecurityUser, mechanism == auth::kMechanismMongoX509);
     } else {
-        session->updateUserName(UserName{user, dbname});
+        session->updateUserName(UserName{user, dbname}, mechanism == auth::kMechanismMongoX509);
     }
 
     if (mechanism.empty()) {
diff --git a/src/mongo/db/commands/authentication_commands.h b/src/mongo/db/commands/authentication_commands.h
index 2e82168e050..814a2ab17d2 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 isMaster, BSONObjBuilder* result);
+void doSpeculativeAuthenticate(OperationContext* opCtx, BSONObj helloCmd, BSONObjBuilder* result);
 
 }  // namespace mongo
diff --git a/src/mongo/db/commands/compact.cpp b/src/mongo/db/commands/compact.cpp
index 6aa009ea216..26ee8ab5d4b 100644
--- a/src/mongo/db/commands/compact.cpp
+++ b/src/mongo/db/commands/compact.cpp
@@ -95,6 +95,9 @@ 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/cqf/cqf_aggregate.cpp b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
index 100da0be582..db65e3083d8 100644
--- a/src/mongo/db/commands/cqf/cqf_aggregate.cpp
+++ b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
@@ -29,6 +29,7 @@
 
 #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"
@@ -62,7 +63,8 @@ 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, false /*includeUnfinished*/);
+    auto indexIterator =
+        indexCatalog.getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
 
     while (indexIterator->more()) {
         const IndexCatalogEntry& catalogEntry = *indexIterator->next();
@@ -335,6 +337,9 @@ 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 c743a405714..f6684487a21 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/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/create_indexes_gen.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -321,11 +321,12 @@ 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);
@@ -447,7 +448,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/dbcheck.cpp b/src/mongo/db/commands/dbcheck.cpp
index c26b09d8831..9084c1c3c1b 100644
--- a/src/mongo/db/commands/dbcheck.cpp
+++ b/src/mongo/db/commands/dbcheck.cpp
@@ -39,7 +39,7 @@
 #include "mongo/db/catalog/health_log.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/test_commands_enabled.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.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/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 0e88f993f93..54574139484 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -58,7 +58,6 @@
 #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"
@@ -169,41 +168,6 @@ 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:
@@ -330,13 +294,19 @@ public:
         bool estimate = jsobj["estimate"].trueValue();
 
         const NamespaceString nss(ns);
-        AutoGetCollectionForReadCommand collection(opCtx, nss);
+        AutoGetCollectionForReadCommand autoColl(opCtx, nss);
+        const auto& collection = autoColl.getCollection();
 
-        const auto collDesc =
-            CollectionShardingState::get(opCtx, nss)->getCollectionDescription(opCtx);
+        if (!collection) {
+            // Collection does not exist
+            result.appendNumber("size", 0);
+            result.appendNumber("numObjects", 0);
+            result.append("millis", timer.millis());
+            return true;
+        }
 
-        if (collDesc.isSharded()) {
-            const ShardKeyPattern shardKeyPattern(collDesc.getKeyPattern());
+        if (collection.isSharded()) {
+            const ShardKeyPattern shardKeyPattern(collection.getShardKeyPattern());
             uassert(ErrorCodes::BadValue,
                     "keyPattern must be empty or must be an object that equals the shard key",
                     keyPattern.isEmpty() ||
@@ -354,10 +324,7 @@ public:
             max = shardKeyPattern.normalizeShardKey(max);
         }
 
-        long long numRecords = 0;
-        if (collection) {
-            numRecords = collection->numRecords(opCtx);
-        }
+        const long long numRecords = collection->numRecords(opCtx);
 
         if (numRecords == 0) {
             result.appendNumber("size", 0);
@@ -377,7 +344,7 @@ public:
                 return 1;
             }
             exec = InternalPlanner::collectionScan(
-                opCtx, &collection.getCollection(), PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY);
+                opCtx, &collection, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
         } else if (min.isEmpty() || max.isEmpty()) {
             errmsg = "only one of min or max specified";
             return false;
@@ -388,7 +355,7 @@ public:
             }
 
             auto shardKeyIdx = findShardKeyPrefixedIndex(opCtx,
-                                                         *collection,
+                                                         collection,
                                                          collection->getIndexCatalog(),
                                                          keyPattern,
                                                          /*requireSingleKey=*/true);
@@ -403,12 +370,12 @@ public:
             max = Helpers::toKeyFormat(kp.extendRangeBound(max, false));
 
             exec = InternalPlanner::shardKeyIndexScan(opCtx,
-                                                      &collection.getCollection(),
+                                                      &collection,
                                                       *shardKeyIdx,
                                                       min,
                                                       max,
                                                       BoundInclusion::kIncludeStartKeyOnly,
-                                                      PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY);
+                                                      PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
         }
 
         CurOpFailpointHelpers::waitWhileFailPointEnabled(
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index 41969950a43..339d442bcaf 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -57,7 +57,8 @@
 #include "mongo/db/commands/profile_common.h"
 #include "mongo/db/commands/profile_gen.h"
 #include "mongo/db/commands/server_status.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/commands/set_profiling_filter_globally_cmd.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
@@ -204,6 +205,8 @@ protected:
 
 } cmdProfile;
 
+SetProfilingFilterGloballyCmd cmdSetProfilingFilterGlobally;
+
 class CmdFileMD5 : public BasicCommand {
 public:
     CmdFileMD5() : BasicCommand("filemd5") {}
@@ -379,7 +382,7 @@ public:
                     try {
                         // RELOCKED
                         ctx.reset(new AutoGetCollectionForReadCommand(opCtx, nss));
-                    } catch (const StaleConfigException&) {
+                    } catch (const ExceptionFor<ErrorCodes::StaleConfig>&) {
                         LOGV2_DEBUG(
                             20453,
                             1,
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index 57585a88401..a0396397a8d 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -296,8 +296,6 @@ 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),
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index f4595688706..8e0a48f013b 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/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/drop_indexes_gen.h"
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index abbc0d834fd..1422a6278dd 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/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/exec/update_stage.h"
@@ -95,16 +95,17 @@ 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}, stats: {stats}, cmd: {cmd}",
-            "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"_attr = exception.toStatus(),
+                      "stats"_attr = redact(stats),
+                      "cmd"_attr = request.toBSON(BSONObj() /* commandPassthroughFields */));
 
         exception.addContext("Plan executor error during findAndModify");
         throw;
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index f7ac781404f..1fdb621add3 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -133,11 +133,17 @@ std::unique_ptr<FindCommandRequest> parseCmdObjectToFindCommandRequest(Operation
 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();
     }
 
     // Although both 'find' and 'aggregate' commands have an ExpressionContext, some of the data
@@ -320,7 +326,16 @@ public:
 
             // Finish the parsing step by using the FindCommandRequest to create a CanonicalQuery.
             const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
-            auto expCtx = makeExpressionContext(opCtx, *findCommand, verbosity);
+
+            // 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);
             const bool isExplain = true;
             auto cq = uassertStatusOK(
                 CanonicalQuery::canonicalize(opCtx,
@@ -370,10 +385,6 @@ 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 =
@@ -505,14 +516,16 @@ public:
             }
 
             // Tailing a replicated capped clustered collection requires majority read concern.
-            const auto coll = ctx->getCollection().get();
-            if (coll) {
+            const auto& collection = ctx->getCollection();
+
+            bool isClusteredCollection = false;
+            if (collection) {
                 const bool isTailable = findCommand->getTailable();
                 const bool isMajorityReadConcern = repl::ReadConcernArgs::get(opCtx).getLevel() ==
                     repl::ReadConcernLevel::kMajorityReadConcern;
-                const bool isClusteredCollection = coll->isClustered();
-                const bool isCapped = coll->isCapped();
-                const bool isReplicated = coll->ns().isReplicated();
+                isClusteredCollection = collection->isClustered();
+                const bool isCapped = collection->isCapped();
+                const bool isReplicated = collection->ns().isReplicated();
                 if (isClusteredCollection && isCapped && isReplicated && isTailable) {
                     uassert(ErrorCodes::Error(6049203),
                             "A tailable cursor on a capped clustered collection requires majority "
@@ -521,12 +534,22 @@ 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);
 
             // 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 expCtx =
+                makeExpressionContext(opCtx, *findCommand, collection, boost::none /* verbosity */);
             auto cq = uassertStatusOK(
                 CanonicalQuery::canonicalize(opCtx,
                                              std::move(findCommand),
@@ -570,8 +593,6 @@ 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.
@@ -656,8 +677,6 @@ 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,14 +734,14 @@ public:
                 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()->getOperationStatistics();
+                        opCtx->recoveryUnit()->computeOperationStatisticsSinceLastCall();
                 }
             } else {
                 endQueryOp(opCtx, collection, *exec, numResults, cursorId);
diff --git a/src/mongo/db/commands/fsync.cpp b/src/mongo/db/commands/fsync.cpp
index f1d92aa636a..de77efdc687 100644
--- a/src/mongo/db/commands/fsync.cpp
+++ b/src/mongo/db/commands/fsync.cpp
@@ -47,7 +47,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/fsync_locked.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/backup_cursor_hooks.h"
 #include "mongo/db/storage/storage_engine.h"
diff --git a/src/mongo/db/commands/generic_servers.cpp b/src/mongo/db/commands/generic_servers.cpp
index 41448e90dde..7a6b87c13fe 100644
--- a/src/mongo/db/commands/generic_servers.cpp
+++ b/src/mongo/db/commands/generic_servers.cpp
@@ -156,8 +156,11 @@ 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 598ce277333..c1f626a7f07 100644
--- a/src/mongo/db/commands/generic_servers.idl
+++ b/src/mongo/db/commands/generic_servers.idl
@@ -64,8 +64,11 @@ 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 35060c90ed4..6f87839e88d 100644
--- a/src/mongo/db/commands/get_cluster_parameter_command.cpp
+++ b/src/mongo/db/commands/get_cluster_parameter_command.cpp
@@ -65,6 +65,11 @@ 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 b3c00996ec6..c699a3262e7 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -431,7 +431,6 @@ 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),
@@ -617,7 +616,6 @@ public:
                 options.atClusterTime = repl::ReadConcernArgs::get(opCtx).getArgsAtClusterTime();
             }
             CursorResponseBuilder nextBatch(reply, options);
-            BSONObj obj;
             std::uint64_t numResults = 0;
             ResourceConsumption::DocumentUnitCounter docUnitsReturned;
 
@@ -634,9 +632,7 @@ public:
                     // Use the commit point of the last batch for exhaust cursors.
                     lastKnownCommittedOpTime = cursorPin->getLastKnownCommittedOpTime();
                 }
-                if (lastKnownCommittedOpTime) {
-                    clientsLastKnownCommittedOpTime(opCtx) = lastKnownCommittedOpTime.get();
-                }
+                clientsLastKnownCommittedOpTime(opCtx) = lastKnownCommittedOpTime;
 
                 awaitDataState(opCtx).shouldWaitForInserts = true;
             }
@@ -699,10 +695,19 @@ public:
 
                 cursorPin->setLeftoverMaxTimeMicros(opCtx->getRemainingMaxTimeMicros());
 
-                if (opCtx->isExhaust() && !clientsLastKnownCommittedOpTime(opCtx).isNull()) {
-                    // Set the commit point of the latest batch.
+                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.
                     auto replCoord = repl::ReplicationCoordinator::get(opCtx);
-                    cursorPin->setLastKnownCommittedOpTime(replCoord->getLastCommittedOpTime());
+                    auto myLastCommittedOpTime = replCoord->getLastCommittedOpTime();
+                    auto clientsLastKnownCommittedOpTime = cursorPin->getLastKnownCommittedOpTime();
+                    if (!clientsLastKnownCommittedOpTime.has_value() ||
+                        clientsLastKnownCommittedOpTime.value() < myLastCommittedOpTime) {
+                        cursorPin->setLastKnownCommittedOpTime(myLastCommittedOpTime);
+                    }
                 }
             } else {
                 curOp->debug().cursorExhausted = true;
diff --git a/src/mongo/db/commands/killoperations_common.h b/src/mongo/db/commands/killoperations_common.h
index b9dd9eb79a9..bae129a4bbe 100644
--- a/src/mongo/db/commands/killoperations_common.h
+++ b/src/mongo/db/commands/killoperations_common.h
@@ -64,7 +64,8 @@ public:
             auto opKeys = Base::request().getOperationKeys();
 
             for (auto& opKey : opKeys) {
-                LOGV2(4615602, "Attempting to kill operation", "operationKey"_attr = opKey);
+                LOGV2_DEBUG(
+                    4615602, 2, "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 f78cfda6308..eb4401a944f 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -167,9 +167,7 @@ BSONObj buildViewBson(const ViewDefinition& view, bool nameOnly) {
     return b.obj();
 }
 
-BSONObj buildTimeseriesBson(OperationContext* opCtx,
-                            const CollectionPtr& collection,
-                            bool nameOnly) {
+BSONObj buildTimeseriesBson(const CollectionPtr& collection, bool nameOnly) {
     invariant(collection);
 
     BSONObjBuilder builder;
@@ -377,8 +375,7 @@ public:
                                     if (auto bucketsCollection = CollectionCatalog::get(opCtx)
                                                                      ->lookupCollectionByNamespace(
                                                                          opCtx, view->viewOn())) {
-                                        return buildTimeseriesBson(
-                                            opCtx, bucketsCollection, nameOnly);
+                                        return buildTimeseriesBson(bucketsCollection, nameOnly);
                                     } else {
                                         // The buckets collection does not exist, so the time-series
                                         // view will be appended when we iterate through the view
@@ -397,21 +394,22 @@ public:
                     } else {
                         auto perCollectionWork = [&](const CollectionPtr& collection) {
                             if (collection && collection->getTimeseriesOptions() &&
-                                !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());
+                                !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());
+                                }
                             }
 
                             if (authorizedCollections &&
@@ -501,7 +499,7 @@ public:
                     batchSize = *listCollRequest.getCursor()->getBatchSize();
                 }
 
-                size_t bytesBuffered = 0;
+                FindCommon::BSONArrayResponseSizeTracker responseSizeTracker;
                 for (long long objCount = 0; objCount < batchSize; objCount++) {
                     BSONObj nextDoc;
                     PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr);
@@ -512,7 +510,7 @@ public:
 
                     // If we can't fit this result inside the current batch, then we stash it for
                     // later.
-                    if (!FindCommon::haveSpaceForNext(nextDoc, objCount, bytesBuffered)) {
+                    if (!responseSizeTracker.haveSpaceForNext(nextDoc)) {
                         exec->stashResult(nextDoc);
                         break;
                     }
@@ -528,7 +526,7 @@ public:
                             "error"_attr = exc);
                         fassertFailed(5254301);
                     }
-                    bytesBuffered += nextDoc.objsize();
+                    responseSizeTracker.add(nextDoc);
                 }
                 if (exec->isEOF()) {
                     return createListCollectionsCursorReply(
diff --git a/src/mongo/db/commands/list_databases.cpp b/src/mongo/db/commands/list_databases.cpp
index 4b59a41f64e..fd60d2b26b5 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/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.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 6f4f4f461b9..22cbb4b11df 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -40,7 +40,6 @@
 #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"
@@ -65,38 +64,40 @@ namespace mongo {
 namespace {
 
 // The allowed fields have to be in sync with those defined in 'src/mongo/db/list_indexes.idl'.
-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};
+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}},
+};
 
 /**
  * Returns index specs, with resolved namespace, from the catalog for this listIndexes request.
@@ -306,7 +307,7 @@ public:
                                             nss));
 
             std::vector<mongo::ListIndexesReplyItem> firstBatch;
-            size_t bytesBuffered = 0;
+            FindCommon::BSONArrayResponseSizeTracker responseSizeTracker;
             for (long long objCount = 0; objCount < batchSize; objCount++) {
                 BSONObj nextDoc;
                 PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr);
@@ -318,7 +319,7 @@ public:
 
                 // If we can't fit this result inside the current batch, then we stash it for
                 // later.
-                if (!FindCommon::haveSpaceForNext(nextDoc, objCount, bytesBuffered)) {
+                if (!responseSizeTracker.haveSpaceForNext(nextDoc)) {
                     exec->stashResult(nextDoc);
                     break;
                 }
@@ -337,7 +338,7 @@ public:
                                           nextDoc.toString(),
                                           exc.toString()));
                 }
-                bytesBuffered += nextDoc.objsize();
+                responseSizeTracker.add(nextDoc);
             }
 
             if (exec->isEOF()) {
diff --git a/src/mongo/db/commands/mr_test.cpp b/src/mongo/db/commands/mr_test.cpp
index 745d883c8e1..95fe13b68ee 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/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.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 7af5c111ca3..9f93f116815 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/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
diff --git a/src/mongo/db/commands/profile.idl b/src/mongo/db/commands/profile.idl
index cd4a295a950..f1c56a88a05 100644
--- a/src/mongo/db/commands/profile.idl
+++ b/src/mongo/db/commands/profile.idl
@@ -66,3 +66,16 @@ 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 48dc95ade4e..4433edae662 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/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.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 9172103b493..ac43ccf8b36 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -221,7 +221,6 @@ 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),
@@ -657,6 +656,93 @@ 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,
+                          const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                          std::shared_ptr<const CollectionCatalog> catalog,
+                          const PrivilegeVector& privileges,
+                          CurOp* curOp,
+                          rpc::ReplyBuilderInterface* result,
+                          const std::function<void(void)>& resetContextFn) {
+    auto nss = request.getNamespace();
+    checkCollectionUUIDMismatch(
+        opCtx, nss, collections.getMainCollection(), request.getCollectionUUID());
+
+    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, 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;
+}
+
 }  // namespace
 
 Status runAggregate(OperationContext* opCtx,
@@ -745,7 +831,8 @@ Status runAggregate(OperationContext* opCtx,
         boost::optional<AutoStatsTracker> statsTracker;
 
         // If this is a change stream, perform special checks and change the execution namespace.
-        if (liteParsedPipeline.hasChangeStream()) {
+        const auto hasChangeStream = liteParsedPipeline.hasChangeStream();
+        if (hasChangeStream) {
             uassert(4928900,
                     str::stream() << AggregateCommandRequest::kCollectionUUIDFieldName
                                   << " is not supported for a change stream",
@@ -820,87 +907,24 @@ Status runAggregate(OperationContext* opCtx,
         // 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.)
-        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;
+        // 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(),
+                                      expCtx,
+                                      catalog,
+                                      privileges,
+                                      curOp,
+                                      result,
+                                      resetContext);
         }
 
         // If collectionUUID was provided, verify the collection exists and has the expected UUID.
@@ -914,6 +938,20 @@ Status runAggregate(OperationContext* opCtx,
         expCtx = makeExpressionContext(
             opCtx, request, std::move(*collatorToUse), uuid, collatorToUseMatchesDefault);
 
+        // 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);
+        });
+
+        // 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;
+            expCtx->ignoreTokenVersionOnResume = true;
+        }
+
         expCtx->startExpressionCounters();
         auto pipeline = Pipeline::parse(request.getPipeline(), expCtx);
         expCtx->stopExpressionCounters();
diff --git a/src/mongo/db/commands/set_cluster_parameter_command.cpp b/src/mongo/db/commands/set_cluster_parameter_command.cpp
index 005f17d0c06..2049b3043e5 100644
--- a/src/mongo/db/commands/set_cluster_parameter_command.cpp
+++ b/src/mongo/db/commands/set_cluster_parameter_command.cpp
@@ -74,6 +74,11 @@ 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 73933d1abe2..2c74781c36e 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -84,6 +84,7 @@
 #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"
@@ -465,7 +466,7 @@ public:
                 if (actualVersion > requestedVersion &&
                     !feature_flags::gOrphanTracking.isEnabledOnVersion(requestedVersion)) {
                     BalancerStatsRegistry::get(opCtx)->terminate();
-                    ScopedRangeDeleterLock rangeDeleterLock(opCtx);
+                    ScopedRangeDeleterLock rangeDeleterLock(opCtx, LockMode::MODE_X);
                     clearOrphanCountersFromRangeDeletionTasks(opCtx);
                 }
 
@@ -750,37 +751,19 @@ private:
                 const auto& dbName = tenantDbName.dbName();
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
-                    opCtx,
-                    tenantDbName,
-                    MODE_X,
-                    [&](const CollectionPtr& collection) {
-                        invariant(collection->getTimeseriesOptions());
-
+                    opCtx, tenantDbName, MODE_X, [&](const CollectionPtr& collection) {
+                        const auto collNs = collection->getTimeseriesOptions()
+                            ? collection->ns().getTimeseriesViewNamespace()
+                            : collection->ns();
                         auto indexCatalog = collection->getIndexCatalog();
-                        auto indexIt = indexCatalog->getIndexIterator(
-                            opCtx, /*includeUnfinishedIndexes=*/true);
 
+                        auto 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
-                            // in 5.2 and up. If the user tries to downgrade the cluster to an
-                            // earlier version, they must first remove all incompatible secondary
-                            // indexes on time-series measurements.
-                            uassert(
-                                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 "
-                                       "partial indexes on time-series collections, before "
-                                       "downgrading. First detected incompatible index name: '"
-                                    << indexEntry->descriptor()->indexName() << "' on collection: '"
-                                    << collection->ns().getTimeseriesViewNamespace() << "'",
-                                timeseries::isBucketsIndexSpecCompatibleForDowngrade(
-                                    *collection->getTimeseriesOptions(),
-                                    indexEntry->descriptor()->infoObj()));
-
                             if (auto filter = indexEntry->getFilterExpression()) {
                                 auto status = IndexCatalogImpl::checkValidFilterExpressions(
                                     filter,
@@ -794,12 +777,43 @@ private:
                                                "partial filter elements before downgrading. First "
                                                "detected incompatible index name: '"
                                             << indexEntry->descriptor()->indexName()
-                                            << "' on collection: '"
-                                            << collection->ns().getTimeseriesViewNamespace() << "'",
+                                            << "' on collection: '" << collNs << "'",
                                         status.isOK());
                             }
                         }
 
+                        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
+                            // in 5.2 and up. If the user tries to downgrade the cluster to an
+                            // earlier version, they must first remove all incompatible secondary
+                            // indexes on time-series measurements.
+                            uassert(
+                                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 "
+                                       "partial indexes on time-series collections, before "
+                                       "downgrading. First detected incompatible index name: '"
+                                    << indexEntry->descriptor()->indexName() << "' on collection: '"
+                                    << collNs << "'",
+                                timeseries::isBucketsIndexSpecCompatibleForDowngrade(
+                                    *collection->getTimeseriesOptions(),
+                                    indexEntry->descriptor()->infoObj()));
+                        }
+
                         if (!collection->getTimeseriesBucketsMayHaveMixedSchemaData()) {
                             // The catalog entry flag has already been removed. This can happen if
                             // the downgrade process was interrupted and is being run again. The
@@ -824,9 +838,6 @@ private:
                         }
 
                         return true;
-                    },
-                    [&](const CollectionPtr& collection) {
-                        return collection->getTimeseriesOptions() != boost::none;
                     });
             }
         }
@@ -868,7 +879,10 @@ private:
                     opCtx, tenantDbName, MODE_X, [&](const CollectionPtr& collection) {
                         auto indexCatalog = collection->getIndexCatalog();
                         auto indexIt = indexCatalog->getIndexIterator(
-                            opCtx, true /* includeUnfinishedIndexes */);
+                            opCtx,
+                            IndexCatalog::InclusionPolicy::kReady |
+                                IndexCatalog::InclusionPolicy::kUnfinished |
+                                IndexCatalog::InclusionPolicy::kFrozen);
                         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
new file mode 100644
index 00000000000..eac2783bbd9
--- /dev/null
+++ b/src/mongo/db/commands/set_profiling_filter_globally_cmd.cpp
@@ -0,0 +1,103 @@
+/**
+ *    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);
+
+    // 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
new file mode 100644
index 00000000000..15c36f0a9cc
--- /dev/null
+++ b/src/mongo/db/commands/set_profiling_filter_globally_cmd.h
@@ -0,0 +1,70 @@
+/**
+ *    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/validate.cpp b/src/mongo/db/commands/validate.cpp
index c6724076b9f..c100d4f8bbc 100644
--- a/src/mongo/db/commands/validate.cpp
+++ b/src/mongo/db/commands/validate.cpp
@@ -35,11 +35,13 @@
 #include "mongo/db/catalog/collection_validation.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.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 {
 
@@ -61,6 +63,79 @@ 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
 
 /**
@@ -123,6 +198,7 @@ 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.
@@ -255,8 +331,8 @@ public:
         }
 
         ValidateResults validateResults;
-        Status status =
-            CollectionValidation::validate(opCtx, nss, mode, repairMode, &validateResults, &result);
+        Status status = CollectionValidation::validate(
+            opCtx, nss, mode, repairMode, &validateResults, &result, logDiagnostics);
         if (!status.isOK()) {
             return CommandHelpers::appendCommandStatusNoThrow(result, status);
         }
@@ -267,6 +343,7 @@ 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 c3a00ff74e6..6350a728672 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -215,8 +215,10 @@ public:
 
             // Ensure there are no unstable indexes.
             const auto* indexCatalog = collection->getIndexCatalog();
-            std::unique_ptr<IndexCatalog::IndexIterator> ii =
-                indexCatalog->getIndexIterator(opCtx, true /* includeUnfinishedIndexes */);
+            auto ii = indexCatalog->getIndexIterator(
+                opCtx,
+                IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                    IndexCatalog::InclusionPolicy::kFrozen);
             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 0254baca47d..9186ff13cbe 100644
--- a/src/mongo/db/commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands.cpp
@@ -30,6 +30,7 @@
 #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"
@@ -73,6 +74,7 @@
 #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"
@@ -527,6 +529,11 @@ 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() &&
@@ -698,16 +705,16 @@ public:
                 OperationSource::kTimeseriesInsert));
         }
 
-        TimeseriesSingleWriteResult _performTimeseriesBucketCompression(
+        void _performTimeseriesBucketCompression(
             OperationContext* opCtx, const BucketCatalog::ClosedBucket& closedBucket) const {
             if (!feature_flags::gTimeseriesBucketCompression.isEnabled(
                     serverGlobalParams.featureCompatibility)) {
-                return {SingleWriteResult(), true};
+                return;
             }
 
             // Buckets with just a single measurement is not worth compressing.
             if (closedBucket.numMeasurements <= 1) {
-                return {SingleWriteResult(), true};
+                return;
             }
 
             bool validateCompression = gValidateTimeseriesCompression.load();
@@ -745,8 +752,8 @@ public:
 
             auto compressionOp =
                 _makeTimeseriesCompressionOp(opCtx, closedBucket.bucketId, bucketCompressionFunc);
-            auto result = _getTimeseriesSingleWriteResult(
-                write_ops_exec::performUpdates(opCtx, compressionOp, OperationSource::kStandard));
+            auto result = _getTimeseriesSingleWriteResult(write_ops_exec::performUpdates(
+                opCtx, compressionOp, OperationSource::kTimeseriesBucketCompression));
 
             // Report stats, if we fail before running the transform function then just skip
             // reporting.
@@ -761,8 +768,6 @@ public:
                     stats.onBucketClosed(*beforeSize, compressionStats);
                 }
             }
-
-            return result;
         }
 
         /**
@@ -776,7 +781,8 @@ public:
                                      std::vector<write_ops::WriteError>* errors,
                                      boost::optional<repl::OpTime>* opTime,
                                      boost::optional<OID>* electionId,
-                                     std::vector<size_t>* docsToRetry) const try {
+                                     std::vector<size_t>* docsToRetry,
+                                     absl::flat_hash_map<int, int>& retryAttemptsForDup) const try {
             auto& bucketCatalog = BucketCatalog::get(opCtx);
 
             auto metadata = bucketCatalog.getMetadata(batch->bucket());
@@ -796,9 +802,18 @@ public:
                     _performTimeseriesInsert(opCtx, batch, metadata, std::move(stmtIds));
                 if (auto error =
                         generateError(opCtx, output.result, start + index, errors->size())) {
-                    errors->emplace_back(std::move(*error));
-                    bucketCatalog.abort(batch, output.result.getStatus());
-                    return output.canContinue;
+                    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;
                 }
 
                 invariant(output.result.getValue().getN() == 1,
@@ -828,12 +843,7 @@ public:
 
             if (closedBucket) {
                 // If this write closed a bucket, compress the bucket
-                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;
-                }
+                _performTimeseriesBucketCompression(opCtx, *closedBucket);
             }
             return true;
         } catch (const DBException& ex) {
@@ -841,19 +851,12 @@ public:
             throw;
         }
 
-        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 {
+        bool _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>>>
@@ -866,7 +869,7 @@ public:
             }
 
             if (batchesToCommit.empty()) {
-                return TimeseriesAtomicWriteResult::kSuccess;
+                return true;
             }
 
             // Sort by bucket so that preparing the commit for each batch cannot deadlock.
@@ -892,7 +895,7 @@ public:
                     auto prepareCommitStatus = bucketCatalog.prepareCommit(batch);
                     if (!prepareCommitStatus.isOK()) {
                         abortStatus = prepareCommitStatus;
-                        return TimeseriesAtomicWriteResult::kContinuableError;
+                        return false;
                     }
 
                     if (batch.get()->numPreviouslyCommittedMeasurements() == 0) {
@@ -909,33 +912,26 @@ 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 TimeseriesAtomicWriteResult::kContinuableError;
+                    return false;
                 }
 
                 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 || !compressClosedBuckets) {
+                    if (!closedBucket) {
                         continue;
                     }
 
                     // If this write closed a bucket, compress the bucket
-                    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;
-                    }
+                    _performTimeseriesBucketCompression(opCtx, *closedBucket);
                 }
             } catch (const DBException& ex) {
                 abortStatus = ex.toStatus();
@@ -943,7 +939,7 @@ public:
             }
 
             batchGuard.dismiss();
-            return TimeseriesAtomicWriteResult::kSuccess;
+            return true;
         }
 
         // For sharded time-series collections, we need to use the granularity from the config
@@ -968,10 +964,7 @@ public:
             }
         }
 
-        std::tuple<TimeseriesBatches,
-                   TimeseriesStmtIds,
-                   size_t /* numInserted */,
-                   bool /* canContinue */>
+        std::tuple<TimeseriesBatches, TimeseriesStmtIds, size_t /* numInserted */>
         _insertIntoBucketCatalog(OperationContext* opCtx,
                                  size_t start,
                                  size_t numDocs,
@@ -1011,7 +1004,6 @@ public:
 
             TimeseriesBatches batches;
             TimeseriesStmtIds stmtIds;
-            bool canContinue = true;
 
             auto insert = [&](size_t index) {
                 invariant(start + index < request().getDocuments().size());
@@ -1057,22 +1049,8 @@ 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
-                    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;
+                    _performTimeseriesBucketCompression(opCtx, closedBucket);
                 }
 
                 return true;
@@ -1083,15 +1061,12 @@ public:
             } else {
                 for (size_t i = 0; i < numDocs; i++) {
                     if (!insert(i) && request().getOrdered()) {
-                        return {std::move(batches), std::move(stmtIds), i, canContinue};
+                        return {std::move(batches), std::move(stmtIds), i};
                     }
                 }
             }
 
-            return {std::move(batches),
-                    std::move(stmtIds),
-                    request().getDocuments().size(),
-                    canContinue};
+            return {std::move(batches), std::move(stmtIds), request().getDocuments().size()};
         }
 
         void _getTimeseriesBatchResults(OperationContext* opCtx,
@@ -1153,30 +1128,25 @@ public:
             }
         }
 
-        TimeseriesAtomicWriteResult _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, canContinue] = _insertIntoBucketCatalog(
+        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(
                 opCtx, 0, request().getDocuments().size(), {}, errors, containsRetry);
-            if (!canContinue) {
-                return TimeseriesAtomicWriteResult::kNonContinuableError;
-            }
 
             hangTimeseriesInsertBeforeCommit.pauseWhileSet();
 
-            auto result = _commitTimeseriesBucketsAtomically(
-                opCtx, &batches, std::move(stmtIds), errors, opTime, electionId);
-            if (result != TimeseriesAtomicWriteResult::kSuccess) {
-                return result;
+            if (!_commitTimeseriesBucketsAtomically(
+                    opCtx, &batches, std::move(stmtIds), errors, opTime, electionId)) {
+                return false;
             }
 
             _getTimeseriesBatchResults(
                 opCtx, batches, 0, batches.size(), true, errors, opTime, electionId);
 
-            return TimeseriesAtomicWriteResult::kSuccess;
+            return true;
         }
 
         /**
@@ -1187,19 +1157,9 @@ public:
                                                boost::optional<repl::OpTime>* opTime,
                                                boost::optional<OID>* electionId,
                                                bool* containsRetry) const {
-            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;
+            if (_performOrderedTimeseriesWritesAtomically(
+                    opCtx, errors, opTime, electionId, containsRetry)) {
+                return request().getDocuments().size();
             }
 
             for (size_t i = 0; i < request().getDocuments().size(); ++i) {
@@ -1218,6 +1178,8 @@ 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,
@@ -1227,17 +1189,16 @@ public:
             std::vector<write_ops::WriteError>* errors,
             boost::optional<repl::OpTime>* opTime,
             boost::optional<OID>* electionId,
-            bool* containsRetry) const {
-            auto [batches, bucketStmtIds, _, canContinue] =
+            bool* containsRetry,
+            absl::flat_hash_map<int, int>& retryAttemptsForDup) const {
+            auto [batches, bucketStmtIds, _] =
                 _insertIntoBucketCatalog(opCtx, start, numDocs, indices, errors, containsRetry);
 
             hangTimeseriesInsertBeforeCommit.pauseWhileSet();
 
             std::vector<size_t> docsToRetry;
 
-            if (!canContinue) {
-                return docsToRetry;
-            }
+            bool canContinue = true;
 
             size_t itr = 0;
             for (; itr < batches.size(); ++itr) {
@@ -1246,7 +1207,6 @@ public:
                     auto stmtIds = isTimeseriesWriteRetryable(opCtx)
                         ? std::move(bucketStmtIds[batch->bucket().id])
                         : std::vector<StmtId>{};
-
                     canContinue = _commitTimeseriesBucket(opCtx,
                                                           batch,
                                                           start,
@@ -1255,7 +1215,8 @@ public:
                                                           errors,
                                                           opTime,
                                                           electionId,
-                                                          &docsToRetry);
+                                                          &docsToRetry,
+                                                          retryAttemptsForDup);
                     batch.reset();
                     if (!canContinue) {
                         break;
@@ -1280,9 +1241,20 @@ 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);
+                docsToRetry = _performUnorderedTimeseriesWrites(opCtx,
+                                                                start,
+                                                                numDocs,
+                                                                docsToRetry,
+                                                                errors,
+                                                                opTime,
+                                                                electionId,
+                                                                containsRetry,
+                                                                retryAttemptsForDup);
+                if (!retryAttemptsForDup.empty()) {
+                    BucketCatalog::get(opCtx).resetBucketOIDCounter();
+                }
             } while (!docsToRetry.empty());
         }
 
@@ -1441,19 +1413,25 @@ 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);
+            }
         }
 
         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;
 
@@ -1641,6 +1619,11 @@ 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;