fix(acl): incompatibilities with acl load (dragonflydb#2867)

* ignore case when reading the acl line prefix user
* acl load returns OK before it kills the active connections

src/server/acl/acl_family.cc

@@ -236,21 +236,21 @@ void AclFamily::Save(CmdArgList args, ConnectionContext* cntx) {
   cntx->SendOk();
 }
 
-std::optional<facade::ErrorReply> AclFamily::LoadToRegistryFromFile(std::string_view full_path,
-                                                                    bool init) {
+GenericError AclFamily::LoadToRegistryFromFile(std::string_view full_path,
+                                               ConnectionContext* cntx) {
   auto is_file_read = io::ReadFileToString(full_path);
   if (!is_file_read) {
     auto error = absl::StrCat("Dragonfly could not load ACL file ", full_path, " with error ",
                               is_file_read.error().message());
 
     LOG(WARNING) << error;
-    return {ErrorReply(std::move(error))};
+    return {std::move(error)};
   }
 
   auto file_contents = std::move(is_file_read.value());
 
   if (file_contents.empty()) {
-    return {facade::ErrorReply("Empty file")};
+    return {"Empty file"};
   }
 
   std::vector<std::string> usernames;
@@ -259,7 +259,7 @@ std::optional<facade::ErrorReply> AclFamily::LoadToRegistryFromFile(std::string_
   if (!materialized) {
     std::string error = "Error materializing acl file";
     LOG(WARNING) << error;
-    return {ErrorReply(std::move(error))};
+    return {std::move(error)};
   }
 
   std::vector<User::UpdateRequest> requests;
@@ -269,15 +269,15 @@ std::optional<facade::ErrorReply> AclFamily::LoadToRegistryFromFile(std::string_
     if (std::holds_alternative<ErrorReply>(req)) {
       auto error = std::move(std::get<ErrorReply>(req));
       LOG(WARNING) << "Error while parsing aclfile: " << error.ToSv();
-      return {std::move(error)};
+      return {std::string(error.ToSv())};
     }
     requests.push_back(std::move(std::get<User::UpdateRequest>(req)));
   }
 
   auto registry_with_wlock = registry_->GetRegistryWithWriteLock();
   auto& registry = registry_with_wlock.registry;
-  // TODO(see what redis is doing here)
-  if (!init) {
+  if (cntx) {
+    cntx->SendOk();
     // Evict open connections for old users
     EvictOpenConnectionsOnAllProactorsWithRegistry(registry);
     registry.clear();
@@ -298,7 +298,7 @@ std::optional<facade::ErrorReply> AclFamily::LoadToRegistryFromFile(std::string_
 
 bool AclFamily::Load() {
   auto acl_file = absl::GetFlag(FLAGS_aclfile);
-  return !LoadToRegistryFromFile(acl_file, true).has_value();
+  return !LoadToRegistryFromFile(acl_file, nullptr);
 }
 
 void AclFamily::Load(CmdArgList args, ConnectionContext* cntx) {
@@ -308,14 +308,11 @@ void AclFamily::Load(CmdArgList args, ConnectionContext* cntx) {
     return;
   }
 
-  const auto is_successfull = LoadToRegistryFromFile(acl_file, !cntx);
+  const auto load_error = LoadToRegistryFromFile(acl_file, cntx);
 
-  if (is_successfull) {
-    cntx->SendError(absl::StrCat("Error loading: ", acl_file, " ", is_successfull->ToSv()));
-    return;
+  if (load_error) {
+    cntx->SendError(absl::StrCat("Error loading: ", acl_file, " ", load_error.Format()));
   }
-
-  cntx->SendOk();
 }
 
 void AclFamily::Log(CmdArgList args, ConnectionContext* cntx) {

src/server/acl/acl_family.h

@@ -59,7 +59,7 @@ class AclFamily final {
   void EvictOpenConnectionsOnAllProactorsWithRegistry(const UserRegistry::RegistryType& registry);
 
   // Helper function that loads the acl state of an acl file into the user registry
-  std::optional<facade::ErrorReply> LoadToRegistryFromFile(std::string_view full_path, bool init);
+  GenericError LoadToRegistryFromFile(std::string_view full_path, ConnectionContext* init);
 
   std::string RegistryToString() const;
 

src/server/acl/helpers.cc

@@ -206,7 +206,7 @@ MaterializedContents MaterializeFileContents(std::vector<std::string>* usernames
     if (command.empty())
       continue;
     std::vector<std::string_view> cmds = absl::StrSplit(command, ' ');
-    if (cmds[0] != "USER" || cmds.size() < 4) {
+    if (!absl::EqualsIgnoreCase(cmds[0], "USER") || cmds.size() < 4) {
       return {};
     }