address comments

Author: ltagliamonte-dd

src/server/server.cc

@@ -2210,3 +2210,5 @@ AuthResult Server::AuthenticateUser(const std::string &user_password, std::strin
   *ns = kDefaultNamespace;
   return AuthResult::IS_ADMIN;
 }
+
+TaskRunner *Server::GetTaskRunner() { return &task_runner_; }

src/server/server.h

@@ -197,6 +197,8 @@ class Server {
   void AdjustWorkerThreads();
 
   Status AddMaster(const std::string &host, uint32_t port, bool force_reconnect);
+
+  TaskRunner *GetTaskRunner() { return &task_runner_; }
   Status RemoveMaster();
   Status AddSlave(redis::Connection *conn, rocksdb::SequenceNumber next_repl_seq);
   void DisconnectSlaves();
@@ -434,6 +436,9 @@ class Server {
   std::thread cron_thread_;
   std::thread compaction_checker_thread_;
   TaskRunner task_runner_;
+
+ public:
+  TaskRunner *GetTaskRunner() { return &task_runner_; }
   std::vector<std::unique_ptr<WorkerThread>> worker_threads_;
   std::unique_ptr<ReplicationThread> replication_thread_;
   tbb::concurrent_queue<std::unique_ptr<WorkerThread>> recycle_worker_threads_;

src/types/redis_hash.cc

@@ -29,6 +29,7 @@
 #include <thread>
 #include <utility>
 
+#include "common/task_runner.h"
 #include "db_util.h"
 #include "logging.h"
 #include "parse_util.h"
@@ -680,6 +681,7 @@ rocksdb::Status Hash::PersistFields(engine::Context &ctx, const Slice &user_key,
 }
 
 void Hash::asyncRepairHash(const std::string &ns_key, const Slice &field, const HashMetadata &metadata) const {
+  // Use the server's task runner to schedule the repair task instead of spawning a thread.
   auto repair_task = [storage = storage_, ns_key, field_str = field.ToString(), version = metadata.version,
                       size = metadata.size]() {
     if (size == 0) {
@@ -689,27 +691,20 @@ void Hash::asyncRepairHash(const std::string &ns_key, const Slice &field, const
     engine::Context ctx(storage);
     std::string sub_key = InternalKey(ns_key, field_str, version, storage->IsSlotIdEncoded()).Encode();
 
-    // Use a retry loop to implement optimistic locking. This is to atomically
-    // decrement the hash size and prevent a race condition when multiple threads
-    // are repairing the same hash.
     for (int i = 0; i < 10; ++i) {
       HashMetadata current_metadata(false);
       std::string metadata_bytes;
       auto s = storage->Get(ctx, ctx.GetReadOptions(), storage->GetCFHandle(ColumnFamilyID::Metadata), ns_key,
                             &metadata_bytes);
       if (!s.ok()) {
-        // If metadata is not found, another repair might have cleaned it up. Stop.
         if (s.IsNotFound()) return;
-        // For other errors, log and stop.
         error("Failed to get metadata for async repair: {}", s.ToString());
         return;
       }
       if (!current_metadata.Decode(metadata_bytes).ok()) {
         error("Failed to decode metadata for async repair");
         return;
       }
-
-      // If size is already 0, no need to decrement.
       if (current_metadata.size == 0) return;
 
       auto batch = storage->GetWriteBatchBase();
@@ -728,7 +723,10 @@ void Hash::asyncRepairHash(const std::string &ns_key, const Slice &field, const
     error("Failed to async repair hash field after multiple retries");
   };
 
-  std::thread(repair_task).detach();
+  // Use the server's task runner and TryPublish
+  if (server_ && server_->GetTaskRunner()) {
+    server_->GetTaskRunner()->TryPublish(std::move(repair_task));
+  }
 }
 
 }  // namespace redis

src/types/redis_hash.h

@@ -28,6 +28,7 @@
 
 #include "common/range_spec.h"
 #include "encoding.h"
+#include "server/server.h"
 #include "storage/redis_db.h"
 #include "storage/redis_metadata.h"
 #include "time_util.h"
@@ -151,7 +152,8 @@ namespace redis {
 
 class Hash : public SubKeyScanner {
  public:
-  Hash(engine::Storage *storage, const std::string &ns) : SubKeyScanner(storage, ns) {}
+  Hash(engine::Storage *storage, const std::string &ns, Server *server = nullptr)
+      : SubKeyScanner(storage, ns), server_(server) {}
 
   rocksdb::Status Size(engine::Context &ctx, const Slice &user_key, uint64_t *size);
   rocksdb::Status Get(engine::Context &ctx, const Slice &user_key, const Slice &field, std::string *value);
@@ -199,6 +201,7 @@ class Hash : public SubKeyScanner {
  private:
   rocksdb::Status GetMetadata(engine::Context &ctx, const Slice &ns_key, HashMetadata *metadata);
   void asyncRepairHash(const std::string &ns_key, const Slice &field, const HashMetadata &metadata) const;
+  Server *server_ = nullptr;
 
   friend struct FieldValueRetriever;
 };