chore: pull latest helio (#4446)

This also pulls the latest abseil library 20240722.0

helio

@@ -1 +1 @@
-Subproject commit 51f9c8b913b44cff65c7bebe0da19a5257f5070b
+Subproject commit 90bc3cc6aabbffc8b274dc0f7801695d68658529

src/core/interpreter.cc

@@ -923,7 +923,7 @@ std::optional<absl::FixedArray<std::string_view, 4>> Interpreter::PrepareArgs()
     switch (lua_type(lua_, idx)) {
       case LUA_TNUMBER:
         if (lua_isinteger(lua_, idx)) {
-          blob_len += absl::AlphaNum{lua_tointeger(lua_, idx)}.size();
+          blob_len += absl::AlphaNum(lua_tointeger(lua_, idx)).size();
         } else {
           int fmt_len = absl::SNPrintF(tmpbuf, sizeof(tmpbuf), "%.17g", lua_tonumber(lua_, idx));
           CHECK_GT(fmt_len, 0);

src/core/search/base.h

@@ -112,11 +112,17 @@ std::optional<double> ParseNumericField(std::string_view value);
    suppress this false warning, we temporarily disable it around this block of code using GCC
    diagnostic directives. */
 template <typename InlinedVector> std::optional<InlinedVector> EmptyAccessResult() {
+#if !defined(__clang__)
   // GCC 13.1 throws spurious warnings around this code.
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wmaybe-uninitialized"
+#endif
+
   return InlinedVector{};
+
+#if !defined(__clang__)
 #pragma GCC diagnostic pop
+#endif
 }
 
 }  // namespace dfly::search

src/facade/dragonfly_listener.cc

@@ -401,8 +401,11 @@ DispatchTracker::DispatchTracker(absl::Span<facade::Listener* const> listeners,
 }
 
 void DispatchTracker::TrackOnThread() {
-  for (auto* listener : listeners_)
+  for (auto* listener : listeners_) {
-    listener->TraverseConnectionsOnThread(absl::bind_front(&DispatchTracker::Handle, this));
+    listener->TraverseConnectionsOnThread(
+        [this](unsigned thread_index, util::Connection* conn) { Handle(thread_index, conn); },
+        UINT32_MAX, nullptr);
+  }
 }
 
 bool DispatchTracker::Wait(absl::Duration duration) {

src/facade/reply_builder.cc

@@ -30,13 +30,7 @@ namespace facade {
 
 namespace {
 
-inline iovec constexpr IoVec(std::string_view s) {
-  iovec r{const_cast<char*>(s.data()), s.size()};
-  return r;
-}
-
 constexpr char kCRLF[] = "\r\n";
-constexpr char kErrPref[] = "-ERR ";
 constexpr char kSimplePref[] = "+";
 constexpr char kLengthPrefix[] = "$";
 constexpr char kDoublePref[] = ",";

src/server/acl/user.cc

@@ -90,7 +90,12 @@ void User::Update(UpdateRequest&& req, const CategoryToIdxStore& cat_to_id,
 void User::SetPasswordHash(std::string_view password, bool is_hashed) {
   nopass_ = false;
   if (is_hashed) {
-    password_hashes_.insert(absl::HexStringToBytes(password));
+    std::string binary;
+    if (absl::HexStringToBytes(password, &binary)) {
+      password_hashes_.insert(binary);
+    } else {
+      LOG(ERROR) << "Invalid password hash: " << password;
+    }
     return;
   }
   password_hashes_.insert(StringSHA256(password));

src/server/cluster/outgoing_slot_migration.cc

@@ -261,7 +261,7 @@ void OutgoingMigration::SyncFb() {
       continue;
     }
 
-    OnAllShards([this](auto& migration) { migration->RunSync(); });
+    OnAllShards([](auto& migration) { migration->RunSync(); });
 
     if (cntx_.GetError()) {
       continue;

src/server/journal/journal_test.cc

@@ -35,7 +35,7 @@ template <> string ConCat(const CmdArgList& list) {
 struct EntryPayloadVisitor {
   void operator()(const Entry::Payload& p) {
     out->append(p.cmd).append(" ");
-    *out += visit([this](const auto& args) { return ConCat(args); }, p.args);
+    *out += visit([](const auto& args) { return ConCat(args); }, p.args);
   }
 
   string* out;

src/server/main_service.cc

@@ -1486,7 +1486,7 @@ void Service::DispatchMC(const MemcacheParser::Command& cmd, std::string_view va
                          MCReplyBuilder* mc_builder, facade::ConnectionContext* cntx) {
   absl::InlinedVector<MutableSlice, 8> args;
   char cmd_name[16];
-  char ttl[16];
+  char ttl[absl::numbers_internal::kFastToBufferSize];
   char store_opt[32] = {0};
   char ttl_op[] = "EXAT";
 

src/server/replica.cc

@@ -645,7 +645,7 @@ error_code Replica::ConsumeRedisStream() {
     }
 
     if (!LastResponseArgs().empty()) {
-      string_view cmd = absl::CHexEscape(ToSV(LastResponseArgs()[0].GetBuf()));
+      string cmd = absl::CHexEscape(ToSV(LastResponseArgs()[0].GetBuf()));
 
       // Valkey and Redis may send MULTI and EXEC as part of their replication commands.
       // Dragonfly disallows some commands, such as SELECT, inside of MULTI/EXEC, so here we simply

src/server/server_family.cc

@@ -1773,8 +1773,9 @@ void ServerFamily::CancelBlockingOnThread(std::function<OpStatus(ArgSlice)> stat
     }
   };
 
-  for (auto* listener : listeners_)
+  for (auto* listener : listeners_) {
-    listener->TraverseConnectionsOnThread(cb);
+    listener->TraverseConnectionsOnThread(cb, UINT32_MAX, nullptr);
+  }
 }
 
 string GetPassword() {

src/server/test_utils.cc

@@ -57,8 +57,8 @@ static vector<string> SplitLines(const std::string& src) {
   return res;
 }
 
-TestConnection::TestConnection(Protocol protocol, io::StringSink* sink)
+TestConnection::TestConnection(Protocol protocol)
-    : facade::Connection(protocol, nullptr, nullptr, nullptr), sink_(sink) {
+    : facade::Connection(protocol, nullptr, nullptr, nullptr) {
   cc_.reset(new dfly::ConnectionContext(this, {}));
   cc_->skip_acl_validation = true;
   SetSocket(ProactorBase::me()->CreateSocket());
@@ -141,7 +141,7 @@ class BaseFamilyTest::TestConnWrapper {
 };
 
 BaseFamilyTest::TestConnWrapper::TestConnWrapper(Protocol proto)
-    : dummy_conn_(new TestConnection(proto, &sink_)) {
+    : dummy_conn_(new TestConnection(proto)) {
   switch (proto) {
     case Protocol::REDIS:
       builder_.reset(new RedisReplyBuilder{&sink_});

src/server/test_utils.h

@@ -25,7 +25,7 @@ void TEST_InvalidateLockTagOptions();
 
 class TestConnection : public facade::Connection {
  public:
-  TestConnection(Protocol protocol, io::StringSink* sink);
+  explicit TestConnection(Protocol protocol);
   std::string RemoteEndpointStr() const override;
 
   void SendPubMessageAsync(PubMessage pmsg) final;
@@ -44,7 +44,6 @@ class TestConnection : public facade::Connection {
   std::vector<InvalidationMessage> invalidate_messages;
 
  private:
-  io::StringSink* sink_;
   bool is_privileged_ = false;
 };
 

src/server/zset_family.cc

@@ -1553,7 +1553,7 @@ OpResult<ScoredArray> ZPopMinMaxInternal(std::string_view key, FilterShards shou
   }
   auto cb = [&](Transaction* t, EngineShard* shard) {
     if (!key_shard.has_value() || *key_shard == shard->shard_id()) {
-      result = std::move(OpPopCount(range_spec, t->GetOpArgs(shard), key));
+      result = OpPopCount(range_spec, t->GetOpArgs(shard), key);
     }
     return OpStatus::OK;
   };

src/server/zset_family.h

@@ -19,7 +19,7 @@ namespace dfly {
 class CommandRegistry;
 struct CommandContext;
 class Transaction;
-class OpArgs;
+struct OpArgs;
 
 class ZSetFamily {
  public: