fix(config): avoid rewriting the config file if it's unnessary

The server start will rewrite the config file to presist namespace/token pairs
if the namespace replication is enabled. But it's unnecessary if the
namespace replication is diabled or no tokens were loaded from the
configuration file, because the purpose of this rewrite is to remove
tokens from the config file and write to the Database.

src/config/config.cc

@@ -885,11 +885,20 @@ Status Config::Set(Server *srv, std::string key, const std::string &value) {
     if (!s.IsOK()) return s.Prefixed("invalid value");
   }
 
+  auto origin_value = field->ToString();
   auto s = field->Set(value);
   if (!s.IsOK()) return s.Prefixed("failed to set new value");
 
   if (field->callback) {
-    return field->callback(srv, key, value);
+    s = field->callback(srv, key, value);
+    if (!s.IsOK()) {
+      // rollback the value if the callback failed
+      auto set_status = field->Set(origin_value);
+      if (!set_status.IsOK()) {
+        return set_status.Prefixed("failed to rollback the value");
+      }
+    }
+    return s;
   }
 
   return Status::OK();

src/server/namespace.cc

@@ -51,36 +51,47 @@ bool Namespace::IsAllowModify() const {
   return config->HasConfigFile() || config->repl_namespace_enabled;
 }
 
+Status Namespace::loadFromDB(std::map<std::string, std::string>* db_tokens) const {
+  std::string value;
+  auto s = storage_->Get(rocksdb::ReadOptions(), cf_, kNamespaceDBKey, &value);
+  if (!s.ok()) {
+    if (s.IsNotFound()) return Status::OK();
+    return {Status::NotOK, s.ToString()};
+  }
+
+  jsoncons::json j = jsoncons::json::parse(value);
+  for (const auto& iter : j.object_range()) {
+    db_tokens->insert({iter.key(), iter.value().as_string()});
+  }
+  return Status::OK();
+}
+
 Status Namespace::LoadAndRewrite() {
   auto config = storage_->GetConfig();
   // Namespace is NOT allowed in the cluster mode, so we don't need to rewrite here.
   if (config->cluster_enabled) return Status::OK();
 
-  // Load from the configuration file first
-  tokens_ = config->load_tokens;
+  std::map<std::string, std::string> db_tokens;
+  auto s = loadFromDB(&db_tokens);
+  if (!s.IsOK()) return s;
 
-  // We would like to load namespaces from db even if repl_namespace_enabled is false,
-  // this can avoid missing some namespaces when turn on/off repl_namespace_enabled.
-  std::string value;
-  auto s = storage_->Get(rocksdb::ReadOptions(), cf_, kNamespaceDBKey, &value);
-  if (!s.ok() && !s.IsNotFound()) {
-    return {Status::NotOK, s.ToString()};
+  if (!db_tokens.empty() && !config->repl_namespace_enabled) {
+    return {Status::NotOK, "cannot switch off repl_namespace_enabled when namespaces exist in db"};
   }
-  if (s.ok()) {
-    // The namespace db key is existed, so it doesn't allow to switch off repl_namespace_enabled
-    if (!config->repl_namespace_enabled) {
-      return {Status::NotOK, "cannot switch off repl_namespace_enabled when namespaces exist in db"};
-    }
 
-    jsoncons::json j = jsoncons::json::parse(value);
-    for (const auto& iter : j.object_range()) {
-      if (tokens_.find(iter.key()) == tokens_.end()) {
-        // merge the namespace from db
-        tokens_[iter.key()] = iter.value().as<std::string>();
-      }
+  // Load from the configuration file first
+  tokens_ = config->load_tokens;
+  // Merge the tokens from the database if the token is not in the configuration file
+  for (const auto& iter : db_tokens) {
+    if (tokens_.find(iter.first) == tokens_.end()) {
+      tokens_[iter.first] = iter.second;
     }
   }
 
+  // The following rewrite is to remove namespace/token pairs from the configuration if the namespace replication
+  // is enabled. So we don't need to do that if no tokens are loaded or the namespace replication is disabled.
+  if (config->load_tokens.empty() || !config->repl_namespace_enabled) return Status::OK();
+
   return Rewrite();
 }
 

src/server/namespace.h

@@ -47,4 +47,6 @@ class Namespace {
   engine::Storage *storage_;
   rocksdb::ColumnFamilyHandle *cf_ = nullptr;
   std::map<std::string, std::string> tokens_;
+
+  Status loadFromDB(std::map<std::string, std::string> *db_tokens) const;
 };

tests/gocase/unit/namespace/namespace_test.go

@@ -238,7 +238,16 @@ func TestNamespaceReplicate(t *testing.T) {
 	})
 
 	t.Run("Turn off namespace replication is not allowed", func(t *testing.T) {
+		r := masterRdb.Do(ctx, "NAMESPACE", "ADD", "test-ns", "ns-token")
+		require.NoError(t, r.Err())
+		require.Equal(t, "OK", r.Val())
 		util.ErrorRegexp(t, masterRdb.ConfigSet(ctx, "repl-namespace-enabled", "no").Err(), ".*cannot switch off repl_namespace_enabled when namespaces exist in db.*")
+
+		// it should be allowed after deleting all namespaces
+		r = masterRdb.Do(ctx, "NAMESPACE", "DEL", "test-ns")
+		require.NoError(t, r.Err())
+		require.Equal(t, "OK", r.Val())
+		require.NoError(t, masterRdb.ConfigSet(ctx, "repl-namespace-enabled", "no").Err())
 	})
 }