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.

(cherry picked from commit 77d99b27649d4f81302fde2cb7cd7f4967b9646c)

This commit differs slightly from
77d99b27649d4f81302fde2cb7cd7f4967b9646c in order to handle the
StatusWith<RecordId> returned by RecordStore::updateRecord().

6 files changed, 932 insertions, 3 deletions

diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index de8e5d25d64..6d7fbd2bcc2 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/namespace_string',
@@ -80,6 +81,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 6a6d1a8e437..ecdfcc2541e 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,193 @@ 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 rid = _catalog->_rs->updateRecord(
+            opCtx, _rid, obj.objdata(), obj.objsize(), enforceQuota, notifier);
+        fassert(40114, rid.getStatus());
+        invariant(_rid == rid.getValue());
+    }
+}
+
 KVCatalog::KVCatalog(RecordStore* rs,
                      bool isRsThreadSafe,
                      bool directoryPerDb,
@@ -132,12 +345,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 da9cae5944e..2c368bdfafe 100644
--- a/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp
+++ b/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp
@@ -31,6 +31,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"
@@ -160,9 +161,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 {