fix(acl): preemptive TlsSocket::Shutdown in atomic section (#5525)

The issue is that we called TlsSocket::Shutdown while traversing the connections. Since Shutdown is preemptive, it violates the atomicity guarantees of the operation

---------

Signed-off-by: kostas <[email protected]>

Author: kostasrim

src/server/acl/acl_family.cc

@@ -64,6 +64,8 @@ void SendAclSecurityEvents(const AclLog::LogEntry& entry, facade::RedisReplyBuil
 
 string AclDbToString(size_t db);
 
+template <typename P>
+void TraverseEvictImpl(P predicate, facade::Listener* main_listener, util::ProactorPool* pool);
 }  // namespace
 
 AclFamily::AclFamily(UserRegistry* registry, util::ProactorPool* pool)
@@ -164,36 +166,17 @@ void AclFamily::SetUser(CmdArgList args, const CommandContext& cmd_cntx) {
 }
 
 void AclFamily::EvictOpenConnectionsOnAllProactors(const absl::flat_hash_set<string_view>& users) {
-  auto close_cb = [&users]([[maybe_unused]] size_t id, util::Connection* conn) {
-    CHECK(conn);
-    auto connection = static_cast<facade::Connection*>(conn);
-    auto ctx = static_cast<ConnectionContext*>(connection->cntx());
-    if (ctx && users.contains(ctx->authed_username)) {
-      connection->ShutdownSelfBlocking();
-    }
-  };
-
-  if (main_listener_) {
-    // TODO(kostas) fix this it might preempt and TraverseConnection shall be atomic
-    main_listener_->TraverseConnections(close_cb);
-  }
+  return TraverseEvictImpl([&](auto* ctx) { return ctx && users.contains(ctx->authed_username); },
+                           main_listener_, pool_);
 }
 
 void AclFamily::EvictOpenConnectionsOnAllProactorsWithRegistry(
     const UserRegistry::RegistryType& registry) {
-  auto close_cb = [&registry]([[maybe_unused]] size_t id, util::Connection* conn) {
-    CHECK(conn);
-    auto connection = static_cast<facade::Connection*>(conn);
-    auto ctx = static_cast<ConnectionContext*>(connection->cntx());
-    if (ctx && ctx->authed_username != "default" && registry.contains(ctx->authed_username)) {
-      connection->ShutdownSelfBlocking();
-    }
-  };
-
-  if (main_listener_) {
-    // TODO(kostas) fix this it might preempt and TraverseConnection shall be atomic
-    main_listener_->TraverseConnections(close_cb);
-  }
+  return TraverseEvictImpl(
+      [&](auto* ctx) {
+        return ctx && ctx->authed_username != "default" && registry.contains(ctx->authed_username);
+      },
+      main_listener_, pool_);
 }
 
 void AclFamily::DelUser(CmdArgList args, const CommandContext& cmd_cntx) {
@@ -933,6 +916,34 @@ std::string AclDbToString(size_t db) {
   return std::numeric_limits<size_t>::max() == db ? "all" : absl::StrCat(db);
 }
 
+// Fetches the connections that predicate P evaluates to true and shuts them
+// down gracefully.
+template <typename P>
+void TraverseEvictImpl(P predicate, facade::Listener* main_listener, util::ProactorPool* pool) {
+  auto close_cb = [&](unsigned idx, util::ProactorBase* p) {
+    std::vector<facade::Connection::WeakRef> connections;
+    auto traverse_cb = [&](unsigned id, util::Connection* conn) {
+      auto connection = static_cast<facade::Connection*>(conn);
+      auto ctx = connection->cntx();
+      if (predicate(ctx)) {
+        connections.push_back(connection->Borrow());
+      }
+    };
+
+    main_listener->TraverseConnectionsOnThread(traverse_cb, UINT32_MAX, nullptr);
+
+    for (auto& tcon : connections) {
+      facade::Connection* conn = tcon.Get();
+      if (conn && conn->socket()->proactor()->GetPoolIndex() == p->GetPoolIndex()) {
+        // preemptive for TlsSocket
+        conn->ShutdownSelfBlocking();
+      }
+    }
+  };
+
+  pool->AwaitFiberOnAll(close_cb);
+}
+
 }  // namespace
 
 std::string AclFamily::AclCatAndCommandToString(const User::CategoryChanges& cat,