MB-61655: Add compat_version parameter

This parameter will hold the bucket configuration compatiblity version.

It will be initialised and updated by ns_server, to the cluster compat
version.

Defining it as a "normal" parameter in configuration.json, has the
benefit that the parameter is automatically reported in STAT(config),
which ns_server monitor for configuration differences between it's
expectation and actual values set in the bucket.

Change-Id: I339bd5f820358bb5825dd2a2e9e4e0bc2eb95f07
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/234660
Tested-by: Build Bot <[email protected]>
Reviewed-by: Faizan Alam <[email protected]>

Author: veselink1

engines/ep/CMakeLists.txt

@@ -251,6 +251,7 @@ ADD_LIBRARY(ep STATIC
             src/collections/vbucket_manifest_handles.cc
             src/collections/vbucket_manifest_scope_entry.cc
             src/configuration.cc
+            src/configuration_types.cc
             src/conflict_resolution.cc
             src/conn_store.cc
             src/connhandler.cc

engines/ep/configuration.json

@@ -62,6 +62,12 @@
  */
 {
     "params": {
+        "compat_version": {
+            "descr": "Compatibility version applied to the bucket, in the format \"major.minor\"",
+            "default": "",
+            "dynamic": true,
+            "type": "std::string"
+        },
         "allow_sanitize_value_in_deletion" : {
             "default" : "true",
             "descr": "Let EPE delete/prepare/del_with_meta prune any invalid body in the payload instead of failing",

engines/ep/src/configuration.cc

@@ -126,6 +126,7 @@ void ValueChangedValidator::validateString(std::string_view key, const char*) {
 Configuration::Configuration(bool isServerless, bool isDevAssertEnabled)
     : isServerless(isServerless), isDevAssertEnabled(isDevAssertEnabled) {
     initialize();
+    initializeCompatVersion();
     initialized = true;
 }
 
@@ -339,6 +340,7 @@ Configuration::Configuration(const Configuration& other)
     : isServerless(other.isServerless),
       isDevAssertEnabled(other.isDevAssertEnabled) {
     initialize();
+    initializeCompatVersion();
     initialized = true;
     for (const auto& [key, value] : other.attributes) {
         (void)attributes[key]->setValue(value->getValue());
@@ -706,6 +708,57 @@ template void Configuration::addValueChangedFunc(std::string_view,
 template void Configuration::addValueChangedFunc(
         std::string_view, std::function<void(std::string_view)>);
 
+class FeatureVersionValidator : public ValueChangedValidator {
+public:
+    void validateString(std::string_view key, const char* value) override {
+        if (value[0] == '\0') {
+            // Allow the default value to be set.
+            return;
+        }
+        cb::config::FeatureVersion::parse(value);
+    }
+};
+
+void Configuration::initializeCompatVersion() {
+    static constexpr auto compatVersionKey = "compat_version";
+    auto itr = attributes.find(compatVersionKey);
+    if (itr == attributes.end()) {
+        throw std::invalid_argument(fmt::format(
+                "Configuration: No such config key '{}'", compatVersionKey));
+    }
+
+    if (itr->second->validator) {
+        throw std::logic_error(
+                fmt::format("Configuration: Validator already set for key '{}'",
+                            compatVersionKey));
+    }
+    // We need a special validator for the compat version, since it's not a
+    // supported type by the configuration.
+    itr->second->validator = std::make_unique<FeatureVersionValidator>();
+
+    // Use a listener to update the compat version atomic when it changes.
+    // We will read this atomic in the getter.
+    std::function<void(std::string_view)> callback =
+            [this](const std::string_view str) {
+                auto version = str.empty()
+                                       ? cb::config::FeatureVersion::max()
+                                       : cb::config::FeatureVersion::parse(str);
+                processCompatVersionChange(version);
+            };
+    itr->second->addChangeListener(
+            std::make_unique<ValueChangedCallback<std::string_view>>(
+                    std::move(callback)));
+}
+
+void Configuration::processCompatVersionChange(
+        cb::config::FeatureVersion version) {
+    compatVersion.store(version, std::memory_order_release);
+}
+
+cb::config::FeatureVersion Configuration::getEffectiveCompatVersion() const {
+    return compatVersion.load(std::memory_order_acquire);
+}
+
 // Explicit instantiations for addParameter for supported types.
 #define INSTANTIATE_TEMPLATES(T)                                \
     template void Configuration::addParameter(std::string_view, \

engines/ep/src/configuration.h

@@ -292,6 +292,13 @@ class Configuration : public ConfigurationIface {
     ParameterValidationMap validateParameters(
             const ParameterMap& parameters) const override;
 
+    /**
+     * Get the compatibility version for the configuration.
+     * If the compatibility version is not set, the maximum feature version is
+     * returned.
+     */
+    cb::config::FeatureVersion getEffectiveCompatVersion() const;
+
     const bool isServerless;
 
     const bool isDevAssertEnabled = false;
@@ -306,6 +313,12 @@ class Configuration : public ConfigurationIface {
      */
     Configuration(const Configuration& other);
 
+    /// The compatibility version for the configuration.
+    void initializeCompatVersion();
+
+    /// Process a change to the compatibility version.
+    void processCompatVersionChange(cb::config::FeatureVersion version);
+
     std::pair<ParameterValidationMap, bool> setParametersInternal(
             const ParameterMap& parameters);
 
@@ -403,6 +416,12 @@ class Configuration : public ConfigurationIface {
      */
     cb::RelaxedAtomic<bool> initialized{false};
 
+    /**
+     * The compatibility version for the configuration.
+     */
+    std::atomic<cb::config::FeatureVersion> compatVersion{
+            cb::config::FeatureVersion::max()};
+
     void initialize();
 
     /**

engines/ep/src/configuration_types.cc

@@ -0,0 +1,72 @@
+/*
+ *     Copyright 2025-Present Couchbase, Inc.
+ *
+ *   Use of this software is governed by the Business Source License included
+ *   in the file licenses/BSL-Couchbase.txt.  As of the Change Date specified
+ *   in that file, in accordance with the Business Source License, use of this
+ *   software will be governed by the Apache License, Version 2.0, included in
+ *   the file licenses/APL2.txt.
+ */
+
+#include "configuration_types.h"
+
+#include <fmt/format.h>
+#include <charconv>
+#include <optional>
+
+namespace cb::config {
+
+FeatureVersion::FeatureVersion(uint8_t major, uint8_t minor)
+    : major_version(major), minor_version(minor) {
+}
+
+static std::optional<uint8_t> parseUint8(std::string_view num) {
+    uint8_t number = 0;
+    auto [ptr, ec] =
+            std::from_chars(num.data(), num.data() + num.size(), number);
+    if (ec != std::errc()) {
+        return std::nullopt;
+    }
+    return number;
+}
+
+static std::pair<uint8_t, uint8_t> parseVersionComponents(
+        std::string_view version) {
+    std::optional<uint8_t> major;
+    std::optional<uint8_t> minor;
+    const auto sep = version.find('.');
+    if (sep != std::string_view::npos) {
+        major = parseUint8(version.substr(0, sep));
+        minor = parseUint8(version.substr(sep + 1));
+    }
+    if (!major || !minor || sep + 2 != version.size()) {
+        throw std::invalid_argument("Invalid FeatureVersion: " +
+                                    std::string(version));
+    }
+    return {*major, *minor};
+}
+
+FeatureVersion FeatureVersion::max() {
+    return {std::numeric_limits<uint8_t>::max(),
+            std::numeric_limits<uint8_t>::max()};
+}
+
+FeatureVersion FeatureVersion::parse(std::string_view version) {
+    auto [major, minor] = parseVersionComponents(version);
+    return {major, minor};
+}
+
+std::strong_ordering FeatureVersion::operator<=>(
+        const FeatureVersion&) const noexcept = default;
+
+bool FeatureVersion::operator==(const FeatureVersion& other) const noexcept =
+        default;
+
+std::string format_as(const FeatureVersion& version) {
+    if (version == FeatureVersion::max()) {
+        return "<max>";
+    }
+    return fmt::format("{}.{}", version.major_version, version.minor_version);
+}
+
+} // namespace cb::config

engines/ep/src/configuration_types.h

@@ -10,6 +10,7 @@
 
 #pragma once
 
+#include <atomic>
 #include <string>
 #include <string_view>
 #include <type_traits>
@@ -40,4 +41,44 @@ std::string to_string(T val) {
     return std::string(cb::config::format_as(val));
 }
 
+/**
+ * A feature version is a major and minor version.
+ *
+ * Supports comparison and ordering.
+ */
+struct FeatureVersion {
+    /**
+     * The maximum feature version.
+     */
+    static FeatureVersion max();
+    /**
+     * Parses a feature version from a string in the format "major.minor".
+     *
+     * @param version the input to parse
+     * @return the parsed feature version
+     * @throws std::invalid_argument
+     */
+    static FeatureVersion parse(std::string_view version);
+
+    /// Construct a feature version.
+    FeatureVersion(uint8_t major, uint8_t minor);
+
+    FeatureVersion(const FeatureVersion&) = default;
+    FeatureVersion& operator=(const FeatureVersion&) = default;
+
+    /// Compare two feature versions.
+    std::strong_ordering operator<=>(const FeatureVersion&) const noexcept;
+    bool operator==(const FeatureVersion&) const noexcept;
+
+    /// The major version
+    uint8_t major_version{};
+    /// The minor version
+    uint8_t minor_version{};
+};
+
+static_assert(std::atomic<FeatureVersion>::is_always_lock_free,
+              "cb::config::FeatureVersion atomic is not lock-free");
+
+std::string format_as(const FeatureVersion& version);
+
 } // namespace cb::config

engines/ep/src/ep_parameters.cc

@@ -182,12 +182,18 @@ static const std::unordered_set<std::string_view> vbucketParamSet{
         "dcp_hlc_invalid_strategy",
 };
 
+/// Parameters which are exclusive to the config category.
+static const std::unordered_set<std::string_view> extraConfigParamSet{
+        "compat_version",
+};
+
 static const std::unordered_set<std::string_view> allParamSet = []() {
     std::unordered_set<std::string_view> ret;
     ret.insert(checkpointParamSet.begin(), checkpointParamSet.end());
     ret.insert(flushParamSet.begin(), flushParamSet.end());
     ret.insert(dcpParamSet.begin(), dcpParamSet.end());
     ret.insert(vbucketParamSet.begin(), vbucketParamSet.end());
+    ret.insert(extraConfigParamSet.begin(), extraConfigParamSet.end());
     return ret;
 }();
 

engines/ep/tests/ep_testsuite.cc

@@ -6279,6 +6279,7 @@ static enum test_result test_mb19687_fixed(EngineIface* h) {
               "ep_bucket_type",
               "ep_bucket_quota_change_task_poll_interval",
               "ep_cache_size",
+              "ep_compat_version",
               "ep_chk_expel_enabled",
               "ep_checkpoint_destruction_tasks",
               "ep_checkpoint_memory_ratio",
@@ -6542,6 +6543,7 @@ static enum test_result test_mb19687_fixed(EngineIface* h) {
               "ep_bucket_type",
               "ep_bucket_quota_change_task_poll_interval",
               "ep_cache_size",
+              "ep_compat_version",
               "ep_chk_expel_enabled",
               "ep_chk_persistence_remains",
               "ep_checkpoint_destruction_tasks",

engines/ep/tests/module_tests/configuration_test.cc

@@ -752,3 +752,41 @@ TEST(ConfigurationTest, ValidateNonDynamic) {
     EXPECT_EQ(validation["param"]["value"], 1);
     EXPECT_EQ(validation["param"]["requiresRestart"], true);
 }
+
+TEST(ConfigurationTest, CompatVersionComparison) {
+    using cb::config::FeatureVersion;
+    EXPECT_EQ(FeatureVersion(8, 0), FeatureVersion(8, 0));
+    EXPECT_LT(FeatureVersion(8, 0), FeatureVersion(8, 1));
+    EXPECT_GE(FeatureVersion(8, 0), FeatureVersion(7, 6));
+}
+
+TEST(ConfigurationTest, CompatVersionDefault) {
+    using cb::config::FeatureVersion;
+    ConfigurationShim configuration;
+    auto version = configuration.getEffectiveCompatVersion();
+    EXPECT_EQ(version, FeatureVersion::max());
+}
+
+TEST(ConfigurationTest, CompatVersionSet) {
+    using cb::config::FeatureVersion;
+    ConfigurationShim configuration;
+    configuration.setParameter("compat_version", "8.0");
+    auto version = configuration.getEffectiveCompatVersion();
+    EXPECT_EQ(version, FeatureVersion(8, 0));
+
+    configuration.setParameter("compat_version", "");
+    auto versionMax = configuration.getEffectiveCompatVersion();
+    EXPECT_EQ(versionMax, FeatureVersion::max());
+}
+
+TEST(ConfigurationTest, CompatVersionSetInvalid) {
+    using cb::config::FeatureVersion;
+    ConfigurationShim configuration;
+    EXPECT_THROW(configuration.setParameter("compat_version", "asd"),
+                 std::invalid_argument);
+    EXPECT_THROW(configuration.setParameter("compat_version", "8"),
+                 std::invalid_argument);
+    EXPECT_THROW(configuration.setParameter("compat_version", "8.0.0"),
+                 std::invalid_argument);
+    EXPECT_EQ(configuration.getEffectiveCompatVersion(), FeatureVersion::max());
+}

engines/ep/tests/module_tests/evp_engine_test.cc

@@ -332,12 +332,28 @@ TEST_P(EPEngineParamTest, DynamicConfigValuesModifiable) {
                 handled.emplace_back("setVBucketParam");
             }
 
+            bool canSetViaConfigurationParameter =
+                    engine->setConfigurationParameter(
+                            std::string(key), value, msg) ==
+                    cb::engine_errc::success;
+            if (!canSetViaConfigurationParameter) {
+                // Any parameter settable via the above param methods
+                // should be settable via the setConfigurationParameter method
+                // too.
+                ADD_FAILURE()
+                        << "Dynamic config key \"" << key
+                        << "\" should be settable via "
+                           "setConfigurationParameter() - actually settable "
+                           "via: ["
+                        << boost::algorithm::join(handled, ", ") << "]";
+            }
+
             if (deprecated.contains(key)) {
                 EXPECT_EQ(0, handled.size())
                         << "Dynamic config key \"" << key
                         << "\" can no longer be set - actually settable via: ["
                         << boost::algorithm::join(handled, ", ") << "]";
-            } else if (handled.empty()) {
+            } else if (handled.empty() && !canSetViaConfigurationParameter) {
                 ADD_FAILURE() << "Dynamic config key \"" << key
                               << "\" cannot be set via any of the set...Param "
                                  "methods.";
@@ -348,20 +364,6 @@ TEST_P(EPEngineParamTest, DynamicConfigValuesModifiable) {
                            "set...Param() method - actually settable via: ["
                         << boost::algorithm::join(handled, ", ") << "]";
             }
-
-            if (engine->setConfigurationParameter(
-                        std::string(key), value, msg) !=
-                cb::engine_errc::success) {
-                // Any parameter settable via the above param methods
-                // should be settable via the setConfigurationParameter method
-                // too.
-                ADD_FAILURE()
-                        << "Dynamic config key \"" << key
-                        << "\" should be settable via "
-                           "setConfigurationParameter() - actually settable "
-                           "via: ["
-                        << boost::algorithm::join(handled, ", ") << "]";
-            }
         }
     });
 }