New upstream version 6.0.20upstream

38 files changed, 1366 insertions, 207 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 0321036103f..e20adc6cf10 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -38,6 +38,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/pipeline/change_stream_pre_and_post_images_options',
+        '$BUILD_DIR/mongo/db/query/query_shape/query_shape',
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
@@ -134,8 +135,8 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/concurrency/exception_util',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/storage/key_string',
         'validate_state',
     ]
@@ -150,8 +151,8 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/concurrency/exception_util',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/query/query_knobs',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/storage/record_store_base',
         '$BUILD_DIR/mongo/db/storage/storage_repair_observer',
         'index_repair',
@@ -264,8 +265,8 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/concurrency/exception_util',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/resumable_index_builds_idl',
         '$BUILD_DIR/mongo/db/service_context',
@@ -369,14 +370,15 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/bson/bson_validate',
         '$BUILD_DIR/mongo/db/collection_index_usage_tracker',
         '$BUILD_DIR/mongo/db/commands/server_status_core',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/op_observer',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/record_id_helpers',
         '$BUILD_DIR/mongo/db/repl/drop_pending_collection_reaper',
         '$BUILD_DIR/mongo/db/repl/oplog',
@@ -413,6 +415,7 @@ env.Library(
         'index_key_validate',
         'index_repair',
         'local_oplog_info',
+        'storage_engine_collection_options_flags_parser',
         'throttle_cursor',
         'validate_idl',
         'validate_state',
@@ -442,17 +445,20 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/bson/bson_validate',
         '$BUILD_DIR/mongo/bson/util/bson_column',
         '$BUILD_DIR/mongo/db/concurrency/exception_util',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/index_names',
         '$BUILD_DIR/mongo/db/multi_key_path_tracker',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/record_id_helpers',
         "$BUILD_DIR/mongo/db/service_context",
         '$BUILD_DIR/mongo/db/storage/execution_context',
         '$BUILD_DIR/mongo/db/storage/key_string',
+        '$BUILD_DIR/mongo/db/timeseries/bucket_catalog',
+        '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
         '$BUILD_DIR/mongo/idl/basic_types',
         'catalog_impl',
         'collection_options',
@@ -469,8 +475,8 @@ env.Library(
         'throttle_cursor.cpp',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/util/fail_point',
         'validate_idl',
     ],
@@ -482,8 +488,8 @@ env.Library(
         'local_oplog_info.cpp',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/logical_time',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/repl/optime',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/storage/flow_control',
@@ -496,6 +502,7 @@ env.Library(
 env.Library(
     target='catalog_helpers',
     source=[
+        'backwards_compatible_collection_options_util.cpp',
         'capped_utils.cpp',
         'collection_catalog_helper.cpp',
         'coll_mod.cpp',
@@ -557,8 +564,8 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/collection_index_usage_tracker',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/fts/base_fts',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/s/sharding_api_d',
         '$BUILD_DIR/mongo/db/service_context',
         'index_catalog',
@@ -613,7 +620,21 @@ env.Library(
 )
 
 if wiredtiger:
-    env.CppUnitTest(
+    wtEnv = env.Clone()
+    wtEnv.InjectThirdParty(libraries=["wiredtiger"])
+
+    wtEnv.Library(
+        target="storage_engine_collection_options_flags_parser",
+        source=[
+            "storage_engine_collection_options_flags_parser.cpp",
+        ],
+        LIBDEPS_PRIVATE=[
+            "$BUILD_DIR/mongo/db/storage/wiredtiger/storage_wiredtiger",
+            "$BUILD_DIR/third_party/shim_pcrecpp",
+        ],
+    )
+
+    wtEnv.CppUnitTest(
         target='db_catalog_test',
         source=[
             'capped_utils_test.cpp',
@@ -635,14 +656,17 @@ if wiredtiger:
             'index_spec_validate_test.cpp',
             'multi_index_block_test.cpp',
             'rename_collection_test.cpp',
+            'storage_engine_collection_options_flags_parser_test.cpp',
             'throttle_cursor_test.cpp',
             'validate_state_test.cpp',
         ],
         LIBDEPS=[
             '$BUILD_DIR/mongo/db/auth/authmocks',
+            '$BUILD_DIR/mongo/db/commands/create_command',
             '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
             '$BUILD_DIR/mongo/db/concurrency/lock_manager',
             '$BUILD_DIR/mongo/db/db_raii',
+            '$BUILD_DIR/mongo/db/dbhelpers',
             '$BUILD_DIR/mongo/db/index_builds_coordinator_mongod',
             '$BUILD_DIR/mongo/db/matcher/expressions',
             '$BUILD_DIR/mongo/db/multitenancy',
@@ -653,6 +677,7 @@ if wiredtiger:
             '$BUILD_DIR/mongo/db/query/query_test_service_context',
             '$BUILD_DIR/mongo/db/repl/drop_pending_collection_reaper',
             '$BUILD_DIR/mongo/db/repl/oplog',
+            '$BUILD_DIR/mongo/db/repl/oplog_application',
             '$BUILD_DIR/mongo/db/repl/optime',
             '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
             '$BUILD_DIR/mongo/db/repl/replmocks',
@@ -661,6 +686,7 @@ if wiredtiger:
             '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
             '$BUILD_DIR/mongo/db/service_context_test_fixture',
             '$BUILD_DIR/mongo/db/storage/wiredtiger/storage_wiredtiger',
+            '$BUILD_DIR/mongo/db/timeseries/timeseries_collmod',
             '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
             '$BUILD_DIR/mongo/unittest/unittest',
             '$BUILD_DIR/mongo/util/clock_source_mock',
@@ -679,6 +705,7 @@ if wiredtiger:
             'index_builds_manager',
             'index_key_validate',
             'multi_index_block',
+            'storage_engine_collection_options_flags_parser',
             'throttle_cursor',
             'validate_idl',
             'validate_state',
diff --git a/src/mongo/db/catalog/backwards_compatible_collection_options_util.cpp b/src/mongo/db/catalog/backwards_compatible_collection_options_util.cpp
new file mode 100644
index 00000000000..d966d6108c6
--- /dev/null
+++ b/src/mongo/db/catalog/backwards_compatible_collection_options_util.cpp
@@ -0,0 +1,86 @@
+/**
+ *    Copyright (C) 2024-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.
+ */
+
+// TODO SERVER-92265 evaluate getting rid of this util
+
+#include "mongo/db/catalog/backwards_compatible_collection_options_util.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/bson/simple_bsonobj_comparator.h"
+#include "mongo/db/repl/oplog_entry.h"
+
+namespace mongo {
+namespace backwards_compatible_collection_options {
+std::pair<BSONObj, BSONObj> getCollModCmdAndAdditionalO2Field(const BSONObj& collModCmd) {
+    const BSONObj collModCmdStrippedBackwardsIncompatibleParams =
+        collModCmd.removeFields(kBackwardsCompatibleCollectionOptions);
+    if (SimpleBSONObjComparator::kInstance.evaluate(collModCmdStrippedBackwardsIncompatibleParams ==
+                                                    collModCmd)) {
+        return {collModCmd, BSONObj()};
+    }
+
+    const BSONObj backwardsIncompatibleFields = [&]() {
+        BSONObjBuilder bob;
+        for (auto [fieldName, elem] : collModCmd) {
+            if (kBackwardsCompatibleCollectionOptions.count(fieldName.toString())) {
+                bob.append(elem);
+            }
+        }
+        return bob.obj();
+    }();
+
+    return {collModCmdStrippedBackwardsIncompatibleParams, backwardsIncompatibleFields};
+}
+
+BSONObj parseCollModCmdFromOplogEntry(const repl::OplogEntry& entry) {
+    uassert(ErrorCodes::IllegalOperation,
+            str::stream() << "Can't extract `collMod` command from non-collMod oplog entry: "
+                          << entry.toBSONForLogging(),
+            entry.getCommandType() == repl::OplogEntry::CommandType::kCollMod);
+
+    if (!entry.getObject2()) {
+        return entry.getObject();
+    }
+
+    BSONObj incompatibleFields = entry.getObject2()->getObjectField(additionalCollModO2Field);
+    if (incompatibleFields.isEmpty()) {
+        return entry.getObject();
+    }
+
+    // Only consider backwards incompatible fields supported in the current [sub-]version
+    for (auto [fieldName, elem] : incompatibleFields) {
+        if (!kBackwardsCompatibleCollectionOptions.count(fieldName.toString())) {
+            incompatibleFields = incompatibleFields.removeField(fieldName);
+        }
+    }
+
+    return entry.getObject().addFields(incompatibleFields);
+}
+
+}  // namespace backwards_compatible_collection_options
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/backwards_compatible_collection_options_util.h b/src/mongo/db/catalog/backwards_compatible_collection_options_util.h
new file mode 100644
index 00000000000..d5c47b45e6d
--- /dev/null
+++ b/src/mongo/db/catalog/backwards_compatible_collection_options_util.h
@@ -0,0 +1,130 @@
+/**
+ *    Copyright (C) 2024-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.
+ */
+
+// TODO SERVER-92265 evaluate getting rid of this file
+
+#include "mongo/db/repl/oplog_entry.h"
+
+namespace mongo {
+
+/**
+ *
+ * This utility is providing primitives to manage catalog parameters for which the actual value may
+ * have been missing or incorrect in previous mongod [sub-]versions. It is only meant to fix catalog
+ * issues in existing versions and must NOT be used for implementing new features.
+ *
+ * The functions under this namespace are offering an abstraction to work around the following
+ * limitations:
+ * - collMod command is strict (so can't simply add a parameter to previous mongod [sub-]versions)
+ * - Collection options are strict (so can't simply add an option to previous mongod [sub-]versions)
+ *
+ */
+namespace backwards_compatible_collection_options {
+
+const std::string kTimeseriesBucketsMayHaveMixedSchemaData =
+    "timeseriesBucketsMayHaveMixedSchemaData";
+
+const std::string additionalCollModO2Field = "backwardsIncompatibleCollModParameters";
+
+/**
+ * Backwards incompatible catalog parameters for which the actual value may have been missing or
+ * incorrect in previous mongod [sub-]versions.
+ */
+const std::set<std::string> kBackwardsCompatibleCollectionOptions{
+    kTimeseriesBucketsMayHaveMixedSchemaData};
+
+/**
+ * Strips backwards incompatible fields from a collMod command and places them into a
+ * different BSON object.
+ *
+ * Returns two BSON objects:
+ * - A backwards compatible collMod oplog entry (not to generate crashes when applied by
+ * incompatible mongod [sub-]versions).
+ * - A field meant to be added to the `o2` sub-object (parsable by new mongod [sub-]versions).
+ *
+ * Example:
+ *
+ * - Original command:
+ * {"collMod":"testdb.system.buckets.testcoll", "timeseriesBucketsMayHaveMixedSchemaData":true }
+ *
+ * - Expected oplog entry with `timeseriesBucketsMayHaveMixedSchemaData` backwards incompatible
+ * collMod parameter.
+ *
+ * {
+ *    "oplogEntry":{
+ *        "op":"c",
+ *        "ns":"testdb.$cmd",
+ *        "ui":"UUID(""7302d025-cb9c-4a16-9222-0d5aeefbc039"")",
+ *        "o":{
+ *            "collMod":"system.buckets.testcoll"
+ *         },
+ *        "o2":{
+ *            "collectionOptions_old":{
+ *                "uuid": UUID("7302d025-cb9c-4a16-9222-0d5aeefbc039"),
+ *                "Validator":{ "...REDACTED..." },
+ *                "clusteredIndex":true,
+ *                "timeseries":{
+ *                    "timeField":"t",
+ *                    “granularity":"seconds",
+ *                    "bucketMaxSpanSeconds":3600
+ *                }
+ *            },
+ *            "backwardsIncompatibleCollModParameters":{
+ *                "timeseriesBucketsMayHaveMixedSchemaData":true
+ *            }
+ *        },
+ *        "ts":Timestamp(1720003401,4),
+ *        "t":1,
+ *        "v":2,
+ *        "wall":new Date(1720003401165)
+ *   }
+ * }
+ *
+ */
+std::pair<BSONObj, BSONObj> getCollModCmdAndAdditionalO2Field(const BSONObj& collModCmd);
+
+/**
+ * Rebuilds a collMod command from an oplog entry.
+ *
+ * Returns a bson object:
+ * - A collMod command inclusive of potential backwards incompatible fields present in the oplog
+ * entry's `o2` sub-object.
+ *
+ * Example: collMod command parsed from the sample oplog entry documented above.
+ *
+ * {
+ *     "collMod":"system.buckets.testcoll",
+ *     "timeseriesBucketsMayHaveMixedSchemaData":true
+ * }
+ *
+ */
+BSONObj parseCollModCmdFromOplogEntry(const repl::OplogEntry& entry);
+
+}  // namespace backwards_compatible_collection_options
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 3bb76fee38b..2c9f03053d0 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -118,6 +118,7 @@ struct ParsedCollModRequest {
     bool dryRun = false;
     boost::optional<long long> cappedSize;
     boost::optional<long long> cappedMax;
+    boost::optional<bool> timeseriesBucketsMayHaveMixedSchemaData;
 };
 
 Status getNotSupportedOnViewError(StringData fieldName) {
@@ -400,6 +401,11 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
         }
 
         if (cmdIndex.getPrepareUnique()) {
+            // Check if prepareUnique is being set on a time-series collection.
+            if (isTimeseries) {
+                return {ErrorCodes::InvalidOptions,
+                        "cannot set 'prepareUnique' for indexes of a time-series collection."};
+            }
             parsed.numModifications++;
             // Attempting to modify with the same value should be treated as a no-op.
             if (cmrIndex->idx->prepareUnique() == *cmdIndex.getPrepareUnique() ||
@@ -605,6 +611,17 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
         timeseries->serialize(&subObjBuilder);
     }
 
+    if (auto mixedSchema = cmr.getTimeseriesBucketsMayHaveMixedSchemaData()) {
+        if (!isTimeseries) {
+            return getOnlySupportedOnTimeseriesError(
+                CollMod::kTimeseriesBucketsMayHaveMixedSchemaDataFieldName);
+        }
+
+        parsed.timeseriesBucketsMayHaveMixedSchemaData = mixedSchema;
+        oplogEntryBuilder.append(CollMod::kTimeseriesBucketsMayHaveMixedSchemaDataFieldName,
+                                 *mixedSchema);
+    }
+
     if (auto& dryRun = cmr.getDryRun()) {
         parsed.dryRun = *dryRun;
         // The dry run option should never be included in a collMod oplog entry.
@@ -930,6 +947,11 @@ Status _collModInternal(OperationContext* opCtx,
                                             *cmd.getExpireAfterSeconds());
         }
 
+        if (auto mixedSchema = cmrNew.timeseriesBucketsMayHaveMixedSchemaData) {
+            coll.getWritableCollection(opCtx)->setTimeseriesBucketsMayHaveMixedSchemaData(
+                opCtx, mixedSchema);
+        }
+
         // Handle index modifications.
         processCollModIndexRequest(
             opCtx, &coll, cmrNew.indexRequest, &indexCollModInfo, result, mode);
@@ -985,24 +1007,24 @@ Status _collModInternal(OperationContext* opCtx,
         // (Generic FCV reference): TODO SERVER-60912: When kLastLTS is 6.0, remove this FCV-gated
         // upgrade/downgrade code.
         const auto currentVersion = serverGlobalParams.featureCompatibility.getVersion();
-        if (coll->getTimeseriesOptions() && !coll->getTimeseriesBucketsMayHaveMixedSchemaData() &&
-            (currentVersion == multiversion::GenericFCV::kUpgradingFromLastLTSToLatest ||
-             currentVersion == multiversion::GenericFCV::kLatest)) {
-            // (Generic FCV reference): While upgrading the FCV from kLastLTS to kLatest, collMod is
-            // called as part of the upgrade process to add the
-            // 'timeseriesBucketsMayHaveMixedSchemaData=true' catalog entry flag for time-series
-            // collections that are missing the flag. This indicates that the time-series collection
-            // existed in earlier server versions and may have mixed-schema data.
-            coll.getWritableCollection(opCtx)->setTimeseriesBucketsMayHaveMixedSchemaData(opCtx,
-                                                                                          true);
-        } else if (coll->getTimeseriesBucketsMayHaveMixedSchemaData() &&
-                   (currentVersion == multiversion::GenericFCV::kDowngradingFromLatestToLastLTS ||
-                    currentVersion == multiversion::GenericFCV::kLastLTS)) {
-            // (Generic FCV reference): While downgrading the FCV to kLastLTS, collMod is called as
-            // part of the downgrade process to remove the 'timeseriesBucketsMayHaveMixedSchemaData'
-            // catalog entry flag for time-series collections that have the flag.
-            coll.getWritableCollection(opCtx)->setTimeseriesBucketsMayHaveMixedSchemaData(
-                opCtx, boost::none);
+        if (coll->getTimeseriesOptions()) {
+            if (currentVersion == multiversion::GenericFCV::kUpgradingFromLastLTSToLatest) {
+                // (Generic FCV reference): While upgrading the FCV from kLastLTS to kLatest,
+                // collMod is called as part of the upgrade process to add the
+                // 'timeseriesBucketsMayHaveMixedSchemaData=true' catalog entry flag for time-series
+                // collections that are missing the flag. This indicates that the time-series
+                // collection existed in earlier server versions and may have mixed-schema data.
+                coll.getWritableCollection(opCtx)->setTimeseriesBucketsMayHaveMixedSchemaData(opCtx,
+                                                                                              true);
+            } else if (currentVersion ==
+                       multiversion::GenericFCV::kDowngradingFromLatestToLastLTS) {
+                // (Generic FCV reference): While downgrading the FCV to kLastLTS, collMod is called
+                // as part of the downgrade process to remove the
+                // 'timeseriesBucketsMayHaveMixedSchemaData' catalog entry flag for time-series
+                // collections that have the flag.
+                coll.getWritableCollection(opCtx)->setTimeseriesBucketsMayHaveMixedSchemaData(
+                    opCtx, boost::none);
+            }
         }
 
         // Only observe non-view collMods, as view operations are observed as operations on the
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index f64af95c32a..9be52a11a11 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -560,9 +560,10 @@ public:
                                                             boost::optional<bool> setting) = 0;
 
     /**
-     * Returns true if the passed in time-series bucket document contains mixed-schema data.
+     * Returns true if the passed in time-series bucket document contains mixed-schema data. Returns
+     * a non-OK status if the bucket's min/max is malformed.
      */
-    virtual bool doesTimeseriesBucketsDocContainMixedSchemaData(
+    virtual StatusWith<bool> doesTimeseriesBucketsDocContainMixedSchemaData(
         const BSONObj& bucketsDoc) const = 0;
 
     /**
diff --git a/src/mongo/db/catalog/collection_catalog_helper.cpp b/src/mongo/db/catalog/collection_catalog_helper.cpp
index 1bdd85ce3e6..ed816f2e1cf 100644
--- a/src/mongo/db/catalog/collection_catalog_helper.cpp
+++ b/src/mongo/db/catalog/collection_catalog_helper.cpp
@@ -66,6 +66,7 @@ void forEachCollectionFromDb(OperationContext* opCtx,
                              CollectionCatalog::CollectionInfoFn predicate) {
 
     auto catalogForIteration = CollectionCatalog::get(opCtx);
+    size_t collectionCount = 0;
     for (auto&& coll : catalogForIteration->range(tenantDbName)) {
         auto uuid = coll->uuid();
         if (predicate && !catalogForIteration->checkIfCollectionSatisfiable(uuid, predicate)) {
@@ -99,7 +100,14 @@ void forEachCollectionFromDb(OperationContext* opCtx,
         if (!callback(collection))
             break;
 
+        // This was a rough heuristic that was found that 400 collections would take 100
+        // milliseconds with calling checkForInterrupt() (with freeStorage: 1).
+        // We made the checkForInterrupt() occur after 200 collections to be conservative.
+        if (!(collectionCount % 200)) {
+            opCtx->checkForInterrupt();
+        }
         hangBeforeGettingNextCollection.pauseWhileSet();
+        collectionCount += 1;
     }
 }
 
diff --git a/src/mongo/db/catalog/collection_compact.cpp b/src/mongo/db/catalog/collection_compact.cpp
index b6cc7cb444d..549aef5d9f1 100644
--- a/src/mongo/db/catalog/collection_compact.cpp
+++ b/src/mongo/db/catalog/collection_compact.cpp
@@ -47,64 +47,18 @@ namespace mongo {
 
 using logv2::LogComponent;
 
-namespace {
-
-CollectionPtr getCollectionForCompact(OperationContext* opCtx,
-                                      const NamespaceString& collectionNss) {
-    invariant(opCtx->lockState()->isCollectionLockedForMode(collectionNss, MODE_IX));
-
-    auto collectionCatalog = CollectionCatalog::get(opCtx);
-    CollectionPtr collection = collectionCatalog->lookupCollectionByNamespace(opCtx, collectionNss);
-
-    if (!collection) {
-        std::shared_ptr<const ViewDefinition> view =
-            collectionCatalog->lookupView(opCtx, collectionNss);
-        uassert(ErrorCodes::CommandNotSupportedOnView, "can't compact a view", !view);
-        uasserted(ErrorCodes::NamespaceNotFound, "collection does not exist");
-    }
-
-    return collection;
-}
-
-}  // namespace
-
-StatusWith<int64_t> compactCollection(OperationContext* opCtx,
-                                      const NamespaceString& collectionNss) {
-    AutoGetDb autoDb(opCtx, collectionNss.db(), MODE_IX);
-    Database* database = autoDb.getDb();
-    uassert(ErrorCodes::NamespaceNotFound, "database does not exist", database);
-
-    // The collection lock will be upgraded to an exclusive lock if the record store does not
-    // support online compaction.
-    boost::optional<Lock::CollectionLock> collLk;
-    collLk.emplace(opCtx, collectionNss, MODE_IX);
-
-    CollectionPtr collection = getCollectionForCompact(opCtx, collectionNss);
+StatusWith<int64_t> compactCollection(OperationContext* opCtx, const CollectionPtr& collection) {
     DisableDocumentValidation validationDisabler(opCtx);
 
+    auto collectionNss = collection->ns();
     auto recordStore = collection->getRecordStore();
 
-    OldClientContext ctx(opCtx, collectionNss.ns());
-
     if (!recordStore->compactSupported())
         return Status(ErrorCodes::CommandNotSupported,
                       str::stream() << "cannot compact collection with record store: "
                                     << recordStore->name());
 
-    if (!recordStore->supportsOnlineCompaction()) {
-        // Storage engines that disallow online compaction should compact under an exclusive lock.
-        collLk.emplace(opCtx, collectionNss, MODE_X);
-
-        // Ensure the collection was not dropped during the re-lock.
-        collection = getCollectionForCompact(opCtx, collectionNss);
-        recordStore = collection->getRecordStore();
-    }
-
-    LOGV2_OPTIONS(20284,
-                  {LogComponent::kCommand},
-                  "compact {namespace} begin",
-                  "Compact begin",
-                  "namespace"_attr = collectionNss);
+    LOGV2_OPTIONS(20284, {LogComponent::kCommand}, "Compact begin", logAttrs(collectionNss));
 
     auto oldTotalSize = recordStore->storageSize(opCtx) + collection->getIndexSize(opCtx);
     auto indexCatalog = collection->getIndexCatalog();
diff --git a/src/mongo/db/catalog/collection_compact.h b/src/mongo/db/catalog/collection_compact.h
index ab22cf07155..c6307360bc0 100644
--- a/src/mongo/db/catalog/collection_compact.h
+++ b/src/mongo/db/catalog/collection_compact.h
@@ -40,7 +40,6 @@ namespace mongo {
  * Returns the number of bytes of stable storage and index size that were freed. If the total
  * size decreased, the return value is positive. Otherwise, the return value is negative.
  */
-StatusWith<int64_t> compactCollection(OperationContext* opCtx,
-                                      const NamespaceString& collectionNss);
+StatusWith<int64_t> compactCollection(OperationContext* opCtx, const CollectionPtr& collection);
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 298772562db..cdf3730fa1e 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -40,6 +40,7 @@
 #include "mongo/bson/simple_bsonobj_comparator.h"
 #include "mongo/crypto/fle_crypto.h"
 #include "mongo/db/auth/security_token.h"
+#include "mongo/db/catalog/backwards_compatible_collection_options_util.h"
 #include "mongo/db/catalog/catalog_stats.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_options.h"
@@ -48,6 +49,7 @@
 #include "mongo/db/catalog/index_consistency.h"
 #include "mongo/db/catalog/index_key_validate.h"
 #include "mongo/db/catalog/local_oplog_info.h"
+#include "mongo/db/catalog/storage_engine_collection_options_flags_parser.h"
 #include "mongo/db/catalog/uncommitted_multikey.h"
 #include "mongo/db/clientcursor.h"
 #include "mongo/db/commands/server_status_metric.h"
@@ -352,35 +354,42 @@ bool indexTypeSupportsPathLevelMultikeyTracking(StringData accessMethod) {
     return accessMethod == IndexNames::BTREE || accessMethod == IndexNames::GEO_2DSPHERE;
 }
 
-bool doesMinMaxHaveMixedSchemaData(const BSONObj& min, const BSONObj& max) {
+StatusWith<bool> doesMinMaxHaveMixedSchemaData(const BSONObj& min, const BSONObj& max) {
     auto minIt = min.begin();
     auto minEnd = min.end();
     auto maxIt = max.begin();
     auto maxEnd = max.end();
 
     while (minIt != minEnd && maxIt != maxEnd) {
-        bool typeMatch = minIt->canonicalType() == maxIt->canonicalType();
-        if (!typeMatch) {
+        // The 'control.min' and 'control.max' fields have the same ordering.
+        if (minIt->fieldNameStringData() != maxIt->fieldNameStringData()) {
+            return Status{
+                ErrorCodes::BadValue,
+                "Encountered inconsistent field name ordering in time-series bucket min/max"};
+        }
+
+        if (minIt->canonicalType() != maxIt->canonicalType()) {
             return true;
         } else if (minIt->type() == Object) {
-            // The 'control.min' and 'control.max' fields have the same ordering.
-            invariant(minIt->fieldNameStringData() == maxIt->fieldNameStringData());
-            if (doesMinMaxHaveMixedSchemaData(minIt->Obj(), maxIt->Obj())) {
-                return true;
+            auto result = doesMinMaxHaveMixedSchemaData(minIt->Obj(), maxIt->Obj());
+            if (!result.isOK() || result.getValue()) {
+                return result;
             }
         } else if (minIt->type() == Array) {
-            if (doesMinMaxHaveMixedSchemaData(minIt->Obj(), maxIt->Obj())) {
-                return true;
+            auto result = doesMinMaxHaveMixedSchemaData(minIt->Obj(), maxIt->Obj());
+            if (!result.isOK() || result.getValue()) {
+                return result;
             }
         }
 
-        invariant(typeMatch);
         minIt++;
         maxIt++;
     }
 
-    // The 'control.min' and 'control.max' fields have the same cardinality.
-    invariant(minIt == minEnd && maxIt == maxEnd);
+    if (minIt != minEnd || maxIt != maxEnd) {
+        return Status{ErrorCodes::BadValue,
+                      "Encountered extra field(s) in time-series bucket min/max"};
+    }
 
     return false;
 }
@@ -1589,6 +1598,19 @@ bool CollectionImpl::isTemporary() const {
 }
 
 boost::optional<bool> CollectionImpl::getTimeseriesBucketsMayHaveMixedSchemaData() const {
+    if (!getTimeseriesOptions()) {
+        return boost::none;
+    }
+
+    // If present, reuse storageEngine options to work around the issue described in SERVER-91194
+    boost::optional<bool> optBackwardsCompatibleFlag = getFlagFromStorageEngineBson(
+        _metadata->options.storageEngine,
+        backwards_compatible_collection_options::kTimeseriesBucketsMayHaveMixedSchemaData);
+    if (optBackwardsCompatibleFlag) {
+        return *optBackwardsCompatibleFlag;
+    }
+
+    // Else, fallback to legacy parameter
     return _metadata->timeseriesBucketsMayHaveMixedSchemaData;
 }
 
@@ -1604,11 +1626,21 @@ void CollectionImpl::setTimeseriesBucketsMayHaveMixedSchemaData(OperationContext
                 "setting"_attr = setting);
 
     _writeMetadata(opCtx, [&](BSONCollectionCatalogEntry::MetaData& md) {
+        // Reuse storageEngine options to work around the issue described in SERVER-91194
+        if (setting.has_value()) {
+            md.options.storageEngine = setFlagToStorageEngineBson(
+                md.options.storageEngine,
+                backwards_compatible_collection_options::kTimeseriesBucketsMayHaveMixedSchemaData,
+                *setting);
+        }
+
+        // Also update legacy parameter for compatibility when downgrading to older sub-versions
+        // only relying on this option (best-effort because it may be lost due to SERVER-91194)
         md.timeseriesBucketsMayHaveMixedSchemaData = setting;
     });
 }
 
-bool CollectionImpl::doesTimeseriesBucketsDocContainMixedSchemaData(
+StatusWith<bool> CollectionImpl::doesTimeseriesBucketsDocContainMixedSchemaData(
     const BSONObj& bucketsDoc) const {
     if (!getTimeseriesOptions()) {
         return false;
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index a8e7dd99863..58662679df1 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -329,7 +329,8 @@ public:
     void setTimeseriesBucketsMayHaveMixedSchemaData(OperationContext* opCtx,
                                                     boost::optional<bool> setting) final;
 
-    bool doesTimeseriesBucketsDocContainMixedSchemaData(const BSONObj& bucketsDoc) const final;
+    StatusWith<bool> doesTimeseriesBucketsDocContainMixedSchemaData(
+        const BSONObj& bucketsDoc) const final;
 
     bool getRequiresTimeseriesExtendedRangeSupport() const final;
     void setRequiresTimeseriesExtendedRangeSupport(OperationContext* opCtx) const final;
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index a9b7fe8c27b..820ed47cdf4 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -253,7 +253,8 @@ public:
         std::abort();
     }
 
-    bool doesTimeseriesBucketsDocContainMixedSchemaData(const BSONObj& bucketsDoc) const {
+    StatusWith<bool> doesTimeseriesBucketsDocContainMixedSchemaData(
+        const BSONObj& bucketsDoc) const {
         std::abort();
     }
 
diff --git a/src/mongo/db/catalog/collection_test.cpp b/src/mongo/db/catalog/collection_test.cpp
index fd73c68bad8..00c131756ac 100644
--- a/src/mongo/db/catalog/collection_test.cpp
+++ b/src/mongo/db/catalog/collection_test.cpp
@@ -403,7 +403,9 @@ TEST_F(CollectionTest, CheckTimeseriesBucketDocsForMixedSchemaData) {
                                              "max" : { "x" : [ 2, 3 ] } } })")};
 
     for (const auto& controlDoc : mixedSchemaControlDocs) {
-        ASSERT_TRUE(coll->doesTimeseriesBucketsDocContainMixedSchemaData(controlDoc));
+        auto mixedSchema = coll->doesTimeseriesBucketsDocContainMixedSchemaData(controlDoc);
+        ASSERT_OK(mixedSchema) << controlDoc;
+        ASSERT_TRUE(mixedSchema.getValue()) << controlDoc;
     }
 
     std::vector<BSONObj> nonMixedSchemaControlDocs = {
@@ -460,7 +462,27 @@ TEST_F(CollectionTest, CheckTimeseriesBucketDocsForMixedSchemaData) {
 
 
     for (const auto& controlDoc : nonMixedSchemaControlDocs) {
-        ASSERT_FALSE(coll->doesTimeseriesBucketsDocContainMixedSchemaData(controlDoc));
+        auto mixedSchema = coll->doesTimeseriesBucketsDocContainMixedSchemaData(controlDoc);
+        ASSERT_OK(mixedSchema) << controlDoc;
+        ASSERT_FALSE(mixedSchema.getValue()) << controlDoc;
+    }
+
+    std::vector<BSONObj> malformedControlDocs = {
+        // Inconsistent field name ordering
+        ::mongo::fromjson(R"({ "control" : { "min" : { "x" : 1, "y" : 1 },
+                                             "max" : { "y" : 2, "x" : 2 } } })"),
+
+        // Extra field in min
+        ::mongo::fromjson(R"({ "control" : { "min" : { "x" : 1, "y" : 1 },
+                                             "max" : { "x" : 2 } } })"),
+
+        // Extra field in max
+        ::mongo::fromjson(R"({ "control" : { "min" : { "y" : 1 },
+                                             "max" : { "y" : 2, "x" : 2 } } })")};
+
+    for (const auto& controlDoc : malformedControlDocs) {
+        ASSERT_NOT_OK(coll->doesTimeseriesBucketsDocContainMixedSchemaData(controlDoc))
+            << controlDoc;
     }
 }
 
diff --git a/src/mongo/db/catalog/collection_validation.cpp b/src/mongo/db/catalog/collection_validation.cpp
index 015bf7e4b5a..9d0c487b41a 100644
--- a/src/mongo/db/catalog/collection_validation.cpp
+++ b/src/mongo/db/catalog/collection_validation.cpp
@@ -162,7 +162,8 @@ void _gatherIndexEntryErrors(OperationContext* opCtx,
         ValidateResults tempValidateResults;
         BSONObjBuilder tempBuilder;
 
-        indexValidator->traverseRecordStore(opCtx, &tempValidateResults, &tempBuilder);
+        indexValidator->traverseRecordStore(
+            opCtx, &tempValidateResults, &tempBuilder, validateState->validationVersion());
     }
 
     LOGV2_OPTIONS(
@@ -454,8 +455,7 @@ void _validateCatalogEntry(OperationContext* opCtx,
             index_key_validate::validateIndexSpec(opCtx, indexEntry->descriptor()->infoObj())
                 .getStatus();
         if (!status.isOK()) {
-            results->valid = false;
-            results->errors.push_back(
+            results->warnings.push_back(
                 fmt::format("The index specification for index '{}' contains invalid fields. {}. "
                             "Run the 'collMod' command on the collection without any arguments "
                             "to fix the invalid index options",
@@ -621,6 +621,14 @@ Status validate(OperationContext* opCtx,
         opCtx->recoveryUnit()->abandonSnapshot();
         opCtx->recoveryUnit()->setPrepareConflictBehavior(oldPrepareConflictBehavior);
     });
+
+    // Relax corruption detection so that we log and continue scanning instead of failing early.
+    auto oldDataCorruptionMode = opCtx->recoveryUnit()->getDataCorruptionDetectionMode();
+    opCtx->recoveryUnit()->setDataCorruptionDetectionMode(
+        DataCorruptionDetectionMode::kLogAndContinue);
+    ON_BLOCK_EXIT(
+        [&] { opCtx->recoveryUnit()->setDataCorruptionDetectionMode(oldDataCorruptionMode); });
+
     if (validateState.fixErrors()) {
         // Note: cannot set PrepareConflictBehavior here, since the validate command with repair
         // needs kIngnoreConflictsAllowWrites, but validate repair at startup cannot set that here
@@ -697,7 +705,8 @@ Status validate(OperationContext* opCtx,
         // the collection. For clustered collections, the validator also verifies that the
         // record key (RecordId) matches the cluster key field in the record value (document's
         // cluster key).
-        indexValidator.traverseRecordStore(opCtx, results, output);
+        indexValidator.traverseRecordStore(
+            opCtx, results, output, additionalOptions.validationVersion);
 
         // Pause collection validation while a lock is held and between collection and index data
         // validation.
@@ -771,8 +780,7 @@ Status validate(OperationContext* opCtx,
             return e.toStatus();
         }
         string err = str::stream() << "exception during collection validation: " << e.toString();
-        results->errors.push_back(err);
-        results->valid = false;
+        results->warnings.push_back(err);
         LOGV2_OPTIONS(5160302,
                       {LogComponent::kIndex},
                       "Validation failed due to exception",
diff --git a/src/mongo/db/catalog/collection_validation.h b/src/mongo/db/catalog/collection_validation.h
index 2a0ea7a6929..f68258fee14 100644
--- a/src/mongo/db/catalog/collection_validation.h
+++ b/src/mongo/db/catalog/collection_validation.h
@@ -29,6 +29,9 @@
 
 #pragma once
 
+#include "mongo/base/status.h"
+#include "mongo/bson/bson_validate.h"
+#include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/catalog/validate_results.h"
 #include "mongo/db/namespace_string.h"
 
@@ -90,7 +93,7 @@ enum class RepairMode {
  * Additional validation options that can run in any mode.
  */
 struct AdditionalOptions {
-    bool warnOnSchemaValidation = false;  // only warn on schema validation failure
+    ValidationVersion validationVersion = currentValidationVersion;
 };
 
 /**
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index 59308be7233..c4669cd85b2 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -261,7 +261,7 @@ void _testDropCollectionThrowsExceptionIfThereAreIndexesInProgress(OperationCont
             collection->ns(), indexInfoObj, IndexBuildMethod::kHybrid, UUID::gen());
         {
             WriteUnitOfWork wuow(opCtx);
-            ASSERT_OK(indexBuildBlock->init(opCtx, collection));
+            ASSERT_OK(indexBuildBlock->init(opCtx, collection, /*forRecovery=*/false));
             wuow.commit();
         }
         ON_BLOCK_EXIT([&indexBuildBlock, opCtx, collection] {
diff --git a/src/mongo/db/catalog/index_build_block.cpp b/src/mongo/db/catalog/index_build_block.cpp
index a5645e9de72..34dbf0c49ad 100644
--- a/src/mongo/db/catalog/index_build_block.cpp
+++ b/src/mongo/db/catalog/index_build_block.cpp
@@ -75,6 +75,11 @@ void IndexBuildBlock::_completeInit(OperationContext* opCtx, Collection* collect
         .registerIndex(desc->indexName(),
                        desc->keyPattern(),
                        IndexFeatures::make(desc, collection->ns().isOnInternalDb()));
+    opCtx->recoveryUnit()->onRollback(
+        [collectionDecorations = collection->getSharedDecorations(), indexName = _indexName] {
+            CollectionIndexUsageTrackerDecoration::get(collectionDecorations)
+                .unregisterIndex(indexName);
+        });
 }
 
 Status IndexBuildBlock::initForResume(OperationContext* opCtx,
@@ -122,7 +127,7 @@ Status IndexBuildBlock::initForResume(OperationContext* opCtx,
     return Status::OK();
 }
 
-Status IndexBuildBlock::init(OperationContext* opCtx, Collection* collection) {
+Status IndexBuildBlock::init(OperationContext* opCtx, Collection* collection, bool forRecovery) {
     // Being in a WUOW means all timestamping responsibility can be pushed up to the caller.
     invariant(opCtx->lockState()->inAWriteUnitOfWork());
 
@@ -150,14 +155,25 @@ Status IndexBuildBlock::init(OperationContext* opCtx, Collection* collection) {
             !replCoord->getMemberState().primary() && isBackgroundIndex;
     }
 
-    // Setup on-disk structures.
-    Status status = collection->prepareForIndexBuild(
-        opCtx, descriptor.get(), _buildUUID, isBackgroundSecondaryBuild);
-    if (!status.isOK())
-        return status;
+    if (!forRecovery) {
+        // Setup on-disk structures. We skip this during startup recovery for unfinished indexes as
+        // everything is already in-place.
+        Status status = collection->prepareForIndexBuild(
+            opCtx, descriptor.get(), _buildUUID, isBackgroundSecondaryBuild);
+        if (!status.isOK())
+            return status;
+    }
 
-    auto indexCatalogEntry = collection->getIndexCatalog()->createIndexEntry(
-        opCtx, collection, std::move(descriptor), CreateIndexEntryFlags::kNone);
+    auto indexCatalog = collection->getIndexCatalog();
+    IndexCatalogEntry* indexCatalogEntry = nullptr;
+    if (forRecovery) {
+        auto desc = indexCatalog->findIndexByName(
+            opCtx, _indexName, IndexCatalog::InclusionPolicy::kUnfinished);
+        indexCatalogEntry = desc->getEntry();
+    } else {
+        indexCatalogEntry = indexCatalog->createIndexEntry(
+            opCtx, collection, std::move(descriptor), CreateIndexEntryFlags::kNone);
+    }
 
     if (_method == IndexBuildMethod::kHybrid) {
         _indexBuildInterceptor = std::make_unique<IndexBuildInterceptor>(opCtx, indexCatalogEntry);
diff --git a/src/mongo/db/catalog/index_build_block.h b/src/mongo/db/catalog/index_build_block.h
index b1086aac5ed..2b94849b46c 100644
--- a/src/mongo/db/catalog/index_build_block.h
+++ b/src/mongo/db/catalog/index_build_block.h
@@ -62,7 +62,7 @@ public:
      *
      * Must be called from within a `WriteUnitOfWork`
      */
-    Status init(OperationContext* opCtx, Collection* collection);
+    Status init(OperationContext* opCtx, Collection* collection, bool forRecovery);
 
     /**
      * Makes sure that an entry for the index was created at startup in the IndexCatalog. Returns
diff --git a/src/mongo/db/catalog/index_build_entry_test.cpp b/src/mongo/db/catalog/index_build_entry_test.cpp
index ffa6a870874..d53c7e0046c 100644
--- a/src/mongo/db/catalog/index_build_entry_test.cpp
+++ b/src/mongo/db/catalog/index_build_entry_test.cpp
@@ -32,6 +32,7 @@
 #include <string>
 #include <vector>
 
+#include "mongo/bson/bson_validate.h"
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/bson/bsontypes.h"
@@ -124,7 +125,7 @@ TEST(IndexBuildEntryTest, SerializeAndDeserialize) {
     entry.setCommitReadyMembers(generateCommitReadyMembers(3));
 
     BSONObj obj = entry.toBSON();
-    ASSERT_TRUE(obj.valid());
+    ASSERT_TRUE(validateBSON(obj).isOK());
 
     IDLParserErrorContext ctx("IndexBuildsEntry Parser");
     IndexBuildEntry rebuiltEntry = IndexBuildEntry::parse(ctx, obj);
diff --git a/src/mongo/db/catalog/index_builds_manager.cpp b/src/mongo/db/catalog/index_builds_manager.cpp
index a5db8f4536f..0c44c44de48 100644
--- a/src/mongo/db/catalog/index_builds_manager.cpp
+++ b/src/mongo/db/catalog/index_builds_manager.cpp
@@ -113,7 +113,8 @@ Status IndexBuildsManager::setUpIndexBuild(OperationContext* opCtx,
     std::vector<BSONObj> indexes;
     try {
         indexes = writeConflictRetry(opCtx, "IndexBuildsManager::setUpIndexBuild", nss.ns(), [&]() {
-            return uassertStatusOK(builder->init(opCtx, collection, specs, onInit, resumeInfo));
+            return uassertStatusOK(
+                builder->init(opCtx, collection, specs, onInit, options.forRecovery, resumeInfo));
         });
     } catch (const DBException& ex) {
         return ex.toStatus();
diff --git a/src/mongo/db/catalog/index_builds_manager.h b/src/mongo/db/catalog/index_builds_manager.h
index cddb46c3d16..37614751eb5 100644
--- a/src/mongo/db/catalog/index_builds_manager.h
+++ b/src/mongo/db/catalog/index_builds_manager.h
@@ -73,6 +73,7 @@ public:
         IndexConstraints indexConstraints = IndexConstraints::kEnforce;
         IndexBuildProtocol protocol = IndexBuildProtocol::kSinglePhase;
         IndexBuildMethod method = IndexBuildMethod::kHybrid;
+        bool forRecovery = false;
     };
 
     IndexBuildsManager() = default;
diff --git a/src/mongo/db/catalog/index_catalog.h b/src/mongo/db/catalog/index_catalog.h
index 58e48e08be5..6f4fa0c8ae3 100644
--- a/src/mongo/db/catalog/index_catalog.h
+++ b/src/mongo/db/catalog/index_catalog.h
@@ -419,6 +419,16 @@ public:
                              const IndexDescriptor* desc) = 0;
 
     /**
+     * Resets the index given its descriptor.
+     *
+     * This can only be called during startup recovery as it involves recreating the index table to
+     * allow bulk cursors to be used again.
+     */
+    virtual Status resetUnfinishedIndexForRecovery(OperationContext* opCtx,
+                                                   Collection* collection,
+                                                   const IndexDescriptor* desc) = 0;
+
+    /**
      * Drops an unfinished index given its descriptor.
      *
      * The caller must hold the collection X lock.
diff --git a/src/mongo/db/catalog/index_catalog_entry.h b/src/mongo/db/catalog/index_catalog_entry.h
index 9761f590c9d..2cf80bb8d3f 100644
--- a/src/mongo/db/catalog/index_catalog_entry.h
+++ b/src/mongo/db/catalog/index_catalog_entry.h
@@ -95,6 +95,7 @@ public:
     /// ---------------------
 
     virtual void setIsReady(bool newIsReady) = 0;
+    virtual void setIsFrozen(bool newIsFrozen) = 0;
 
     virtual void setDropped() = 0;
     virtual bool isDropped() const = 0;
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index 498165df637..e104623ef2d 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -179,6 +179,10 @@ void IndexCatalogEntryImpl::setIsReady(bool newIsReady) {
     _isReady = newIsReady;
 }
 
+void IndexCatalogEntryImpl::setIsFrozen(bool newIsFrozen) {
+    _isFrozen = newIsFrozen;
+}
+
 void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
                                         const CollectionPtr& collection,
                                         const KeyStringSet& multikeyMetadataKeys,
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.h b/src/mongo/db/catalog/index_catalog_entry_impl.h
index 30ef5a80921..0760989b99a 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -110,6 +110,8 @@ public:
 
     void setIsReady(bool newIsReady) final;
 
+    void setIsFrozen(bool newIsFrozen) final;
+
     void setDropped() final {
         _isDropped.store(true);
     }
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 4a4d0b64826..2c3086a54cf 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -623,7 +623,7 @@ StatusWith<BSONObj> IndexCatalogImpl::createIndexOnEmptyCollection(OperationCont
     boost::optional<UUID> buildUUID = boost::none;
     IndexBuildBlock indexBuildBlock(
         collection->ns(), spec, IndexBuildMethod::kForeground, buildUUID);
-    status = indexBuildBlock.init(opCtx, collection);
+    status = indexBuildBlock.init(opCtx, collection, /*forRecovery=*/false);
     if (!status.isOK())
         return status;
 
@@ -1230,6 +1230,73 @@ Status IndexCatalogImpl::dropIndex(OperationContext* opCtx,
     return dropIndexEntry(opCtx, collection, entry);
 }
 
+Status IndexCatalogImpl::resetUnfinishedIndexForRecovery(OperationContext* opCtx,
+                                                         Collection* collection,
+                                                         const IndexDescriptor* desc) {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
+    invariant(opCtx->lockState()->inAWriteUnitOfWork());
+
+    IndexCatalogEntry* entry = desc->getEntry();
+    const std::string indexName = entry->descriptor()->indexName();
+
+    // Only indexes that aren't ready can be reset.
+    invariant(!collection->isIndexReady(indexName));
+
+    auto released = [&] {
+        if (auto released = _readyIndexes.release(entry->descriptor())) {
+            invariant(!released, "Cannot reset a ready index");
+        }
+        if (auto released = _buildingIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        if (auto released = _frozenIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        MONGO_UNREACHABLE;
+    }();
+
+    LOGV2(6987700,
+          "Resetting unfinished index",
+          logAttrs(collection->ns()),
+          "index"_attr = indexName,
+          "ident"_attr = released->getIdent());
+
+    invariant(released.get() == entry);
+
+    // Drop the ident if it exists. The storage engine will return OK if the ident is not found.
+    auto engine = opCtx->getServiceContext()->getStorageEngine();
+    const std::string ident = released->getIdent();
+    Status status = engine->getEngine()->dropIdent(opCtx->recoveryUnit(), ident);
+    if (!status.isOK()) {
+        return status;
+    }
+
+    // Recreate the ident on-disk. DurableCatalog::createIndex() will lookup the ident internally
+    // using the catalogId and index name.
+    status = DurableCatalog::get(opCtx)->createIndex(opCtx,
+                                                     collection->getCatalogId(),
+                                                     collection->ns(),
+                                                     collection->getCollectionOptions(),
+                                                     released->descriptor());
+    if (!status.isOK()) {
+        return status;
+    }
+
+    // Update the index entry state in preparation to rebuild the index.
+    if (!released->accessMethod()) {
+        std::unique_ptr<SortedDataInterface> sdi = engine->getEngine()->getSortedDataInterface(
+            opCtx, collection->ns(), collection->getCollectionOptions(), ident, desc);
+        std::unique_ptr<IndexAccessMethod> accessMethod =
+            IndexAccessMethod::make(released.get(), std::move(sdi));
+        released->setAccessMethod(std::move(accessMethod));
+    }
+
+    released->setIsFrozen(false);
+    _buildingIndexes.add(std::move(released));
+
+    return Status::OK();
+}
+
 Status IndexCatalogImpl::dropUnfinishedIndex(OperationContext* opCtx,
                                              Collection* collection,
                                              const IndexDescriptor* desc) {
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index f9baf272f11..223399e5bf0 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -200,6 +200,9 @@ public:
     Status dropIndex(OperationContext* opCtx,
                      Collection* collection,
                      const IndexDescriptor* desc) override;
+    Status resetUnfinishedIndexForRecovery(OperationContext* opCtx,
+                                           Collection* collection,
+                                           const IndexDescriptor* desc) override;
     Status dropUnfinishedIndex(OperationContext* opCtx,
                                Collection* collection,
                                const IndexDescriptor* desc) override;
diff --git a/src/mongo/db/catalog/index_key_validate.cpp b/src/mongo/db/catalog/index_key_validate.cpp
index 8433b958894..20dd9f37fc0 100644
--- a/src/mongo/db/catalog/index_key_validate.cpp
+++ b/src/mongo/db/catalog/index_key_validate.cpp
@@ -497,6 +497,9 @@ StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& in
                             << "' is only allowed when '" << IndexDescriptor::kKeyPatternFieldName
                             << "' is {\"$**\": ±1}"};
             }
+            if (key.nFields() != 1) {
+                return {ErrorCodes::CannotCreateIndex, "wildcard indexes do not allow compounding"};
+            }
 
             if (indexSpecElem.embeddedObject().isEmpty()) {
                 return {ErrorCodes::FailedToParse,
diff --git a/src/mongo/db/catalog/multi_index_block.cpp b/src/mongo/db/catalog/multi_index_block.cpp
index 10ac2a47fdd..a0e8382959e 100644
--- a/src/mongo/db/catalog/multi_index_block.cpp
+++ b/src/mongo/db/catalog/multi_index_block.cpp
@@ -185,7 +185,7 @@ StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(OperationContext* opCtx,
                                                        const BSONObj& spec,
                                                        OnInitFn onInit) {
     const auto indexes = std::vector<BSONObj>(1, spec);
-    return init(opCtx, collection, indexes, onInit, boost::none);
+    return init(opCtx, collection, indexes, onInit, /*forRecovery=*/false, boost::none);
 }
 
 StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(
@@ -193,6 +193,7 @@ StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(
     CollectionWriter& collection,
     const std::vector<BSONObj>& indexSpecs,
     OnInitFn onInit,
+    bool forRecovery,
     const boost::optional<ResumeIndexInfo>& resumeInfo) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X),
               str::stream() << "Collection " << collection->ns() << " with UUID "
@@ -246,27 +247,31 @@ StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(
 
         for (size_t i = 0; i < indexSpecs.size(); i++) {
             BSONObj info = indexSpecs[i];
-            StatusWith<BSONObj> statusWithInfo =
-                collection->getIndexCatalog()->prepareSpecForCreate(
-                    opCtx, collection.get(), info, resumeInfo);
-            Status status = statusWithInfo.getStatus();
-            if (!status.isOK()) {
-                // If we were given two identical indexes to build, we will run into an error trying
-                // to set up the same index a second time in this for-loop. This is the only way to
-                // encounter this error because callers filter out ready/in-progress indexes and
-                // start the build while holding a lock throughout.
-                if (status == ErrorCodes::IndexBuildAlreadyInProgress) {
-                    invariant(indexSpecs.size() > 1,
-                              str::stream()
-                                  << "Collection: " << collection->ns() << " (" << _collectionUUID
-                                  << "), Index spec: " << indexSpecs.front());
-                    return {
-                        ErrorCodes::OperationFailed,
-                        "Cannot build two identical indexes. Try again without duplicate indexes."};
+            if (!forRecovery) {
+                // We skip this step when initializing unfinished index builds during startup
+                // recovery as they are already in the index catalog.
+                StatusWith<BSONObj> statusWithInfo =
+                    collection->getIndexCatalog()->prepareSpecForCreate(
+                        opCtx, collection.get(), info, resumeInfo);
+                Status status = statusWithInfo.getStatus();
+                if (!status.isOK()) {
+                    // If we were given two identical indexes to build, we will run into an error
+                    // trying to set up the same index a second time in this for-loop. This is the
+                    // only way to encounter this error because callers filter out ready/in-progress
+                    // indexes and start the build while holding a lock throughout.
+                    if (status == ErrorCodes::IndexBuildAlreadyInProgress) {
+                        invariant(indexSpecs.size() > 1,
+                                  str::stream() << "Collection: " << collection->ns() << " ("
+                                                << _collectionUUID
+                                                << "), Index spec: " << indexSpecs.front());
+                        return {ErrorCodes::OperationFailed,
+                                "Cannot build two identical indexes. Try again without duplicate "
+                                "indexes."};
+                    }
+                    return status;
                 }
-                return status;
+                info = statusWithInfo.getValue();
             }
-            info = statusWithInfo.getValue();
             indexInfoObjs.push_back(info);
 
             boost::optional<TimeseriesOptions> options = collection->getTimeseriesOptions();
@@ -302,7 +307,7 @@ StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(
                 status = index.block->initForResume(
                     opCtx, collection.getWritableCollection(), *stateInfo, resumeInfo->getPhase());
             } else {
-                status = index.block->init(opCtx, collection.getWritableCollection());
+                status = index.block->init(opCtx, collection.getWritableCollection(), forRecovery);
             }
             if (!status.isOK())
                 return status;
@@ -685,10 +690,10 @@ Status MultiIndexBlock::_insert(OperationContext* opCtx,
     // collection to have it.
     if (_containsIndexBuildOnTimeseriesMeasurement &&
         *collection->getTimeseriesBucketsMayHaveMixedSchemaData()) {
-        bool docHasMixedSchemaData =
+        auto docHasMixedSchemaData =
             collection->doesTimeseriesBucketsDocContainMixedSchemaData(doc);
 
-        if (docHasMixedSchemaData) {
+        if (docHasMixedSchemaData.isOK() && docHasMixedSchemaData.getValue()) {
             LOGV2(6057700,
                   "Detected mixed-schema data in time-series bucket collection",
                   logAttrs(collection->ns()),
@@ -704,7 +709,8 @@ Status MultiIndexBlock::_insert(OperationContext* opCtx,
         auto replCoord = repl::ReplicationCoordinator::get(opCtx);
         const bool replSetAndNotPrimary = !replCoord->canAcceptWritesFor(opCtx, collection->ns());
 
-        if (docHasMixedSchemaData && !replSetAndNotPrimary) {
+        if (docHasMixedSchemaData.isOK() && docHasMixedSchemaData.getValue() &&
+            !replSetAndNotPrimary) {
             return timeseriesMixedSchemaDataFailure(collection.get());
         }
     }
@@ -972,15 +978,21 @@ Status MultiIndexBlock::commit(OperationContext* opCtx,
 
     onCommit();
 
-    // Update the 'timeseriesBucketsMayHaveMixedSchemaData' catalog entry flag to false in order to
-    // allow subsequent index builds to skip checking bucket documents for mixed-schema data.
+    // We can't update the 'timeseriesBucketsMayHaveMixedSchemaData' catalog entry flag here as it
+    // requires the change to be driven by the router role. It means that subsequent index builds
+    // and other systems needs to treat this collection as-if it contains mixed-schema data even if
+    // it might not. We log a warning that can be used to initiate changing the flag. Note: just
+    // because this node doesn't contain mixed-schema it doesn't mean that other shards can't have
+    // mixed schema data. This flag needs to be consistent across the shards.
     if (_containsIndexBuildOnTimeseriesMeasurement && !_timeseriesBucketContainsMixedSchemaData) {
         boost::optional<bool> mayContainMixedSchemaData =
             collection->getTimeseriesBucketsMayHaveMixedSchemaData();
         invariant(mayContainMixedSchemaData);
 
         if (*mayContainMixedSchemaData) {
-            collection->setTimeseriesBucketsMayHaveMixedSchemaData(opCtx, false);
+            LOGV2_WARNING(9301400,
+                          "Index build finished for time-series collection marked as containing "
+                          "mixed schema buckets without detecting any buckets with mixed schema.");
         }
     }
 
diff --git a/src/mongo/db/catalog/multi_index_block.h b/src/mongo/db/catalog/multi_index_block.h
index 840770595cf..5220f8dd6f0 100644
--- a/src/mongo/db/catalog/multi_index_block.h
+++ b/src/mongo/db/catalog/multi_index_block.h
@@ -114,6 +114,7 @@ public:
         CollectionWriter& collection,
         const std::vector<BSONObj>& specs,
         OnInitFn onInit,
+        bool forRecovery,
         const boost::optional<ResumeIndexInfo>& resumeInfo = boost::none);
     StatusWith<std::vector<BSONObj>> init(OperationContext* opCtx,
                                           CollectionWriter& collection,
diff --git a/src/mongo/db/catalog/multi_index_block_test.cpp b/src/mongo/db/catalog/multi_index_block_test.cpp
index 2fb9caf7371..e9153ba35dc 100644
--- a/src/mongo/db/catalog/multi_index_block_test.cpp
+++ b/src/mongo/db/catalog/multi_index_block_test.cpp
@@ -90,8 +90,11 @@ TEST_F(MultiIndexBlockTest, CommitWithoutInsertingDocuments) {
     AutoGetCollection autoColl(operationContext(), getNSS(), MODE_X);
     CollectionWriter coll(operationContext(), autoColl);
 
-    auto specs = unittest::assertGet(indexer->init(
-        operationContext(), coll, std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
+    auto specs = unittest::assertGet(indexer->init(operationContext(),
+                                                   coll,
+                                                   std::vector<BSONObj>(),
+                                                   MultiIndexBlock::kNoopOnInitFn,
+                                                   /*forRecovery=*/false));
     ASSERT_EQUALS(0U, specs.size());
 
     ASSERT_OK(indexer->dumpInsertsFromBulk(operationContext(), coll.get()));
@@ -113,8 +116,11 @@ TEST_F(MultiIndexBlockTest, CommitAfterInsertingSingleDocument) {
     AutoGetCollection autoColl(operationContext(), getNSS(), MODE_X);
     CollectionWriter coll(operationContext(), autoColl);
 
-    auto specs = unittest::assertGet(indexer->init(
-        operationContext(), coll, std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
+    auto specs = unittest::assertGet(indexer->init(operationContext(),
+                                                   coll,
+                                                   std::vector<BSONObj>(),
+                                                   MultiIndexBlock::kNoopOnInitFn,
+                                                   /*forRecovery=*/false));
     ASSERT_EQUALS(0U, specs.size());
 
     ASSERT_OK(
@@ -146,8 +152,11 @@ TEST_F(MultiIndexBlockTest, AbortWithoutCleanupAfterInsertingSingleDocument) {
     AutoGetCollection autoColl(operationContext(), getNSS(), MODE_X);
     CollectionWriter coll(operationContext(), autoColl);
 
-    auto specs = unittest::assertGet(indexer->init(
-        operationContext(), coll, std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
+    auto specs = unittest::assertGet(indexer->init(operationContext(),
+                                                   coll,
+                                                   std::vector<BSONObj>(),
+                                                   MultiIndexBlock::kNoopOnInitFn,
+                                                   /*forRecovery=*/false));
     ASSERT_EQUALS(0U, specs.size());
     ASSERT_OK(
         indexer->insertSingleDocumentForInitialSyncOrRecovery(operationContext(),
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index ee4b9477100..01bc87f7e2b 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -885,6 +885,10 @@ void validateNamespacesForRenameCollection(OperationContext* opCtx,
             "renaming system.views collection or renaming to system.views is not allowed",
             !source.isSystemDotViews() && !target.isSystemDotViews());
 
+    uassert(ErrorCodes::IllegalOperation,
+            "renaming system.users collection or renaming to system.users is not allowed",
+            !source.isSystemDotUsers() && !target.isSystemDotUsers());
+
     if (source.isTimeseriesBucketsCollection()) {
         uassert(ErrorCodes::IllegalOperation,
                 "Renaming system.buckets collections is not allowed",
diff --git a/src/mongo/db/catalog/storage_engine_collection_options_flags_parser.cpp b/src/mongo/db/catalog/storage_engine_collection_options_flags_parser.cpp
new file mode 100644
index 00000000000..66ed55b6e57
--- /dev/null
+++ b/src/mongo/db/catalog/storage_engine_collection_options_flags_parser.cpp
@@ -0,0 +1,161 @@
+/**
+ *    Copyright (C) 2024-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include <algorithm>
+#include <cstring>
+#include <fmt/format.h>
+#include <pcrecpp.h>
+
+#include "mongo/base/string_data.h"
+#include "mongo/db/catalog/storage_engine_collection_options_flags_parser.h"
+#include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
+#include "mongo/util/ctype.h"
+
+namespace mongo {
+
+const static StaticImmortal<pcrecpp::RE> appMetadataRegex(
+    R"re(((?<=^|,)\s*(?:app_metadata|\"app_metadata\")\s*[=:]\s*[({[]\s*))re");
+
+static pcrecpp::RE flagMatchRegex(StringData flagName) {
+    // This check is overly strict, but it suffices for now and ensures that both:
+    // - The flag name is a valid WiredTiger identifier, and
+    // - It can be used in the regular expression without needing to escape it
+    invariant(std::all_of(flagName.begin(), flagName.end(), ctype::isAlpha));
+
+    // Some examples of possible matches:
+    // `flag=false`
+    // `flag:true`
+    // `flag`
+    // `     "flag"  =   false     `
+    // `     "flag"     `
+    return pcrecpp::RE(fmt::format(
+        R"re(((?<=[,({{[])\s*(?:{0}|\"{0}\")(?:\s*[=:]\s*(true|false))?\s*(?=[,)}}\]])))re",
+        flagName));
+}
+
+static std::map<StringData, boost::optional<bool>> getFlagsFromWtConfigStringAppMetadata(
+    const std::string& configString, const std::vector<StringData>& flagNames) {
+    std::map<StringData, boost::optional<bool>> flags;
+
+    for (const auto& flagName : flagNames) {
+        auto flagRegex = flagMatchRegex(flagName);
+        pcrecpp::StringPiece fullMatch, flagValueStr;
+        auto matches = flagRegex.PartialMatch(configString, &fullMatch, &flagValueStr);
+        flags.emplace(flagName,
+                      matches ? boost::optional<bool>(flagValueStr == "" || flagValueStr == "true")
+                              : boost::none);
+    }
+
+    return flags;
+}
+
+std::map<StringData, boost::optional<bool>> getFlagsFromStorageEngineBson(
+    const BSONObj& storageEngineOptions, const std::vector<StringData>& flagNames) {
+    auto configString = WiredTigerUtil::getConfigStringFromStorageOptions(storageEngineOptions);
+    return getFlagsFromWtConfigStringAppMetadata(configString.value_or(""), flagNames);
+}
+
+boost::optional<bool> getFlagFromStorageEngineBson(const BSONObj& storageEngineOptions,
+                                                   StringData flagName) {
+    return getFlagsFromStorageEngineBson(storageEngineOptions, {flagName})[flagName];
+}
+
+// Finds or adds the 'app_metadata=(...)' struct inside a WiredTiger config string
+// Returns the position inside the struct (after the delimiter, before the first key-value)
+static size_t findOrAddAppMetadataStructToConfigString(std::string& configString) {
+    pcrecpp::StringPiece fullMatch;
+    auto matches = appMetadataRegex->PartialMatch(configString, &fullMatch);
+    if (!matches)
+        configString += configString.empty() ? "app_metadata=()" : ",app_metadata=()";
+    return matches ? (fullMatch.data() + fullMatch.size() - configString.data())
+                   : configString.size() - 1;
+}
+
+// Expand a [pos, len) range inside a config string to include a leading or trailing comma separator
+static void expandRangeToIncludeSeparator(const std::string& configString,
+                                          size_t& pos,
+                                          size_t& len) {
+    if (pos > 0 && configString[pos - 1] == ',') {
+        pos--;
+        len++;
+    } else if (pos + len < configString.size() && configString[pos + len] == ',') {
+        len++;
+    }
+}
+
+static void setFlagsToWtConfigStringAppMetadata(
+    std::string& configString, const std::map<StringData, boost::optional<bool>>& flags) {
+    auto metadataPos = findOrAddAppMetadataStructToConfigString(configString);
+
+    for (const auto& [flagName, flagValue] : flags) {
+        auto flagRegex = flagMatchRegex(flagName);
+        pcrecpp::StringPiece fullMatch;
+        // "- 1" allows the positive lookbehind (?<=) at the start of the regex to work
+        auto matches = flagRegex.PartialMatch(&configString[metadataPos - 1], &fullMatch);
+        if (matches) {
+            size_t pos = fullMatch.data() - configString.data(), len = fullMatch.size();
+
+            if (flagValue.has_value()) {  // Replace existing flag
+                auto flagItem = fmt::format("{}={}", flagName, *flagValue);
+                configString.replace(pos, len, flagItem);
+            } else {  // Unset existing flag
+                expandRangeToIncludeSeparator(configString, pos, len);
+                configString.erase(pos, len);
+            }
+        } else if (flagValue.has_value()) {  // Add new flag
+            auto metadataEmpty = strchr(")]}", configString[metadataPos]) != nullptr;
+            auto flagItem = fmt::format("{}={}{}", flagName, *flagValue, metadataEmpty ? "" : ",");
+            configString.insert(metadataPos, flagItem);
+        }
+    }
+}
+
+BSONObj setFlagsToStorageEngineBson(const BSONObj& storageEngineOptions,
+                                    const std::map<StringData, boost::optional<bool>>& flags) {
+    auto configString =
+        WiredTigerUtil::getConfigStringFromStorageOptions(storageEngineOptions).value_or("");
+    setFlagsToWtConfigStringAppMetadata(configString, flags);
+
+    // Both for safety, and because the regex-based parser can not handle some theoretical cases,
+    // sanity check that the resulting string is a valid WiredTiger configuration string
+    auto configStringObj = BSON(WiredTigerUtil::kConfigStringField << configString);
+    tassert(9218600,
+            "The resulting WiredTiger configuration string is not valid",
+            WiredTigerUtil::checkTableCreationOptions(configStringObj.firstElement()).isOK());
+
+    return WiredTigerUtil::setConfigStringToStorageOptions(storageEngineOptions, configString);
+}
+
+BSONObj setFlagToStorageEngineBson(const BSONObj& storageEngineOptions,
+                                   StringData flagName,
+                                   boost::optional<bool> flagValue) {
+    return setFlagsToStorageEngineBson(storageEngineOptions, {{flagName, flagValue}});
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/storage_engine_collection_options_flags_parser.h b/src/mongo/db/catalog/storage_engine_collection_options_flags_parser.h
new file mode 100644
index 00000000000..cb4a1b20e2f
--- /dev/null
+++ b/src/mongo/db/catalog/storage_engine_collection_options_flags_parser.h
@@ -0,0 +1,62 @@
+/**
+ *    Copyright (C) 2024-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include <boost/optional/optional.hpp>
+#include <map>
+#include <vector>
+
+#include "mongo/base/string_data.h"
+#include "mongo/bson/bsonobj.h"
+
+namespace mongo {
+
+/**
+ * Utility functions to get or set boolean flags from/to a storage engine options object
+ * (see `CollectionOptions::storageEngine`).
+ *
+ * The idea is that for exceptional (workaround) purposes, we can use the storage engine
+ * options object as a flexible structure where new fields can be added retroactively,
+ * unlike the other parts of the catalog which generally have non-flexible / strict validations.
+ * For more information, see: SERVER-91195, SERVER-92186.
+ */
+
+std::map<StringData, boost::optional<bool>> getFlagsFromStorageEngineBson(
+    const BSONObj& storageEngineOptions, const std::vector<StringData>& flagNames);
+
+boost::optional<bool> getFlagFromStorageEngineBson(const BSONObj& storageEngineOptions,
+                                                   StringData flagName);
+
+[[nodiscard]] BSONObj setFlagsToStorageEngineBson(
+    const BSONObj& storageEngineOptions, const std::map<StringData, boost::optional<bool>>& flags);
+
+[[nodiscard]] BSONObj setFlagToStorageEngineBson(const BSONObj& storageEngineOptions,
+                                                 StringData flagName,
+                                                 boost::optional<bool> flagValue);
+
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/storage_engine_collection_options_flags_parser_test.cpp b/src/mongo/db/catalog/storage_engine_collection_options_flags_parser_test.cpp
new file mode 100644
index 00000000000..5305668e8fc
--- /dev/null
+++ b/src/mongo/db/catalog/storage_engine_collection_options_flags_parser_test.cpp
@@ -0,0 +1,250 @@
+/**
+ *    Copyright (C) 2024-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/catalog/storage_engine_collection_options_flags_parser.h"
+#include "mongo/db/storage/wiredtiger/wiredtiger_record_store.h"
+#include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
+#include "mongo/unittest/bson_test_util.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+static BSONObj makeStorageEngineWithConfigString(StringData configString) {
+    return BSON(kWiredTigerEngineName << BSON(WiredTigerUtil::kConfigStringField << configString));
+}
+
+static BSONObj addExtraFields(const BSONObj& storageEngineOptions) {
+    auto wtObj = storageEngineOptions[kWiredTigerEngineName].Obj();
+    return storageEngineOptions.addFields(BSON("dummy1"
+                                               << "value1" << kWiredTigerEngineName
+                                               << wtObj.addFields(BSON("dummy2"
+                                                                       << "value2"))));
+}
+
+TEST(StorageEngineFlagsParserTest, GetEmptyOptionalWhenNoWiredTigerConfigString) {
+    auto options = BSONObj();
+
+    auto flag = getFlagFromStorageEngineBson(options, "flagA");
+
+    ASSERT_EQ(boost::none, flag);
+}
+
+TEST(StorageEngineFlagsParserTest, GetEmptyOptionalWhenOptionsDoesNotContainMetadata) {
+    auto options = makeStorageEngineWithConfigString("access_pattern_hint=random");
+
+    auto flag = getFlagFromStorageEngineBson(options, "flagA");
+
+    ASSERT_EQ(boost::none, flag);
+}
+
+TEST(StorageEngineFlagsParserTest, GetEmptyOptionalWhenMetadataDoesNotContainTheFlag) {
+    auto options = makeStorageEngineWithConfigString("app_metadata=(formatVersion=1)");
+
+    auto flags = getFlagsFromStorageEngineBson(options, {"flagA", "flagB"});
+
+    ASSERT_EQ(boost::none, flags["flagA"]);
+    ASSERT_EQ(boost::none, flags["flagB"]);
+}
+
+TEST(StorageEngineFlagsParserTest, GetValueWhenMetadataContainsASingleFlag) {
+    auto options = makeStorageEngineWithConfigString("app_metadata=(formatVersion=1,flagA=true)");
+
+    auto flags = getFlagsFromStorageEngineBson(options, {"flagA", "flagB"});
+
+    ASSERT_EQ(true, flags["flagA"]);
+    ASSERT_EQ(boost::none, flags["flagB"]);
+}
+
+TEST(StorageEngineFlagsParserTest, GetValueWhenMetadataContainsMultipleFlags) {
+    auto options = makeStorageEngineWithConfigString("app_metadata=(flagB=true,flagA=false)");
+
+    auto flags = getFlagsFromStorageEngineBson(options, {"flagA", "flagB"});
+
+    ASSERT_EQ(false, flags["flagA"]);
+    ASSERT_EQ(true, flags["flagB"]);
+}
+
+TEST(StorageEngineFlagsParserTest, GetEmptyOptionalWhenMetadataContainsAnInvalidValue) {
+    auto options =
+        makeStorageEngineWithConfigString("app_metadata=(flagB=(hello=world),flagA=true)");
+
+    auto flags = getFlagsFromStorageEngineBson(options, {"flagA", "flagB"});
+
+    ASSERT_EQ(true, flags["flagA"]);
+    ASSERT_EQ(boost::none, flags["flagB"]);
+}
+
+TEST(StorageEngineFlagsParserTest, GetTrueWhenMetadataContainsAKeyWithNoValue) {
+    auto options = makeStorageEngineWithConfigString("app_metadata=(formatVersion=1,flagA)");
+
+    auto flag = getFlagFromStorageEngineBson(options, "flagA");
+
+    ASSERT_EQ(true, flag);
+}
+
+TEST(StorageEngineFlagsParserTest, GetIgnoresUnknownStorageEngineFields) {
+    auto options = addExtraFields(makeStorageEngineWithConfigString("app_metadata=(flagA=true)"));
+
+    auto flag = getFlagFromStorageEngineBson(options, "flagA");
+
+    ASSERT_EQ(true, flag);
+}
+
+TEST(StorageEngineFlagsParserTest, GetHandlesTrickyFormatting) {
+    auto options = addExtraFields(makeStorageEngineWithConfigString(
+        "   access_pattern_hint   =    random   ,    \"app_metadata\"    :   [    x=y   ,    "
+        "\"flagB\":   true   ,   z :   t  ]"));
+
+    auto flags = getFlagsFromStorageEngineBson(options, {"flagA", "flagB"});
+
+    ASSERT_EQ(boost::none, flags["flagA"]);
+    ASSERT_EQ(true, flags["flagB"]);
+}
+
+TEST(StorageEngineFlagsParserTest, AddFlagToEmptyStorageEngineBson) {
+    auto options = BSONObj();
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagA", true);
+
+    ASSERT_BSONOBJ_EQ(newOptions, makeStorageEngineWithConfigString("app_metadata=(flagA=true)"));
+}
+
+TEST(StorageEngineFlagsParserTest, AddFlagToEmptyConfigString) {
+    auto options = makeStorageEngineWithConfigString("");
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagA", true);
+
+    ASSERT_BSONOBJ_EQ(newOptions, makeStorageEngineWithConfigString("app_metadata=(flagA=true)"));
+}
+
+TEST(StorageEngineFlagsParserTest, AddFlagToExistingConfigStringWithNoMetadata) {
+    auto options = makeStorageEngineWithConfigString("access_pattern_hint=random");
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagA", true);
+
+    ASSERT_BSONOBJ_EQ(
+        newOptions,
+        makeStorageEngineWithConfigString("access_pattern_hint=random,app_metadata=(flagA=true)"));
+}
+
+TEST(StorageEngineFlagsParserTest, AddFlagToExistingConfigStringWithEmptyMetadata) {
+    auto options = makeStorageEngineWithConfigString("access_pattern_hint=random,app_metadata=()");
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagA", false);
+
+    ASSERT_BSONOBJ_EQ(
+        newOptions,
+        makeStorageEngineWithConfigString("access_pattern_hint=random,app_metadata=(flagA=false)"));
+}
+
+TEST(StorageEngineFlagsParserTest, AddFlagToExistingConfigStringWithOtherFlags) {
+    auto options = makeStorageEngineWithConfigString(
+        "access_pattern_hint=random,app_metadata=(hello2=world2,flagB=true)");
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagA", false);
+
+    ASSERT_BSONOBJ_EQ(
+        newOptions,
+        makeStorageEngineWithConfigString(
+            "access_pattern_hint=random,app_metadata=(flagA=false,hello2=world2,flagB=true)"));
+}
+
+TEST(StorageEngineFlagsParserTest, SetExistingFlag) {
+    auto options =
+        makeStorageEngineWithConfigString("access_pattern_hint=random,app_metadata=(flagA=false)");
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagA", true);
+
+    ASSERT_BSONOBJ_EQ(
+        newOptions,
+        makeStorageEngineWithConfigString("access_pattern_hint=random,app_metadata=(flagA=true)"));
+}
+
+TEST(StorageEngineFlagsParserTest, RemoveExistingFlag) {
+    auto options = makeStorageEngineWithConfigString(
+        "access_pattern_hint=random,app_metadata=(flagB=true,x=y,z=t)");
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagB", boost::none);
+
+    ASSERT_BSONOBJ_EQ(
+        newOptions,
+        makeStorageEngineWithConfigString("access_pattern_hint=random,app_metadata=(x=y,z=t)"));
+}
+
+TEST(StorageEngineFlagsParserTest, SetMultipleFlags) {
+    auto options = makeStorageEngineWithConfigString(
+        "access_pattern_hint=random,app_metadata=(x=y,flagB=true,z=t,flagC=true)");
+
+    auto newOptions = setFlagsToStorageEngineBson(
+        options, {{"flagB", false}, {"flagA", true}, {"flagC", boost::none}});
+
+    ASSERT_BSONOBJ_EQ(
+        newOptions,
+        makeStorageEngineWithConfigString(
+            "access_pattern_hint=random,app_metadata=(flagA=true,x=y,flagB=false,z=t)"));
+}
+
+TEST(StorageEngineFlagsParserTest, SetFlagWhenMetadataContainsAKeyWithNoValue) {
+    auto options = makeStorageEngineWithConfigString("app_metadata=(formatVersion=1,flagA)");
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagA", false);
+
+    ASSERT_BSONOBJ_EQ(
+        newOptions,
+        makeStorageEngineWithConfigString("app_metadata=(formatVersion=1,flagA=false)"));
+}
+
+TEST(StorageEngineFlagsParserTest, SetPreservesUnknownStorageEngineFields) {
+    auto options = addExtraFields(makeStorageEngineWithConfigString(
+        "access_pattern_hint=random,app_metadata=(x=y,flagB=false,z=t)"));
+
+    auto newOptions = setFlagToStorageEngineBson(options, "flagB", true);
+
+    auto expected = addExtraFields(makeStorageEngineWithConfigString(
+        "access_pattern_hint=random,app_metadata=(x=y,flagB=true,z=t)"));
+    ASSERT_BSONOBJ_EQ(newOptions, expected);
+}
+
+TEST(StorageEngineFlagsParserTest, SetHandlesTrickyFormatting) {
+    auto options = addExtraFields(makeStorageEngineWithConfigString(
+        "   access_pattern_hint   =    random   ,    \"app_metadata\"    :   [    x=y   ,    "
+        "\"flagB\":   false   ,   z :   t   , flagC    :  true ]"));
+
+    auto newOptions = setFlagsToStorageEngineBson(
+        options, {{"flagA", false}, {"flagB", true}, {"flagC", boost::none}});
+
+    auto expected = addExtraFields(makeStorageEngineWithConfigString(
+        "   access_pattern_hint   =    random   ,    \"app_metadata\"    :   [    flagA=false,x=y  "
+        " ,flagB=true,   z :   t   ]"));
+    ASSERT_BSONOBJ_EQ(newOptions, expected);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/validate_adaptor.cpp b/src/mongo/db/catalog/validate_adaptor.cpp
index 3967bd93d82..9dc4a8dc71b 100644
--- a/src/mongo/db/catalog/validate_adaptor.cpp
+++ b/src/mongo/db/catalog/validate_adaptor.cpp
@@ -53,6 +53,10 @@
 #include "mongo/db/storage/execution_context.h"
 #include "mongo/db/storage/key_string.h"
 #include "mongo/db/storage/record_store.h"
+#include "mongo/db/storage/storage_parameters_gen.h"
+#include "mongo/db/timeseries/flat_bson.h"
+#include "mongo/db/timeseries/timeseries_constants.h"
+#include "mongo/db/timeseries/timeseries_options.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/object_check.h"
 #include "mongo/util/fail_point.h"
@@ -64,6 +68,7 @@ namespace {
 
 MONGO_FAIL_POINT_DEFINE(crashOnMultikeyValidateFailure);
 MONGO_FAIL_POINT_DEFINE(failIndexKeyOrdering);
+MONGO_FAIL_POINT_DEFINE(failRecordStoreTraversal);
 
 // Set limit for size of corrupted records that will be reported.
 const long long kMaxErrorSizeBytes = 1 * 1024 * 1024;
@@ -73,6 +78,9 @@ const long long kInterruptIntervalNumBytes = 50 * 1024 * 1024;  // 50MB.
 static constexpr const char* kSchemaValidationFailedReason =
     "Detected one or more documents not compliant with the collection's schema. Check logs for log "
     "id 5363500.";
+static constexpr const char* kTimeseriesValidationInconsistencyReason =
+    "Detected one or more documents in this collection incompatible with time-series "
+    "specifications. For more info, see logs with log id 6698300.";
 
 /**
  * Validate that for each record in a clustered RecordStore the record key (RecordId) matches the
@@ -119,17 +127,163 @@ void schemaValidationFailed(CollectionValidation::ValidateState* state,
 
     state->setCollectionSchemaViolated();
 
-    // TODO SERVER-65078: remove the testing proctor check.
-    // When testing is enabled, only warn about non-compliant documents to prevent test failures.
-    if (TestingProctor::instance().isEnabled() ||
-        Collection::SchemaValidationResult::kWarn == result || state->warnOnSchemaValidation()) {
+    if (result != Collection::SchemaValidationResult::kPass) {
         results->warnings.push_back(kSchemaValidationFailedReason);
-    } else if (Collection::SchemaValidationResult::kError == result) {
-        results->errors.push_back(kSchemaValidationFailedReason);
-        results->valid = false;
     }
 }
+/**
+ * Checks the value of the bucket's version and if it matches the types of 'data' fields.
+ */
+Status _validateTimeseriesControlVersion(const BSONObj& recordBson) {
+    int controlVersion = recordBson.getField(timeseries::kBucketControlFieldName)
+                             .Obj()
+                             .getField(timeseries::kBucketControlVersionFieldName)
+                             .Number();
+    if (controlVersion != 1 && controlVersion != 2) {
+        return Status(
+            ErrorCodes::BadValue,
+            fmt::format("Invalid value for 'control.version'. Expected 1 or 2, but got {}.",
+                        controlVersion));
+    }
+    auto dataType = controlVersion == 1 ? BSONType::Object : BSONType::BinData;
+    // In addition to checking dataType, make sure that closed buckets have BinData Column subtype
+    auto isCorrectType = [&](BSONElement el) {
+        if (controlVersion == 1) {
+            return el.type() == BSONType::Object;
+        } else {
+            return el.type() == BSONType::BinData && el.binDataType() == BinDataType::Column;
+        }
+    };
+    BSONObj data = recordBson.getField(timeseries::kBucketDataFieldName).Obj();
+    for (BSONObjIterator bi(data); bi.more();) {
+        BSONElement e = bi.next();
+        if (!isCorrectType(e)) {
+            return Status(ErrorCodes::TypeMismatch,
+                          fmt::format("Mismatch between time-series schema version and data field "
+                                      "type. Expected type {}, but got {}.",
+                                      mongo::typeName(dataType),
+                                      mongo::typeName(e.type())));
+        }
+    }
+    return Status::OK();
+}
+
+/**
+ * Checks the equivalence between the min and max fields in 'control' for a bucket and
+ * the corresponding value in 'data'.
+ */
+Status _validateTimeseriesMinMax(const BSONObj& recordBson, const CollectionPtr& coll) {
+    BSONObj data = recordBson.getField(timeseries::kBucketDataFieldName).Obj();
+    BSONObj control = recordBson.getField(timeseries::kBucketControlFieldName).Obj();
+    BSONObj controlMin = control.getField(timeseries::kBucketControlMinFieldName).Obj();
+    BSONObj controlMax = control.getField(timeseries::kBucketControlMaxFieldName).Obj();
+
+    auto dataFields = data.getFieldNames<std::set<std::string>>();
+    auto controlMinFields = controlMin.getFieldNames<std::set<std::string>>();
+    auto controlMaxFields = controlMax.getFieldNames<std::set<std::string>>();
+
+    // Checks that the number of 'control.min' and 'control.max' fields agrees with number of 'data'
+    // fields.
+    if (dataFields.size() != controlMinFields.size() ||
+        dataFields.size() != controlMaxFields.size()) {
+        return Status(
+            ErrorCodes::BadValue,
+            fmt::format(
+                "Mismatch between the number of time-series control fields and the number "
+                "of data fields. "
+                "Control had {} min fields and {} max fields, but observed data had {} fields.",
+                controlMinFields.size(),
+                controlMaxFields.size(),
+                dataFields.size()));
+    };
+
+    // Used when checking min timestamp, which is rounded down by granularity.
+    auto granularity = coll->getTimeseriesOptions()->getGranularity();
+
+    // Validates that the 'control.min' and 'control.max' field values agree with 'data' field
+    // values.
+    for (auto fieldName : dataFields) {
+        timeseries::MinMax minmax;
+        auto field = data.getField(fieldName);
+
+        for (BSONElement el : field.Obj()) {
+            minmax.update(el.wrap(fieldName), boost::none, coll->getDefaultCollator());
+        }
 
+        auto controlFieldMin = controlMin.getField(fieldName);
+        auto controlFieldMax = controlMax.getField(fieldName);
+        auto min = minmax.min();
+        auto max = minmax.max();
+
+        // Checks whether the min and max values between 'control' and 'data' match, taking
+        // timestamp granularity into account.
+        auto checkMinAndMaxMatch = [&]() {
+            if (fieldName == coll->getTimeseriesOptions()->getTimeField()) {
+                return controlFieldMin.Date() ==
+                    timeseries::roundTimestampToGranularity(min.getField(fieldName).Date(),
+                                                            granularity) &&
+                    controlFieldMax.Date() == max.getField(fieldName).Date();
+            } else {
+                return controlFieldMin.wrap().woCompare(min) == 0 &&
+                    controlFieldMax.wrap().woCompare(max) == 0;
+            }
+        };
+
+        if (!checkMinAndMaxMatch()) {
+            return Status(
+                ErrorCodes::BadValue,
+                fmt::format(
+                    "Mismatch between time-series control and observed min or max for field {}. "
+                    "Control had min {} and max {}, but observed data had min {} and max {}.",
+                    fieldName,
+                    controlFieldMin.toString(),
+                    controlFieldMax.toString(),
+                    min.toString(),
+                    max.toString()));
+        }
+    }
+
+    return Status::OK();
+}
+
+/**
+ * Validates the consistency of a time-series bucket.
+ */
+Status _validateTimeSeriesBucketRecord(const CollectionPtr& collection,
+                                       const BSONObj& recordBson,
+                                       ValidateResults* results) {
+
+    if (Status status = _validateTimeseriesControlVersion(recordBson); !status.isOK()) {
+        return status;
+    }
+
+    int version = recordBson.getField(timeseries::kBucketControlFieldName)
+                      .Obj()
+                      .getField(timeseries::kBucketControlVersionFieldName)
+                      .Number();
+
+    // TODO(SERVER-67023): Check closed bucket as part of validation.
+    if (version == 1) {
+        if (Status status = _validateTimeseriesMinMax(recordBson, collection); !status.isOK()) {
+            return status;
+        }
+    }
+
+
+    return Status::OK();
+}
+
+
+void _timeseriesValidationFailed(CollectionValidation::ValidateState* state,
+                                 ValidateResults* results) {
+    if (state->isTimeseriesDataInconsistent()) {
+        // Only report the warning message once.
+        return;
+    }
+    state->setTimeseriesDataInconsistent();
+
+    results->warnings.push_back(kTimeseriesValidationInconsistencyReason);
+}
 
 BSONObj rehydrateKey(const BSONObj& keyPattern, const BSONObj& indexKey) {
     // We need to rehydrate the indexKey for improved readability.
@@ -154,8 +308,9 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
                                        const RecordId& recordId,
                                        const RecordData& record,
                                        size_t* dataSize,
-                                       ValidateResults* results) {
-    const Status status = validateBSON(record.data(), record.size());
+                                       ValidateResults* results,
+                                       ValidationVersion validationVersion) {
+    const Status status = validateBSON(record.data(), record.size(), validationVersion);
     if (!status.isOK())
         return status;
 
@@ -328,15 +483,16 @@ namespace {
 // Ensures that index entries are in increasing or decreasing order.
 void _validateKeyOrder(OperationContext* opCtx,
                        const IndexCatalogEntry* index,
-                       const KeyString::Value& currKey,
-                       const KeyString::Value& prevKey,
+                       const KeyStringEntry& currKey,
+                       const KeyStringEntry& prevKey,
                        IndexValidateResults* results) {
     auto descriptor = index->descriptor();
     bool unique = descriptor->unique();
 
     // KeyStrings will be in strictly increasing order because all keys are sorted and they are in
     // the format (Key, RID), and all RecordIDs are unique.
-    if (currKey.compare(prevKey) <= 0 || MONGO_unlikely(failIndexKeyOrdering.shouldFail())) {
+    if (currKey.keyString.compare(prevKey.keyString) <= 0 ||
+        MONGO_unlikely(failIndexKeyOrdering.shouldFail())) {
         if (results && results->valid) {
             results->errors.push_back(str::stream()
                                       << "index '" << descriptor->indexName()
@@ -350,21 +506,20 @@ void _validateKeyOrder(OperationContext* opCtx,
 
     if (unique) {
         // Unique indexes must not have duplicate keys.
-        int cmp = currKey.compareWithoutRecordIdLong(prevKey);
+        int cmp = currKey.loc.isLong()
+            ? currKey.keyString.compareWithoutRecordIdLong(prevKey.keyString)
+            : currKey.keyString.compareWithoutRecordIdStr(prevKey.keyString);
         if (cmp != 0) {
             return;
         }
 
         if (results && results->valid) {
-            auto bsonKey = KeyString::toBson(currKey, Ordering::make(descriptor->keyPattern()));
-            auto firstRecordId =
-                KeyString::decodeRecordIdLongAtEnd(prevKey.getBuffer(), prevKey.getSize());
-            auto secondRecordId =
-                KeyString::decodeRecordIdLongAtEnd(currKey.getBuffer(), currKey.getSize());
+            auto bsonKey =
+                KeyString::toBson(currKey.keyString, Ordering::make(descriptor->keyPattern()));
             results->errors.push_back(str::stream() << "Unique index '" << descriptor->indexName()
                                                     << "' has duplicate key: " << bsonKey
-                                                    << ", first record: " << firstRecordId
-                                                    << ", second record: " << secondRecordId);
+                                                    << ", first record: " << prevKey.loc
+                                                    << ", second record: " << currKey.loc);
         }
         if (results) {
             results->valid = false;
@@ -383,8 +538,6 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
     IndexInfo& indexInfo = _indexConsistency->getIndexInfo(indexName);
     int64_t numKeys = 0;
 
-    bool isFirstEntry = true;
-
     // The progress meter will be inactive after traversing the record store to allow the message
     // and the total to be set to different values.
     if (!_progress->isActive()) {
@@ -399,7 +552,7 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
     KeyString::Builder firstKeyStringBuilder(
         version, BSONObj(), indexInfo.ord, KeyString::Discriminator::kExclusiveBefore);
     KeyString::Value firstKeyString = firstKeyStringBuilder.getValueCopy();
-    KeyString::Value prevIndexKeyStringValue;
+    boost::optional<KeyStringEntry> prevIndexKeyStringEntry;
 
     // Ensure that this index has an open index cursor.
     const auto indexCursorIt = _validateState->getIndexCursors().find(indexName);
@@ -430,9 +583,8 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
     bool foundOldUniqueIndexKeys = false;
 
     while (indexEntry) {
-        if (!isFirstEntry) {
-            _validateKeyOrder(
-                opCtx, index, indexEntry->keyString, prevIndexKeyStringValue, &indexResults);
+        if (prevIndexKeyStringEntry) {
+            _validateKeyOrder(opCtx, index, *indexEntry, *prevIndexKeyStringEntry, &indexResults);
         }
 
         if (!foundOldUniqueIndexKeys && !descriptor->isIdIndex() && descriptor->unique() &&
@@ -463,8 +615,7 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
 
         _progress->hit();
         numKeys++;
-        isFirstEntry = false;
-        prevIndexKeyStringValue = indexEntry->keyString;
+        prevIndexKeyStringEntry = indexEntry;
 
         if (numKeys % kInterruptIntervalNumRecords == 0) {
             // Periodically checks for interrupts and yields.
@@ -480,7 +631,7 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
                             "Error advancing index cursor",
                             "error"_attr = ex.toString(),
                             "index"_attr = indexName,
-                            "prevKey"_attr = prevIndexKeyStringValue.toString());
+                            "prevKey"_attr = prevIndexKeyStringEntry->keyString.toString());
             }
             throw;
         }
@@ -566,7 +717,8 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
 
 void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
                                           ValidateResults* results,
-                                          BSONObjBuilder* output) {
+                                          BSONObjBuilder* output,
+                                          ValidationVersion validationVersion) {
     _numRecords = 0;  // need to reset it because this function can be called more than once.
     long long dataSizeTotal = 0;
     long long interruptIntervalNumBytes = 0;
@@ -590,9 +742,10 @@ void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
 
     // Because the progress meter is intended as an approximation, it's sufficient to get the number
     // of records when we begin traversing, even if this number may deviate from the final number.
+    const auto& coll = _validateState->getCollection();
     const char* curopMessage = "Validate: scanning documents";
-    const auto totalRecords = _validateState->getCollection()->getRecordStore()->numRecords(opCtx);
-    const auto rs = _validateState->getCollection()->getRecordStore();
+    const auto totalRecords = coll->getRecordStore()->numRecords(opCtx);
+    const auto rs = coll->getRecordStore();
     {
         stdx::unique_lock<Client> lk(*opCtx->getClient());
         _progress.set(CurOp::get(opCtx)->setProgress_inlock(curopMessage, totalRecords));
@@ -603,6 +756,9 @@ void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
         return;
     }
 
+    bool bucketMixedSchemaDataError = false;
+    bool bucketMinMaxMalformedError = false;
+    bool bucketMixedSchemaDataWarning = false;
     bool corruptRecordsSizeLimitWarning = false;
     const std::unique_ptr<SeekableRecordThrottleCursor>& traverseRecordStoreCursor =
         _validateState->getTraverseRecordStoreCursor();
@@ -616,7 +772,26 @@ void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
         interruptIntervalNumBytes += dataSize;
         dataSizeTotal += dataSize;
         size_t validatedSize = 0;
-        Status status = validateRecord(opCtx, record->id, record->data, &validatedSize, results);
+        Status status = validateRecord(
+            opCtx, record->id, record->data, &validatedSize, results, validationVersion);
+
+        // Log the out-of-order entries as errors.
+        //
+        // Validate uses a DataCorruptionDetectionMode::kLogAndContinue mode such that data
+        // corruption errors are logged without throwing, so certain checks must be duplicated here
+        // as well.
+        if ((prevRecordId.isValid() && prevRecordId > record->id) ||
+            MONGO_unlikely(failRecordStoreTraversal.shouldFail())) {
+            // TODO SERVER-78040: Clean this up once we can insert errors blindly into the list and
+            // not care about deduplication.
+            static constexpr auto kErrorMessage = "Detected out-of-order documents. See logs.";
+            if (results->valid ||
+                std::find(results->errors.begin(), results->errors.end(), kErrorMessage) ==
+                    results->errors.end()) {
+                results->errors.push_back(kErrorMessage);
+                results->valid = false;
+            }
+        }
 
         // validatedSize = dataSize is not a general requirement as some storage engines may use
         // padding, but we still require that they return the unpadded record data.
@@ -646,8 +821,14 @@ void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
                 results->numRemovedCorruptRecords++;
                 _numRecords--;
             } else {
-                if (results->valid) {
-                    results->errors.push_back("Detected one or more invalid documents. See logs.");
+                // TODO SERVER-78040: Clean this up once we can insert errors blindly into the list
+                // and not care about deduplication.
+                static constexpr auto kErrorMessage =
+                    "Detected one or more invalid documents. See logs.";
+                if (results->valid ||
+                    std::find(results->errors.begin(), results->errors.end(), kErrorMessage) ==
+                        results->errors.end()) {
+                    results->errors.push_back(kErrorMessage);
                     results->valid = false;
                 }
 
@@ -666,18 +847,76 @@ void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
             // If the document is not corrupted, validate the document against this collection's
             // schema validator. Don't treat invalid documents as errors since documents can bypass
             // document validation when being inserted or updated.
-            auto result =
-                _validateState->getCollection()->checkValidation(opCtx, record->data.toBson());
+            auto result = coll->checkValidation(opCtx, record->data.toBson());
 
             if (result.first != Collection::SchemaValidationResult::kPass) {
                 LOGV2_WARNING(5363500,
                               "Document is not compliant with the collection's schema",
-                              logAttrs(_validateState->getCollection()->ns()),
+                              logAttrs(coll->ns()),
                               "recordId"_attr = record->id,
                               "reason"_attr = result.second);
 
                 nNonCompliantDocuments++;
                 schemaValidationFailed(_validateState, result.first, results);
+            } else if (coll->getTimeseriesOptions()) {
+                // Checks for time-series collection consistency.
+                Status bucketStatus =
+                    _validateTimeSeriesBucketRecord(coll, record->data.toBson(), results);
+
+                // This log id should be kept in sync with the associated warning messages that are
+                // returned to the client.
+                if (!bucketStatus.isOK()) {
+                    LOGV2_WARNING(6698300,
+                                  "Document is not compliant with time-series specifications",
+                                  logAttrs(coll->ns()),
+                                  "recordId"_attr = record->id,
+                                  "reason"_attr = bucketStatus);
+                    nNonCompliantDocuments++;
+                    _timeseriesValidationFailed(_validateState, results);
+                } else {
+                    auto containsMixedSchemaDataResponse =
+                        coll->doesTimeseriesBucketsDocContainMixedSchemaData(record->data.toBson());
+                    if (!containsMixedSchemaDataResponse.isOK() && !bucketMinMaxMalformedError) {
+                        bucketMinMaxMalformedError = true;
+                        LOGV2_WARNING(8469900,
+                                      "Detected a time-series bucket with malformed min/max values",
+                                      logAttrs(coll->ns()),
+                                      "bucketId"_attr = record->id,
+                                      "error"_attr = containsMixedSchemaDataResponse.getStatus());
+                        results->errors.push_back(
+                            str::stream()
+                            << "Detected a time-series bucket with malformed min/max values");
+                        results->valid = false;
+                    } else if (containsMixedSchemaDataResponse.isOK() &&
+                               containsMixedSchemaDataResponse.getValue()) {
+                        bool mixedSchemaAllowed =
+                            coll->getTimeseriesBucketsMayHaveMixedSchemaData().value_or(true);
+                        if (mixedSchemaAllowed && !bucketMixedSchemaDataWarning) {
+                            bucketMixedSchemaDataWarning = true;
+                            LOGV2_WARNING(8469901,
+                                          "Detected a time-series bucket with mixed schema data",
+                                          logAttrs(coll->ns()),
+                                          "bucketId"_attr = record->id);
+                            results->warnings.push_back(
+                                str::stream()
+                                << "Detected a time-series bucket with mixed schema data");
+                        } else if (!mixedSchemaAllowed && !bucketMixedSchemaDataError) {
+                            bucketMixedSchemaDataError = true;
+                            LOGV2_WARNING(8469902,
+                                          "Detected a time-series bucket with mixed schema data "
+                                          "when timeseriesBucketsMayHaveMixedSchemaData is false. "
+                                          "You can run the collMod command to set this flag",
+                                          logAttrs(coll->ns()),
+                                          "bucketId"_attr = record->id);
+                            results->errors.push_back(
+                                str::stream()
+                                << "Detected a time-series bucket with mixed schema data when "
+                                   "timeseriesBucketsMayHaveMixedSchemaData is false. You can run "
+                                   "the collMod command to set this flag");
+                            results->valid = false;
+                        }
+                    }
+                }
             }
         }
 
@@ -700,20 +939,18 @@ void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
                                                   << " invalid documents.");
     }
 
-    const auto fastCount = _validateState->getCollection()->numRecords(opCtx);
+    const auto fastCount = coll->numRecords(opCtx);
     if (_validateState->shouldEnforceFastCount() && fastCount != _numRecords) {
-        results->errors.push_back(str::stream() << "fast count (" << fastCount
-                                                << ") does not match number of records ("
-                                                << _numRecords << ") for collection '"
-                                                << _validateState->getCollection()->ns() << "'");
+        results->errors.push_back(
+            str::stream() << "fast count (" << fastCount << ") does not match number of records ("
+                          << _numRecords << ") for collection '" << coll->ns() << "'");
         results->valid = false;
     }
 
     // Do not update the record store stats if we're in the background as we've validated a
     // checkpoint and it may not have the most up-to-date changes.
     if (results->valid && !_validateState->isBackground()) {
-        _validateState->getCollection()->getRecordStore()->updateStatsAfterRepair(
-            opCtx, _numRecords, dataSizeTotal);
+        coll->getRecordStore()->updateStatsAfterRepair(opCtx, _numRecords, dataSizeTotal);
     }
 }
 
diff --git a/src/mongo/db/catalog/validate_adaptor.h b/src/mongo/db/catalog/validate_adaptor.h
index ba548ca6358..99c0a2fceb1 100644
--- a/src/mongo/db/catalog/validate_adaptor.h
+++ b/src/mongo/db/catalog/validate_adaptor.h
@@ -57,7 +57,8 @@ public:
                                   const RecordId& recordId,
                                   const RecordData& record,
                                   size_t* dataSize,
-                                  ValidateResults* results);
+                                  ValidateResults* results,
+                                  ValidationVersion validationVersion = currentValidationVersion);
 
     /**
      * Traverses the index getting index entries to validate them and keep track of the index keys
@@ -74,7 +75,8 @@ public:
      */
     void traverseRecordStore(OperationContext* opCtx,
                              ValidateResults* results,
-                             BSONObjBuilder* output);
+                             BSONObjBuilder* output,
+                             ValidationVersion validationVersion);
 
     /**
      * Validates that the number of document keys matches the number of index keys previously
diff --git a/src/mongo/db/catalog/validate_state.cpp b/src/mongo/db/catalog/validate_state.cpp
index f40ebe0062e..01f80ac61cc 100644
--- a/src/mongo/db/catalog/validate_state.cpp
+++ b/src/mongo/db/catalog/validate_state.cpp
@@ -81,16 +81,26 @@ ValidateState::ValidateState(OperationContext* opCtx,
         _collection = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, _nss);
 
     if (!_collection) {
-        if (CollectionCatalog::get(opCtx)->lookupView(opCtx, _nss)) {
-            uasserted(ErrorCodes::CommandNotSupportedOnView, "Cannot validate a view");
+        auto view = CollectionCatalog::get(opCtx)->lookupView(opCtx, _nss);
+        if (!view) {
+            uasserted(ErrorCodes::NamespaceNotFound,
+                      str::stream() << "Collection '" << _nss << "' does not exist to validate.");
+        } else {
+            // Uses the bucket collection in place of the time-series collection view.
+            if (!view->timeseries()) {
+                uasserted(ErrorCodes::CommandNotSupportedOnView, "Cannot validate a view");
+            }
+            _nss = _nss.makeTimeseriesBucketsNamespace();
+            if (isBackground()) {
+                _collectionLock.emplace(opCtx, _nss, MODE_IS);
+            } else {
+                _collectionLock.emplace(opCtx, _nss, MODE_X);
+            }
+            _collection = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, _nss);
         }
-
-        uasserted(ErrorCodes::NamespaceNotFound,
-                  str::stream() << "Collection '" << _nss << "' does not exist to validate.");
     }
 
-    // Return warnings instead of errors on schema validation failures.
-    _warnOnSchemaValidation = additionalOptions.warnOnSchemaValidation;
+    _validationVersion = additionalOptions.validationVersion;
 
     // RepairMode is incompatible with the ValidateModes kBackground and
     // kForegroundFullEnforceFastCount.
diff --git a/src/mongo/db/catalog/validate_state.h b/src/mongo/db/catalog/validate_state.h
index 42bc8660e33..bb8f431d2d6 100644
--- a/src/mongo/db/catalog/validate_state.h
+++ b/src/mongo/db/catalog/validate_state.h
@@ -93,6 +93,13 @@ public:
         _collectionSchemaViolated = true;
     }
 
+    bool isTimeseriesDataInconsistent() {
+        return _timeseriesDataInconsistency;
+    }
+    void setTimeseriesDataInconsistent() {
+        _timeseriesDataInconsistency = true;
+    }
+
     bool fixErrors() const {
         return _repairMode == RepairMode::kFixErrors;
     }
@@ -168,8 +175,8 @@ public:
         return _logDiagnostics;
     }
 
-    bool warnOnSchemaValidation() const {
-        return _warnOnSchemaValidation;
+    ValidationVersion validationVersion() const {
+        return _validationVersion;
     }
 
     boost::optional<Timestamp> getValidateTimestamp() {
@@ -214,7 +221,8 @@ private:
     ValidateMode _mode;
     RepairMode _repairMode;
     bool _collectionSchemaViolated = false;
-    bool _warnOnSchemaValidation = false;
+    bool _timeseriesDataInconsistency = false;
+    ValidationVersion _validationVersion = currentValidationVersion;
 
     boost::optional<ShouldNotConflictWithSecondaryBatchApplicationBlock> _noPBWM;
     boost::optional<Lock::GlobalLock> _globalLock;