Implementation of operational limit check based on current indexes an…

src/schema_manager.cc

@@ -69,6 +69,11 @@ vmsdk::config::Number &GetMaxIndexes() {
   return dynamic_cast<vmsdk::config::Number &>(*max_indexes);
 }
 
+// register callback for max index creation
+void MaxIndexCreationCallback(uint32_t current_indexes) {
+  vmsdk::config::SetGetMaxIndexesCallback(
+      GetMaxIndexes, &current_indexes);  // register callback
+}
 /// Register the "--backfill-batch-size" flag. Controls the max number of
 /// indexes we can have.
 static auto backfill_batch_size =
@@ -84,6 +89,10 @@ vmsdk::config::Number &GetBackfillBatchSize() {
   return dynamic_cast<vmsdk::config::Number &>(*backfill_batch_size);
 }
 
+uint64_t GetMaxIndexesOptionUpdate() {
+  return vmsdk::config::g_updatedmax_index;
+}
+
 }  // namespace options
 
 // Randomly generated 32 bit key for fingerprinting the metadata.
@@ -212,7 +221,8 @@ absl::StatusOr<coordinator::IndexFingerprintVersion>
 SchemaManager::CreateIndexSchema(
     ValkeyModuleCtx *ctx, const data_model::IndexSchema &index_schema_proto) {
   const auto max_indexes = options::GetMaxIndexes().GetValue();
-
+  uint32_t current_indexes = SchemaManager::Instance().GetNumberOfIndexSchemas() + 1;
+  options::MaxIndexCreationCallback(current_indexes); 
   VMSDK_RETURN_IF_ERROR(vmsdk::VerifyRange(
       SchemaManager::Instance().GetNumberOfIndexSchemas() + 1, std::nullopt,
       max_indexes))
@@ -610,7 +620,14 @@ absl::Status SchemaManager::LoadIndex(
                          _ << "Failed to load index schema from RDB!");
   uint32_t db_num = index_schema->GetDBNum();
   const std::string &name = index_schema->GetName();
-
+  uint32_t numIndex = this->GetNumberOfIndexSchemas();
+  uint32_t updatedindex = options::GetMaxIndexesOptionUpdate();
+  if (updatedindex != 0 && numIndex >= updatedindex) {
+         return absl::InternalError(
+                 absl::StrFormat("Unable to load indexes for max index %d as current "
+                        "indexes exceeds the limit %d",
+                        updatedindex, numIndex));
+  }
   // Select the DB number in the context for subsequent usage.
   if (ValkeyModule_SelectDb(ctx, db_num) != VALKEYMODULE_OK) {
     return absl::InternalError(

src/schema_manager.h

@@ -40,7 +40,8 @@ namespace options {
 
 /// Return the maximum number of indexes allowed to create.
 vmsdk::config::Number &GetMaxIndexes();
-
+//callback for config update
+void MaxIndexCreationCallback(uint32_t);
 }  // namespace options
 class SchemaManager {
  public:

vmsdk/src/module_config.cc

@@ -22,6 +22,14 @@ namespace config {
 /// Valkey to load the configurations first time when the module loaded. Once
 /// this is done, we set it back to false. If the user passes "--debug-mode yes"
 /// we will change it back to "true".
+static GetMaxIndexesCallback g_getMaxIndexes = nullptr;
+static uint32_t g_current_index = 0;
+void SetGetMaxIndexesCallback(GetMaxIndexesCallback cb,
+                              uint32_t *current_index) {
+  g_getMaxIndexes = cb;
+  if (current_index != nullptr) g_current_index = *current_index;
+}
+
 static auto debug_mode = BooleanBuilder(kDebugMode, true).Hidden().Build();
 
 bool IsDebugModeEnabled() { return debug_mode->GetValue(); }
@@ -41,6 +49,21 @@ template <typename T>
 static int OnSetConfig(const char *config_name, T value, void *priv_data,
                        ValkeyModuleString **err) {
   auto entry = static_cast<ConfigBase<T> *>(priv_data);
+  // check if max_indexes operational limit is higher than current set
+  if (std::strcmp(config_name, "max-indexes") == 0) {
+      if (g_getMaxIndexes != nullptr) {
+          if ((g_current_index != 0 && g_current_index <= g_getMaxIndexes().GetValue()) &&
+              static_cast<uint32_t>(value) <= g_current_index) {
+                  ValkeyModule_CreateStringPrintf(
+                  nullptr, "%s - %d",
+            "Command rejected, configured operational limit is lower than the "
+            "configured index schema",
+            static_cast<uint32_t>(value));
+        return VALKEYMODULE_ERR;
+
+      }
+    }
+  }
   CHECK(entry) << "null private data for configuration Number entry.";
   auto res = entry->SetValue(value);  // Calls "Validate" internally
   if (!res.ok()) {
@@ -167,6 +190,14 @@ absl::Status ModuleConfigManager::UpdateConfigFromKeyVal(
   // update the configuration entry
   VMSDK_LOG(NOTICE, ctx) << "Parsed command line argument: " << key << " -> "
                          << value;
+   // load max index passed from command line
+  absl::string_view sv1 = "max-indexes";
+  if (where->first == sv1) {
+    if (!absl::SimpleAtoi(value, &g_updatedmax_index)) {
+      return absl::InvalidArgumentError(
+          absl::StrFormat("Failed to convert '%s' into a number", value));
+    }
+  }
   VMSDK_RETURN_IF_ERROR(where->second->FromString(value));
   return absl::OkStatus();
 }

vmsdk/src/module_config.h

@@ -38,7 +38,8 @@ enum Flags {
 /// Return true if debug mode is enabled. "search.debug-mode == yes"
 constexpr absl::string_view kDebugMode{"debug-mode"};
 bool IsDebugModeEnabled();
-
+//config updated persist
+int64_t g_updatedmax_index;
 /// Support Valkey configuration entries in a one-liner.
 ///
 /// Example usage:
@@ -322,6 +323,9 @@ class String : public ConfigBase<std::string> {
   mutable UniqueValkeyString cached_string_;
   FRIEND_TEST(Builder, ConfigBuilder);
 };
+// callbacks for internal module changes for configs #376
+using GetMaxIndexesCallback = Number &(*)();
+void SetGetMaxIndexesCallback(GetMaxIndexesCallback cb, uint32_t *);
 
 template <typename ValkeyT>
 class ConfigBuilder {