SERVER-23116 Add versioning scheme to the KVCatalog.

The feature document is inserted into the KVCatalog only after the first
feature is enabled on the data files.

6 files changed, 931 insertions, 3 deletions

diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index c4756f9a5da..acfbeb9b537 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -7,6 +7,7 @@ env.Library(
         'kv_collection_catalog_entry.cpp',
         ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/bson/util/bson_extract',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/index/index_descriptor',
         '$BUILD_DIR/mongo/db/index_names',
@@ -83,6 +84,7 @@ env.Library(
 env.Library(
     target='kv_engine_test_harness',
     source=[
+        'kv_catalog_feature_tracker_test.cpp',
         'kv_engine_test_harness.cpp',
         'kv_engine_test_snapshots.cpp',
         ],
diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index bff423434b4..8141ca89f01 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -34,11 +34,15 @@
 
 #include <stdlib.h>
 
+#include "mongo/bson/util/bson_extract.h"
+#include "mongo/bson/util/builder.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/recovery_unit.h"
+#include "mongo/platform/bits.h"
 #include "mongo/platform/random.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
@@ -51,6 +55,28 @@ namespace {
 //
 // NOTE: Must be locked *before* _identLock.
 const ResourceId resourceIdCatalogMetadata(RESOURCE_METADATA, 1ULL);
+
+const char kIsFeatureDocumentFieldName[] = "isFeatureDoc";
+const char kNamespaceFieldName[] = "ns";
+const char kNonRepairableFeaturesFieldName[] = "nonRepairable";
+const char kRepairableFeaturesFieldName[] = "repairable";
+
+void appendPositionsOfBitsSet(uint64_t value, StringBuilder* sb) {
+    invariant(sb);
+
+    *sb << "[ ";
+    bool firstIteration = true;
+    while (value) {
+        const int lowestSetBitPosition = countTrailingZeros64(value);
+        if (!firstIteration) {
+            *sb << ", ";
+        }
+        *sb << lowestSetBitPosition;
+        value ^= (1 << lowestSetBitPosition);
+        firstIteration = false;
+    }
+    *sb << " ]";
+}
 }
 
 using std::unique_ptr;
@@ -87,6 +113,192 @@ public:
     const Entry _entry;
 };
 
+bool KVCatalog::FeatureTracker::isFeatureDocument(BSONObj obj) {
+    BSONElement firstElem = obj.firstElement();
+    if (firstElem.fieldNameStringData() == kIsFeatureDocumentFieldName) {
+        return firstElem.booleanSafe();
+    }
+    return false;
+}
+
+Status KVCatalog::FeatureTracker::isCompatibleWithCurrentCode(OperationContext* opCtx) const {
+    FeatureBits versionInfo = getInfo(opCtx);
+
+    uint64_t unrecognizedNonRepairableFeatures =
+        versionInfo.nonRepairableFeatures & ~_usedNonRepairableFeaturesMask;
+    if (unrecognizedNonRepairableFeatures) {
+        StringBuilder sb;
+        sb << "The data files use features not recognized by this version of mongod; the NR feature"
+              " bits in positions ";
+        appendPositionsOfBitsSet(unrecognizedNonRepairableFeatures, &sb);
+        sb << " aren't recognized by this version of mongod";
+        return {ErrorCodes::MustUpgrade, sb.str()};
+    }
+
+    uint64_t unrecognizedRepairableFeatures =
+        versionInfo.repairableFeatures & ~_usedRepairableFeaturesMask;
+    if (unrecognizedRepairableFeatures) {
+        StringBuilder sb;
+        sb << "The data files use features not recognized by this version of mongod; the R feature"
+              " bits in positions ";
+        appendPositionsOfBitsSet(unrecognizedRepairableFeatures, &sb);
+        sb << " aren't recognized by this version of mongod";
+        return {ErrorCodes::CanRepairToDowngrade, sb.str()};
+    }
+
+    return Status::OK();
+}
+
+std::unique_ptr<KVCatalog::FeatureTracker> KVCatalog::FeatureTracker::get(OperationContext* opCtx,
+                                                                          KVCatalog* catalog,
+                                                                          RecordId rid) {
+    auto record = catalog->_rs->dataFor(opCtx, rid);
+    BSONObj obj = record.toBson();
+    invariant(isFeatureDocument(obj));
+    return std::unique_ptr<KVCatalog::FeatureTracker>(new KVCatalog::FeatureTracker(catalog, rid));
+}
+
+std::unique_ptr<KVCatalog::FeatureTracker> KVCatalog::FeatureTracker::create(
+    OperationContext* opCtx, KVCatalog* catalog) {
+    return std::unique_ptr<KVCatalog::FeatureTracker>(
+        new KVCatalog::FeatureTracker(catalog, RecordId()));
+}
+
+bool KVCatalog::FeatureTracker::isNonRepairableFeatureInUse(OperationContext* opCtx,
+                                                            NonRepairableFeature feature) const {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_S));
+    }
+
+    FeatureBits versionInfo = getInfo(opCtx);
+    return versionInfo.nonRepairableFeatures & static_cast<NonRepairableFeatureMask>(feature);
+}
+
+void KVCatalog::FeatureTracker::markNonRepairableFeatureAsInUse(OperationContext* opCtx,
+                                                                NonRepairableFeature feature) {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
+    }
+
+    FeatureBits versionInfo = getInfo(opCtx);
+    versionInfo.nonRepairableFeatures |= static_cast<NonRepairableFeatureMask>(feature);
+    putInfo(opCtx, versionInfo);
+}
+
+void KVCatalog::FeatureTracker::markNonRepairableFeatureAsNotInUse(OperationContext* opCtx,
+                                                                   NonRepairableFeature feature) {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
+    }
+
+    FeatureBits versionInfo = getInfo(opCtx);
+    versionInfo.nonRepairableFeatures &= ~static_cast<NonRepairableFeatureMask>(feature);
+    putInfo(opCtx, versionInfo);
+}
+
+bool KVCatalog::FeatureTracker::isRepairableFeatureInUse(OperationContext* opCtx,
+                                                         RepairableFeature feature) const {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_S));
+    }
+
+    FeatureBits versionInfo = getInfo(opCtx);
+    return versionInfo.repairableFeatures & static_cast<RepairableFeatureMask>(feature);
+}
+
+void KVCatalog::FeatureTracker::markRepairableFeatureAsInUse(OperationContext* opCtx,
+                                                             RepairableFeature feature) {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
+    }
+
+    FeatureBits versionInfo = getInfo(opCtx);
+    versionInfo.repairableFeatures |= static_cast<RepairableFeatureMask>(feature);
+    putInfo(opCtx, versionInfo);
+}
+
+void KVCatalog::FeatureTracker::markRepairableFeatureAsNotInUse(OperationContext* opCtx,
+                                                                RepairableFeature feature) {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
+    }
+
+    FeatureBits versionInfo = getInfo(opCtx);
+    versionInfo.repairableFeatures &= ~static_cast<RepairableFeatureMask>(feature);
+    putInfo(opCtx, versionInfo);
+}
+
+KVCatalog::FeatureTracker::FeatureBits KVCatalog::FeatureTracker::getInfo(
+    OperationContext* opCtx) const {
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        invariant(opCtx->lockState()->isLockHeldForMode(resourceIdCatalogMetadata, MODE_S));
+    }
+
+    if (_rid.isNull()) {
+        return {};
+    }
+
+    auto record = _catalog->_rs->dataFor(opCtx, _rid);
+    BSONObj obj = record.toBson();
+    invariant(isFeatureDocument(obj));
+
+    BSONElement nonRepairableFeaturesElem;
+    auto nonRepairableFeaturesStatus = bsonExtractTypedField(
+        obj, kNonRepairableFeaturesFieldName, BSONType::NumberLong, &nonRepairableFeaturesElem);
+    fassert(40111, nonRepairableFeaturesStatus);
+
+    BSONElement repairableFeaturesElem;
+    auto repairableFeaturesStatus = bsonExtractTypedField(
+        obj, kRepairableFeaturesFieldName, BSONType::NumberLong, &repairableFeaturesElem);
+    fassert(40112, repairableFeaturesStatus);
+
+    FeatureBits versionInfo;
+    versionInfo.nonRepairableFeatures =
+        static_cast<NonRepairableFeatureMask>(nonRepairableFeaturesElem.numberLong());
+    versionInfo.repairableFeatures =
+        static_cast<RepairableFeatureMask>(repairableFeaturesElem.numberLong());
+    return versionInfo;
+}
+
+void KVCatalog::FeatureTracker::putInfo(OperationContext* opCtx, const FeatureBits& versionInfo) {
+    if (!_catalog->_isRsThreadSafe && opCtx->lockState()) {
+        invariant(opCtx->lockState()->isLockHeldForMode(resourceIdCatalogMetadata, MODE_X));
+    }
+
+    BSONObjBuilder bob;
+    bob.appendBool(kIsFeatureDocumentFieldName, true);
+    // We intentionally include the "ns" field with a null value in the feature document to prevent
+    // older versions that do 'obj["ns"].String()' from starting up. This way only versions that are
+    // aware of the feature document's existence can successfully start up.
+    bob.appendNull(kNamespaceFieldName);
+    bob.append(kNonRepairableFeaturesFieldName,
+               static_cast<long long>(versionInfo.nonRepairableFeatures));
+    bob.append(kRepairableFeaturesFieldName,
+               static_cast<long long>(versionInfo.repairableFeatures));
+    BSONObj obj = bob.done();
+
+    if (_rid.isNull()) {
+        // This is the first time a feature is being marked as in-use or not in-use, so we must
+        // insert the feature document rather than update it.
+        const bool enforceQuota = false;
+        auto rid = _catalog->_rs->insertRecord(opCtx, obj.objdata(), obj.objsize(), enforceQuota);
+        fassert(40113, rid.getStatus());
+        _rid = rid.getValue();
+    } else {
+        const bool enforceQuota = false;
+        UpdateNotifier* notifier = nullptr;
+        auto status = _catalog->_rs->updateRecord(
+            opCtx, _rid, obj.objdata(), obj.objsize(), enforceQuota, notifier);
+        fassert(40114, status);
+    }
+}
+
 KVCatalog::KVCatalog(RecordStore* rs,
                      bool isRsThreadSafe,
                      bool directoryPerDb,
@@ -132,12 +344,28 @@ void KVCatalog::init(OperationContext* opCtx) {
     while (auto record = cursor->next()) {
         BSONObj obj = record->data.releaseToBson();
 
+        if (FeatureTracker::isFeatureDocument(obj)) {
+            // There should be at most one version document in the catalog.
+            invariant(!_featureTracker);
+
+            // Initialize the feature tracker and skip over the version document because it doesn't
+            // correspond to a namespace entry.
+            _featureTracker = std::move(FeatureTracker::get(opCtx, this, record->id));
+            continue;
+        }
+
         // No rollback since this is just loading already committed data.
         string ns = obj["ns"].String();
         string ident = obj["ident"].String();
         _idents[ns] = Entry(ident, record->id);
     }
 
+    if (!_featureTracker) {
+        // If there wasn't a feature document, then just an initialize a feature tracker that
+        // doesn't manage a feature document yet.
+        _featureTracker = std::move(KVCatalog::FeatureTracker::create(opCtx, this));
+    }
+
     // In the unlikely event that we have used this _rand before generate a new one.
     while (_hasEntryCollidingWithRand()) {
         _rand = _newRand();
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
index 577fdba1faf..d84edfdde3b 100644
--- a/src/mongo/db/storage/kv/kv_catalog.h
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -31,6 +31,7 @@
 #pragma once
 
 #include <map>
+#include <memory>
 #include <string>
 
 #include "mongo/base/string_data.h"
@@ -46,6 +47,8 @@ class RecordStore;
 
 class KVCatalog {
 public:
+    class FeatureTracker;
+
     /**
      * @param rs - does NOT take ownership
      */
@@ -82,6 +85,11 @@ public:
 
     bool isUserDataIdent(StringData ident) const;
 
+    FeatureTracker* getFeatureTracker() const {
+        invariant(_featureTracker);
+        return _featureTracker.get();
+    }
+
 private:
     class AddIdentChange;
     class RemoveIdentChange;
@@ -117,5 +125,9 @@ private:
     typedef std::map<std::string, Entry> NSToIdentMap;
     NSToIdentMap _idents;
     mutable stdx::mutex _identsLock;
+
+    // Manages the feature document that may be present in the KVCatalog. '_featureTracker' is
+    // guaranteed to be non-null after KVCatalog::init() is called.
+    std::unique_ptr<FeatureTracker> _featureTracker;
 };
 }
diff --git a/src/mongo/db/storage/kv/kv_catalog_feature_tracker.h b/src/mongo/db/storage/kv/kv_catalog_feature_tracker.h
new file mode 100644
index 00000000000..5a0654c9cc8
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_catalog_feature_tracker.h
@@ -0,0 +1,188 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <type_traits>
+
+#include "mongo/db/storage/kv/kv_catalog.h"
+
+namespace mongo {
+
+class OperationContext;
+class RecordId;
+class RecordStore;
+
+/**
+ * Manages the contents of a document in the KVCatalog used to restrict downgrade compatibility.
+ *
+ * When a new feature is enabled on a collection or index in the data files, a bit is set in one of
+ * the fields of the document. Older versions won't recognize this bit and will fail to start up as
+ * a result.
+ *
+ * The inserted document serves a similar purpose to the DataFileVersion class used with the MMAPv1
+ * storage engine.
+ */
+class KVCatalog::FeatureTracker {
+public:
+    /**
+     * Bit flags representing whether a particular feature is enabled on a least one collection or
+     * index in the data files. Features included in this enumeration always require user
+     * intervention on downgrade.
+     *
+     * The next feature added to this enumeration should use the current value of 'kNextFeatureBit',
+     * and 'kNextFeatureBit' should be changed to the next largest power of two.
+     */
+    enum class NonRepairableFeature : std::uint64_t { kNextFeatureBit = 1 << 0 };
+
+    using NonRepairableFeatureMask = std::underlying_type<NonRepairableFeature>::type;
+
+    /**
+     * Bit flags representing whether a particular feature is enabled on a least one collection or
+     * index in the data files. Features included in this enumeration either (a) don't require user
+     * intervention on downgrade, or (b) are no longer enabled if --repair is done with an older
+     * version.
+     *
+     * The next feature added to this enumeration should use the current value of 'kNextFeatureBit',
+     * and 'kNextFeatureBit' should be changed to the next largest power of two.
+     */
+    enum class RepairableFeature : std::uint64_t { kNextFeatureBit = 1 << 0 };
+
+    using RepairableFeatureMask = std::underlying_type<RepairableFeature>::type;
+
+    /**
+     * Returns true if 'obj' represents the contents of the feature document that was previously
+     * inserted into the KVCatalog, and returns false otherwise.
+     *
+     * This function should return true for at most one document in the KVCatalog.
+     */
+    static bool isFeatureDocument(BSONObj obj);
+
+    /**
+     * Returns a FeatureTracker instance to manage the contents of the feature document located at
+     * 'rid' in the record store 'catalog->_rs'.
+     *
+     * It is invalid to call this function when isFeatureDocument() returns false for the record
+     * data associated with 'rid'.
+     */
+    static std::unique_ptr<FeatureTracker> get(OperationContext* opCtx,
+                                               KVCatalog* catalog,
+                                               RecordId rid);
+
+    /**
+     * Returns a FeatureTracker instance to manage the contents of a feature document. The feature
+     * document isn't inserted into 'rs' as a result of calling this function. Instead, the feature
+     * document is inserted into 'rs' when putInfo() is first called.
+     *
+     * It is invalid to call this function when isFeatureDocument() returns true for some document
+     * in the record store 'catalog->_rs'.
+     */
+    static std::unique_ptr<FeatureTracker> create(OperationContext* opCtx, KVCatalog* catalog);
+
+    /**
+     * Returns whethers the data files are compatible with the current code:
+     *
+     *   - Status::OK() if the data files are compatible with the current code.
+     *
+     *   - ErrorCodes::CanRepairToDowngrade if the data files are incompatible with the current
+     *     code, but a --repair would make them compatible. For example, when rebuilding all indexes
+     *     in the data files would resolve the incompatibility.
+     *
+     *   - ErrorCodes::MustUpgrade if the data files are incompatible with the current code and a
+     *     newer version is required to start up.
+     */
+    Status isCompatibleWithCurrentCode(OperationContext* opCtx) const;
+
+    /**
+     * Returns true if 'feature' is tracked in the document, and returns false otherwise.
+     */
+    bool isNonRepairableFeatureInUse(OperationContext* opCtx, NonRepairableFeature feature) const;
+
+    /**
+     * Sets the specified non-repairable feature as being enabled on at least one collection or
+     * index in the data files.
+     */
+    void markNonRepairableFeatureAsInUse(OperationContext* opCtx, NonRepairableFeature feature);
+
+    /**
+     * Sets the specified non-repairable feature as not being enabled on any collection or index in
+     * the data files.
+     */
+    void markNonRepairableFeatureAsNotInUse(OperationContext* opCtx, NonRepairableFeature feature);
+
+    /**
+     * Returns true if 'feature' is tracked in the document, and returns false otherwise.
+     */
+    bool isRepairableFeatureInUse(OperationContext* opCtx, RepairableFeature feature) const;
+
+    /**
+     * Sets the specified repairable feature as being enabled on at least one collection or index in
+     * the data files.
+     */
+    void markRepairableFeatureAsInUse(OperationContext* opCtx, RepairableFeature feature);
+
+    /**
+     * Sets the specified repairable feature as not being enabled on any collection or index in the
+     * data files.
+     */
+    void markRepairableFeatureAsNotInUse(OperationContext* opCtx, RepairableFeature feature);
+
+    void setUsedNonRepairableFeaturesMaskForTestingOnly(NonRepairableFeatureMask mask) {
+        _usedNonRepairableFeaturesMask = mask;
+    }
+
+    void setUsedRepairableFeaturesMaskForTestingOnly(RepairableFeatureMask mask) {
+        _usedRepairableFeaturesMask = mask;
+    }
+
+private:
+    // Must go through FeatureTracker::get() or FeatureTracker::create().
+    FeatureTracker(KVCatalog* catalog, RecordId rid) : _catalog(catalog), _rid(rid) {}
+
+    struct FeatureBits {
+        NonRepairableFeatureMask nonRepairableFeatures;
+        RepairableFeatureMask repairableFeatures;
+    };
+
+    FeatureBits getInfo(OperationContext* opCtx) const;
+
+    void putInfo(OperationContext* opCtx, const FeatureBits& versionInfo);
+
+    KVCatalog* _catalog;
+    RecordId _rid;
+
+    NonRepairableFeatureMask _usedNonRepairableFeaturesMask =
+        static_cast<NonRepairableFeatureMask>(NonRepairableFeature::kNextFeatureBit) - 1;
+
+    RepairableFeatureMask _usedRepairableFeaturesMask =
+        static_cast<RepairableFeatureMask>(RepairableFeature::kNextFeatureBit) - 1;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/storage/kv/kv_catalog_feature_tracker_test.cpp b/src/mongo/db/storage/kv/kv_catalog_feature_tracker_test.cpp
new file mode 100644
index 00000000000..41e62e12c75
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_catalog_feature_tracker_test.cpp
@@ -0,0 +1,498 @@
+/**
+ *    Copyright (C) 2016 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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/platform/basic.h"
+
+#include "mongo/db/storage/kv/kv_engine_test_harness.h"
+
+#include "mongo/db/operation_context_noop.h"
+#include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
+#include "mongo/db/storage/kv/kv_engine.h"
+#include "mongo/db/storage/record_store.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+using NonRepairableFeature = KVCatalog::FeatureTracker::NonRepairableFeature;
+using NonRepairableFeatureMask = KVCatalog::FeatureTracker::NonRepairableFeatureMask;
+using RepairableFeature = KVCatalog::FeatureTracker::RepairableFeature;
+using RepairableFeatureMask = KVCatalog::FeatureTracker::RepairableFeatureMask;
+
+class KVCatalogFeatureTrackerTest : public unittest::Test {
+public:
+    static const NonRepairableFeature kNonRepairableFeature1 =
+        static_cast<NonRepairableFeature>(1 << 0);
+
+    static const NonRepairableFeature kNonRepairableFeature2 =
+        static_cast<NonRepairableFeature>(1 << 1);
+
+    static const NonRepairableFeature kNonRepairableFeature3 =
+        static_cast<NonRepairableFeature>(1 << 2);
+
+    static const RepairableFeature kRepairableFeature1 = static_cast<RepairableFeature>(1 << 0);
+
+    static const RepairableFeature kRepairableFeature2 = static_cast<RepairableFeature>(1 << 1);
+
+    static const RepairableFeature kRepairableFeature3 = static_cast<RepairableFeature>(1 << 2);
+
+    KVCatalogFeatureTrackerTest() : _helper(KVHarnessHelper::create()) {}
+
+    std::unique_ptr<OperationContext> newOperationContext() {
+        return stdx::make_unique<OperationContextNoop>(_helper->getEngine()->newRecoveryUnit());
+    }
+
+    void setUp() final {
+        auto opCtx = newOperationContext();
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            ASSERT_OK(_helper->getEngine()->createRecordStore(
+                opCtx.get(), "catalog", "catalog", CollectionOptions()));
+            _rs.reset(_helper->getEngine()->getRecordStore(
+                opCtx.get(), "catalog", "catalog", CollectionOptions()));
+            wuow.commit();
+        }
+
+        _catalog = stdx::make_unique<KVCatalog>(_rs.get(), true, false, false);
+        _catalog->init(opCtx.get());
+
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            _featureTracker = KVCatalog::FeatureTracker::create(opCtx.get(), _catalog.get());
+            wuow.commit();
+        }
+    }
+
+    RecordStore* getRecordStore() const {
+        return _rs.get();
+    }
+
+    KVCatalog::FeatureTracker* getFeatureTracker() const {
+        return _featureTracker.get();
+    }
+
+private:
+    std::unique_ptr<KVHarnessHelper> _helper;
+    std::unique_ptr<RecordStore> _rs;
+    std::unique_ptr<KVCatalog> _catalog;
+    std::unique_ptr<KVCatalog::FeatureTracker> _featureTracker;
+};
+
+TEST_F(KVCatalogFeatureTrackerTest, FeatureDocumentIsNotEagerlyCreated) {
+    auto opCtx = newOperationContext();
+    auto cursor = getRecordStore()->getCursor(opCtx.get());
+    ASSERT_FALSE(static_cast<bool>(cursor->next()));
+}
+
+TEST_F(KVCatalogFeatureTrackerTest, CanMarkNonRepairableFeatureAsInUse) {
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+
+        // Marking the same non-repairable feature as in-use again does nothing.
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+    }
+
+    // The repairable feature bit in the same position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+    }
+
+    // The non-repairable feature in a different position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature2));
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest, CanMarkNonRepairableFeatureAsNotInUse) {
+    {
+        auto opCtx = newOperationContext();
+
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature2));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature2);
+            wuow.commit();
+        }
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature2));
+    }
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsNotInUse(opCtx.get(),
+                                                                    kNonRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+
+        // Marking the same non-repairable feature as not in-use again does nothing.
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsNotInUse(opCtx.get(),
+                                                                    kNonRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+    }
+
+    // The repairable feature bit in the same position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+    }
+
+    // The non-repairable feature in a different position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature2));
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest, CanMarkRepairableFeatureAsInUse) {
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+
+        // Marking the same repairable feature as in-use again does nothing.
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+    }
+
+    // The non-repairable feature bit in the same position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+    }
+
+    // The repairable feature in a different position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature2));
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest, CanMarkRepairableFeatureAsNotInUse) {
+    {
+        auto opCtx = newOperationContext();
+
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+
+        ASSERT(
+            !getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature2));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature2);
+            wuow.commit();
+        }
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature2));
+    }
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsNotInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+
+        // Marking the same repairable feature as not in-use again does nothing.
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsNotInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT(!getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature1));
+    }
+
+    // The non-repairable feature bit in the same position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(
+            getFeatureTracker()->isNonRepairableFeatureInUse(opCtx.get(), kNonRepairableFeature1));
+    }
+
+    // The repairable feature in a different position is unaffected.
+    {
+        auto opCtx = newOperationContext();
+        ASSERT(getFeatureTracker()->isRepairableFeatureInUse(opCtx.get(), kRepairableFeature2));
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest, DataFileAreCompatibleWithRecognizedNonRepairableFeature) {
+    getFeatureTracker()->setUsedNonRepairableFeaturesMaskForTestingOnly(0ULL);
+    getFeatureTracker()->setUsedRepairableFeaturesMaskForTestingOnly(0ULL);
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+
+    getFeatureTracker()->setUsedNonRepairableFeaturesMaskForTestingOnly(
+        static_cast<NonRepairableFeatureMask>(kNonRepairableFeature1));
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest,
+       DataFilesAreIncompatibleWithAnUnrecognizedNonRepairableFeature) {
+    getFeatureTracker()->setUsedNonRepairableFeaturesMaskForTestingOnly(0ULL);
+    getFeatureTracker()->setUsedRepairableFeaturesMaskForTestingOnly(0ULL);
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+
+    {
+        auto opCtx = newOperationContext();
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature1);
+            wuow.commit();
+        }
+
+        auto status = getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get());
+        ASSERT_EQ(ErrorCodes::MustUpgrade, status.code());
+        ASSERT_EQ(
+            "The data files use features not recognized by this version of mongod; the NR feature"
+            " bits in positions [ 0 ] aren't recognized by this version of mongod",
+            status.reason());
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest,
+       DataFilesAreIncompatibleWithMultipleUnrecognizedNonRepairableFeatures) {
+    getFeatureTracker()->setUsedNonRepairableFeaturesMaskForTestingOnly(
+        static_cast<NonRepairableFeatureMask>(kNonRepairableFeature1));
+    getFeatureTracker()->setUsedRepairableFeaturesMaskForTestingOnly(0ULL);
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+
+    {
+        auto opCtx = newOperationContext();
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature2);
+            wuow.commit();
+        }
+
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx.get(),
+                                                                 kNonRepairableFeature3);
+            wuow.commit();
+        }
+
+        auto status = getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get());
+        ASSERT_EQ(ErrorCodes::MustUpgrade, status.code());
+        ASSERT_EQ(
+            "The data files use features not recognized by this version of mongod; the NR feature"
+            " bits in positions [ 1, 2 ] aren't recognized by this version of mongod",
+            status.reason());
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest, DataFilesAreCompatibleWithRecognizedRepairableFeature) {
+    getFeatureTracker()->setUsedNonRepairableFeaturesMaskForTestingOnly(0ULL);
+    getFeatureTracker()->setUsedRepairableFeaturesMaskForTestingOnly(0ULL);
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+
+    getFeatureTracker()->setUsedRepairableFeaturesMaskForTestingOnly(
+        static_cast<RepairableFeatureMask>(kRepairableFeature1));
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest, DataFilesAreIncompatibleWithAnUnrecognizedRepairableFeature) {
+    getFeatureTracker()->setUsedNonRepairableFeaturesMaskForTestingOnly(0ULL);
+    getFeatureTracker()->setUsedRepairableFeaturesMaskForTestingOnly(0ULL);
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+
+    {
+        auto opCtx = newOperationContext();
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature1);
+            wuow.commit();
+        }
+
+        auto status = getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get());
+        ASSERT_EQ(ErrorCodes::CanRepairToDowngrade, status.code());
+        ASSERT_EQ(
+            "The data files use features not recognized by this version of mongod; the R feature"
+            " bits in positions [ 0 ] aren't recognized by this version of mongod",
+            status.reason());
+    }
+}
+
+TEST_F(KVCatalogFeatureTrackerTest,
+       DataFilesAreIncompatibleWithMultipleUnrecognizedRepairableFeatures) {
+    getFeatureTracker()->setUsedNonRepairableFeaturesMaskForTestingOnly(0ULL);
+    getFeatureTracker()->setUsedRepairableFeaturesMaskForTestingOnly(
+        static_cast<RepairableFeatureMask>(kRepairableFeature1));
+
+    {
+        auto opCtx = newOperationContext();
+        ASSERT_OK(getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get()));
+    }
+
+    {
+        auto opCtx = newOperationContext();
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature2);
+            wuow.commit();
+        }
+
+        {
+            WriteUnitOfWork wuow(opCtx.get());
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx.get(), kRepairableFeature3);
+            wuow.commit();
+        }
+
+        auto status = getFeatureTracker()->isCompatibleWithCurrentCode(opCtx.get());
+        ASSERT_EQ(ErrorCodes::CanRepairToDowngrade, status.code());
+        ASSERT_EQ(
+            "The data files use features not recognized by this version of mongod; the R feature"
+            " bits in positions [ 1, 2 ] aren't recognized by this version of mongod",
+            status.reason());
+    }
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp
index 3ff4a725501..687ed63b3f7 100644
--- a/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp
+++ b/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/storage/kv/kv_database_catalog_entry.h"
 
 #include "mongo/db/operation_context.h"
+#include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/kv/kv_collection_catalog_entry.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/kv_storage_engine.h"
@@ -162,9 +163,8 @@ void KVDatabaseCatalogEntry::appendExtraStats(OperationContext* opCtx,
                                               double scale) const {}
 
 Status KVDatabaseCatalogEntry::currentFilesCompatible(OperationContext* opCtx) const {
-    // TODO SERVER-23116: Delegate to the KVCatalog::FeatureTracker as to whether the data files are
-    // compatible or not.
-    return Status::OK();
+    // Delegate to the FeatureTracker as to whether the data files are compatible or not.
+    return _engine->getCatalog()->getFeatureTracker()->isCompatibleWithCurrentCode(opCtx);
 }
 
 void KVDatabaseCatalogEntry::getCollectionNamespaces(std::list<std::string>* out) const {