src/commands/cmd_replication.cc - kvrocks - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */

 #include "commander.h"
 #include "error_constants.h"
 #include "event_util.h"
 #include "io_util.h"
 #include "scope_exit.h"
 #include "server/redis_reply.h"
 #include "server/server.h"
 #include "thread_util.h"
 #include "time_util.h"
 #include "unique_fd.h"

 namespace redis {

 class CommandPSync : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     size_t seq_arg = 1;
     if (args.size() == 3) {
       seq_arg = 2;
       new_psync_ = true;
     }

     auto parse_result = ParseInt<uint64_t>(args[seq_arg], 10);
     if (!parse_result) {
       return {Status::RedisParseErr, "value is not an unsigned long long or out of range"};
     }

     next_repl_seq_ = static_cast<rocksdb::SequenceNumber>(*parse_result);
     if (new_psync_) {
       assert(args.size() == 3);
       replica_replid_ = args[1];
       if (replica_replid_.size() != kReplIdLength) {
         return {Status::RedisParseErr, "Wrong replication id length"};
       }
     }

     return Commander::Parse(args);
   }

   Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     INFO(
         "Slave {}, listening port: {}, announce ip: {} asks for synchronization "
         "with next sequence: {}, replication id: {}, and local sequence: {}",
         conn->GetAddr(), conn->GetListeningPort(), conn->GetAnnounceIP(), next_repl_seq_,
         (replica_replid_.length() ? replica_replid_ : "not supported"), srv->storage->LatestSeqNumber());
     bool need_full_sync = false;

     // Check replication id of the last sequence log
     if (new_psync_ && srv->GetConfig()->use_rsid_psync) {
       std::string replid_in_wal = srv->storage->GetReplIdFromWalBySeq(next_repl_seq_ - 1);
       INFO("Replication id in WAL: {}", replid_in_wal);

       // We check replication id only when WAL has this sequence, since there may be no WAL,
       // Or WAL may have nothing when starting from db of old version kvrocks.
       if (replid_in_wal.length() == kReplIdLength && replid_in_wal != replica_replid_) {
         *output = "wrong replication id of the last log";
         need_full_sync = true;
       }
     }

     // Check Log sequence
     if (!need_full_sync && !checkWALBoundary(srv->storage, next_repl_seq_).IsOK()) {
       *output = "sequence out of range, please use fullsync";
       need_full_sync = true;
     }

     if (need_full_sync) {
       srv->stats.IncrPSyncErrCount();
       return {Status::RedisExecErr, *output};
     }

     // Server would spawn a new thread to sync the batch, and connection would
     // be taken over, so should never trigger any event in worker thread.
     conn->Detach();
     conn->EnableFlag(redis::Connection::kSlave);
     auto s = util::SockSetBlocking(conn->GetFD(), 1);
     if (!s.IsOK()) {
       conn->EnableFlag(redis::Connection::kCloseAsync);
       return s.Prefixed("failed to set blocking mode on socket");
     }

     srv->stats.IncrPSyncOKCount();
     s = srv->AddSlave(conn, next_repl_seq_);
     if (!s.IsOK()) {
       std::string err = redis::Error(s);
       s = util::SockSend(conn->GetFD(), err, conn->GetBufferEvent());
       if (!s.IsOK()) {
         WARN("failed to send error message to the replica: {}", s.Msg());
       }
       conn->EnableFlag(redis::Connection::kCloseAsync);
       WARN("Failed to add replica: {} to start incremental syncing", conn->GetAddr());
     } else {
       INFO("New replica: {} was added, start incremental syncing", conn->GetAddr());
     }
     return s;
   }

  private:
   rocksdb::SequenceNumber next_repl_seq_ = 0;
   bool new_psync_ = false;
   std::string replica_replid_;

   // Return OK if the seq is in the range of the current WAL
   static Status checkWALBoundary(engine::Storage *storage, rocksdb::SequenceNumber seq) {
     if (seq == storage->LatestSeqNumber() + 1) {
       return Status::OK();
     }

     // Upper bound
     if (seq > storage->LatestSeqNumber() + 1) {
       return {Status::NotOK};
     }

     // Lower bound
     std::unique_ptr<rocksdb::TransactionLogIterator> iter;
     auto s = storage->GetWALIter(seq, &iter);
     if (s.IsOK() && iter->Valid()) {
       auto batch = iter->GetBatch();
       if (seq != batch.sequence) {
         if (seq > batch.sequence) {
           ERROR("checkWALBoundary with sequence: {}, but GetWALIter return older sequence: {}", seq, batch.sequence);
         }
         return {Status::NotOK};
       }
       return Status::OK();
     }
     return {Status::NotOK};
   }
 };

 class CommandReplConf : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     if (args.size() % 2 == 0) {
       return {Status::RedisParseErr, errWrongNumOfArguments};
     }

     for (size_t i = 1; i < args.size(); i += 2) {
       Status s = ParseParam(util::ToLower(args[i]), args[i + 1]);
       if (!s.IsOK()) {
         return s;
       }
     }

     return Commander::Parse(args);
   }

   Status ParseParam(const std::string &option, const std::string &value) {
     if (option == "listening-port") {
       auto parse_result = ParseInt<int>(value, NumericRange<int>{1, PORT_LIMIT - 1}, 10);
       if (!parse_result) {
         return {Status::RedisParseErr, "listening-port should be number or out of range"};
       }

       port_ = *parse_result;
     } else if (option == "ip-address") {
       if (value == "") {
         return {Status::RedisParseErr, "ip-address should not be empty"};
       }
       ip_address_ = value;
     } else {
       return {Status::RedisParseErr, errUnknownOption};
     }

     return Status::OK();
   }

   Status Execute([[maybe_unused]] engine::Context &ctx, [[maybe_unused]] Server *srv, Connection *conn,
                  std::string *output) override {
     if (port_ != 0) {
       conn->SetListeningPort(port_);
     }
     if (!ip_address_.empty()) {
       conn->SetAnnounceIP(ip_address_);
     }
     *output = redis::RESP_OK;
     return Status::OK();
   }

  private:
   int port_ = 0;
   std::string ip_address_;
 };

 class CommandFetchMeta : public Commander {
  public:
   Status Parse([[maybe_unused]] const std::vector<std::string> &args) override { return Status::OK(); }

   Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn,
                  [[maybe_unused]] std::string *output) override {
     int repl_fd = conn->GetFD();
     std::string ip = conn->GetAnnounceIP();

     auto s = util::SockSetBlocking(repl_fd, 1);
     if (!s.IsOK()) {
       return s.Prefixed("failed to set blocking mode on socket");
     }

     conn->NeedNotFreeBufferEvent();
     conn->EnableFlag(redis::Connection::kCloseAsync);
     srv->stats.IncrFullSyncCount();

     // Feed-replica-meta thread
     auto t = GET_OR_RET(util::CreateThread("feed-repl-info", [srv, repl_fd, ip, bev = conn->GetBufferEvent()] {
       srv->IncrFetchFileThread();
       auto exit = MakeScopeExit([srv, bev] {
         bufferevent_free(bev);
         srv->DecrFetchFileThread();
       });

       std::string files;
       auto s = engine::Storage::ReplDataManager::GetFullReplDataInfo(srv->storage, &files);
       if (!s.IsOK()) {
         WARN("[replication] Failed to get full data file info: {}", s.Msg());
         s = util::SockSend(repl_fd, redis::Error({Status::RedisErrorNoPrefix, "can't create db checkpoint"}), bev);
         if (!s.IsOK()) {
           WARN("[replication] Failed to send error response: {}", s.Msg());
         }
         return;
       }
       // Send full data file info
       if (auto s = util::SockSend(repl_fd, files + CRLF, bev)) {
         INFO("[replication] Succeed sending full data file info to {}", ip);
       } else {
         WARN("[replication] Fail to send full data file info {}, error: {}", ip, s.Msg());
       }
       auto now_secs = static_cast<time_t>(util::GetTimeStamp());
       srv->storage->SetCheckpointAccessTimeSecs(now_secs);
     }));

     if (auto s = util::ThreadDetach(t); !s) {
       return s;
     }

     return Status::OK();
   }
 };

 class CommandFetchFile : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     files_str_ = args[1];
     return Status::OK();
   }

   Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn,
                  [[maybe_unused]] std::string *output) override {
     std::vector<std::string> files = util::Split(files_str_, ",");

     int repl_fd = conn->GetFD();
     std::string ip = conn->GetAnnounceIP();

     auto s = util::SockSetBlocking(repl_fd, 1);
     if (!s.IsOK()) {
       return s.Prefixed("failed to set blocking mode on socket");
     }

     conn->NeedNotFreeBufferEvent();  // Feed-replica-file thread will close the replica bufferevent
     conn->EnableFlag(redis::Connection::kCloseAsync);

     auto t = GET_OR_RET(util::CreateThread("feed-repl-file", [srv, repl_fd, ip, files, bev = conn->GetBufferEvent()]() {
       auto exit = MakeScopeExit([bev] { bufferevent_free(bev); });
       srv->IncrFetchFileThread();

       for (const auto &file : files) {
         if (srv->IsStopped()) break;

         uint64_t file_size = 0, max_replication_bytes = 0;
         if (srv->GetConfig()->max_replication_mb > 0 && srv->GetFetchFileThreadNum() != 0) {
           max_replication_bytes = (srv->GetConfig()->max_replication_mb * MiB) / srv->GetFetchFileThreadNum();
         }
         auto start = std::chrono::high_resolution_clock::now();
         auto fd = UniqueFD(engine::Storage::ReplDataManager::OpenDataFile(srv->storage, file, &file_size));
         if (!fd) break;

         // Send file size and content
         auto s = util::SockSend(repl_fd, std::to_string(file_size) + CRLF, bev);
         if (s) {
           s = util::SockSendFile(repl_fd, *fd, file_size, bev);
         }
         if (s) {
           INFO("[replication] Succeed sending file {} to {}", file, ip);
         } else {
           WARN("[replication] Fail to send file {} to {}, error: {}", file, ip, s.Msg());
           break;
         }
         fd.Close();

         // Sleep if the speed of sending file is more than replication speed limit
         auto end = std::chrono::high_resolution_clock::now();
         uint64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
         if (max_replication_bytes > 0) {
           auto shortest = static_cast<uint64_t>(static_cast<double>(file_size) /
                                                 static_cast<double>(max_replication_bytes) * (1000 * 1000));
           if (duration < shortest) {
             INFO("[replication] Need to sleep {} ms since of sending files too quickly", (shortest - duration) / 1000);
             usleep(shortest - duration);
           }
         }
       }
       auto now_secs = util::GetTimeStamp<std::chrono::seconds>();
       srv->storage->SetCheckpointAccessTimeSecs(now_secs);
       srv->DecrFetchFileThread();
     }));

     if (auto s = util::ThreadDetach(t); !s) {
       return s;
     }

     return Status::OK();
   }

  private:
   std::string files_str_;
 };

 class CommandDBName : public Commander {
  public:
   Status Parse([[maybe_unused]] const std::vector<std::string> &args) override { return Status::OK(); }

   Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn,
                  [[maybe_unused]] std::string *output) override {
     conn->Reply(srv->storage->GetName() + CRLF);
     return Status::OK();
   }
 };

 class CommandWait : public Commander,
                     private EventCallbackBase<CommandWait>,
                     private EvbufCallbackBase<CommandWait, false> {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     auto num_replicas_result = ParseInt<int64_t>(args[1], 10);
     if (!num_replicas_result || *num_replicas_result <= 0) {
       return {Status::RedisParseErr, "numreplicas should be a positive integer"};
     }

     num_replicas_ = *num_replicas_result;

     auto timeout_result = ParseInt<int64_t>(args[2], 10);
     if (!timeout_result || *timeout_result < 0) {
       return {Status::RedisParseErr, "timeout should be a non-negative integer"};
     }

     timeout_ = *timeout_result * 1000;

     return Commander::Parse(args);
   }

   Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     // Only master can execute WAIT command
     if (srv->IsSlave()) {
       return {Status::RedisExecErr, "WAIT command can only be executed on master"};
     }

     // Get current sequence number
     auto current_seq = srv->storage->LatestSeqNumber();

     // Check if we already have enough replicas at the current sequence
     size_t reached_replicas = srv->GetReplicasReachedSequence(current_seq);

     // If we already have enough replicas, return immediately
     if (reached_replicas >= num_replicas_) {
       *output = redis::Integer(reached_replicas);
       return Status::OK();
     }

     conn_ = conn;
     srv_ = srv;

     // Block the connection and wait for replicas to catch up
     srv->BlockOnWait(conn, current_seq, num_replicas_);

     // set callback to use the callbacks defined in this class
     SetCB(conn->GetBufferEvent());

     // Disable read event so the connection will not process any other commands
     // Disable write event so the connection will not send response
     bufferevent_disable(conn->GetBufferEvent(), EV_READ | EV_WRITE);

     if (timeout_ > 0) {
       initTimer(current_seq, timeout_);
     }

     // The connection will be woken up by WakeupWaitConnections when enough replicas
     // have reached the target sequence
     return {Status::BlockingCmd};
   }

   void OnWrite(bufferevent *bev) {
     // Before unblocking, we must confirm that the wait condition has actually been met or that a timeout occurred.
     // Ideally, write callback is called after WAIT response is sent, but it may be called for previous commands.
     // so we need to check if the connection is still waiting.
     // For example, considering the following scenario:
     // 1. SET k1 v1
     // 2. WAIT 1 0
     // 3. SET k1 v2
     // After WAIT 1 0 is executed, the connection is blocked, and the write callback is called for SET k1 v1.
     size_t reached_replicas = srv_->GetReplicasReachedSequence(target_seq_);
     bool wait_condition_met = (reached_replicas >= num_replicas_);

     // The timer is reset in the TimerCB before the connection is woken up.
     // If the timer is null, it means we were woken up by a timeout.
     bool timed_out = (timeout_ > 0 && timer_ == nullptr);

     if (!wait_condition_met && !timed_out) {
       return;  // This is a premature write, so we do nothing and keep the connection blocked.
     }

     if (timer_ != nullptr) {
       timer_.reset();
     }

     conn_->SetCB(bev);

     bufferevent_enable(bev, EV_READ);
     // We need to manually trigger the read event since we will stop processing commands
     // in connection after the blocking command, so there may have some commands to be processed.
     // Related issue: https://github.com/apache/kvrocks/issues/831
     bufferevent_trigger(bev, EV_READ, BEV_TRIG_IGNORE_WATERMARKS);
   }

   void TimerCB(int, int16_t) {
     timer_.reset();
     // Wake up the connection upon timeout.
     // WakeupWaitConnection will hold the lock of the connection during the execution,
     // holding the lock is necessary to avoid race condition that timeout and replication ack happen at the same time.
     srv_->WakeupWaitConnection(conn_, target_seq_);
   }

   void OnEvent(bufferevent *bev, int16_t events) {
     if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
       if (timer_ != nullptr) {
         timer_.reset();
       }
     }

     conn_->OnEvent(bev, events);
   }

  private:
   // variables used for timeout only
   int64_t timeout_ = 0;  // microseconds
   UniqueEvent timer_;
   rocksdb::SequenceNumber target_seq_ = 0;

   // variables used for all cases
   Server *srv_ = nullptr;
   uint64_t num_replicas_ = 0;
   Connection *conn_ = nullptr;

   void initTimer(rocksdb::SequenceNumber target_seq, int64_t timeout) {
     target_seq_ = target_seq;

     // init timer
     auto bev = conn_->GetBufferEvent();
     timer_.reset(NewTimer(bufferevent_get_base(bev)));
     int64_t timeout_second = timeout / 1000 / 1000;
     int64_t timeout_microsecond = timeout % (1000 * 1000);
     timeval tm = {timeout_second, static_cast<int>(timeout_microsecond)};
     evtimer_add(timer_.get(), &tm);
   }
 };

 REDIS_REGISTER_COMMANDS(Replication, MakeCmdAttr<CommandReplConf>("replconf", -3, "read-only no-script", NO_KEY),
                         MakeCmdAttr<CommandPSync>("psync", -2, "read-only no-multi no-script admin", NO_KEY),
                         MakeCmdAttr<CommandFetchMeta>("_fetch_meta", 1, "read-only no-multi no-script admin", NO_KEY),
                         MakeCmdAttr<CommandFetchFile>("_fetch_file", 2, "read-only no-multi no-script admin", NO_KEY),
                         MakeCmdAttr<CommandDBName>("_db_name", 1, "read-only no-multi", NO_KEY),
                         MakeCmdAttr<CommandWait>("wait", 3, "read-only no-multi no-script blocking", NO_KEY), )

 }  // namespace redis
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/

	#include "commander.h"
	#include "error_constants.h"
	#include "event_util.h"
	#include "io_util.h"
	#include "scope_exit.h"
	#include "server/redis_reply.h"
	#include "server/server.h"
	#include "thread_util.h"
	#include "time_util.h"
	#include "unique_fd.h"

	namespace redis {

	class CommandPSync : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	size_t seq_arg = 1;
	if (args.size() == 3) {
	seq_arg = 2;
	new_psync_ = true;
	}

	auto parse_result = ParseInt<uint64_t>(args[seq_arg], 10);
	if (!parse_result) {
	return {Status::RedisParseErr, "value is not an unsigned long long or out of range"};
	}

	next_repl_seq_ = static_cast<rocksdb::SequenceNumber>(*parse_result);
	if (new_psync_) {
	assert(args.size() == 3);
	replica_replid_ = args[1];
	if (replica_replid_.size() != kReplIdLength) {
	return {Status::RedisParseErr, "Wrong replication id length"};
	}
	}

	return Commander::Parse(args);
	}

	Status Execute([[maybe_unused]] engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	INFO(
	"Slave {}, listening port: {}, announce ip: {} asks for synchronization "
	"with next sequence: {}, replication id: {}, and local sequence: {}",
	conn->GetAddr(), conn->GetListeningPort(), conn->GetAnnounceIP(), next_repl_seq_,
	(replica_replid_.length() ? replica_replid_ : "not supported"), srv->storage->LatestSeqNumber());
	bool need_full_sync = false;

	// Check replication id of the last sequence log
	if (new_psync_ && srv->GetConfig()->use_rsid_psync) {
	std::string replid_in_wal = srv->storage->GetReplIdFromWalBySeq(next_repl_seq_ - 1);
	INFO("Replication id in WAL: {}", replid_in_wal);

	// We check replication id only when WAL has this sequence, since there may be no WAL,
	// Or WAL may have nothing when starting from db of old version kvrocks.
	if (replid_in_wal.length() == kReplIdLength && replid_in_wal != replica_replid_) {
	*output = "wrong replication id of the last log";
	need_full_sync = true;
	}
	}

	// Check Log sequence
	if (!need_full_sync && !checkWALBoundary(srv->storage, next_repl_seq_).IsOK()) {
	*output = "sequence out of range, please use fullsync";
	need_full_sync = true;
	}

	if (need_full_sync) {
	srv->stats.IncrPSyncErrCount();
	return {Status::RedisExecErr, *output};
	}

	// Server would spawn a new thread to sync the batch, and connection would
	// be taken over, so should never trigger any event in worker thread.
	conn->Detach();
	conn->EnableFlag(redis::Connection::kSlave);
	auto s = util::SockSetBlocking(conn->GetFD(), 1);
	if (!s.IsOK()) {
	conn->EnableFlag(redis::Connection::kCloseAsync);
	return s.Prefixed("failed to set blocking mode on socket");
	}

	srv->stats.IncrPSyncOKCount();
	s = srv->AddSlave(conn, next_repl_seq_);
	if (!s.IsOK()) {
	std::string err = redis::Error(s);
	s = util::SockSend(conn->GetFD(), err, conn->GetBufferEvent());
	if (!s.IsOK()) {
	WARN("failed to send error message to the replica: {}", s.Msg());
	}
	conn->EnableFlag(redis::Connection::kCloseAsync);
	WARN("Failed to add replica: {} to start incremental syncing", conn->GetAddr());
	} else {
	INFO("New replica: {} was added, start incremental syncing", conn->GetAddr());
	}
	return s;
	}

	private:
	rocksdb::SequenceNumber next_repl_seq_ = 0;
	bool new_psync_ = false;
	std::string replica_replid_;

	// Return OK if the seq is in the range of the current WAL
	static Status checkWALBoundary(engine::Storage *storage, rocksdb::SequenceNumber seq) {
	if (seq == storage->LatestSeqNumber() + 1) {
	return Status::OK();
	}

	// Upper bound
	if (seq > storage->LatestSeqNumber() + 1) {
	return {Status::NotOK};
	}

	// Lower bound
	std::unique_ptr<rocksdb::TransactionLogIterator> iter;
	auto s = storage->GetWALIter(seq, &iter);
	if (s.IsOK() && iter->Valid()) {
	auto batch = iter->GetBatch();
	if (seq != batch.sequence) {
	if (seq > batch.sequence) {
	ERROR("checkWALBoundary with sequence: {}, but GetWALIter return older sequence: {}", seq, batch.sequence);
	}
	return {Status::NotOK};
	}
	return Status::OK();
	}
	return {Status::NotOK};
	}
	};

	class CommandReplConf : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	if (args.size() % 2 == 0) {
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}

	for (size_t i = 1; i < args.size(); i += 2) {
	Status s = ParseParam(util::ToLower(args[i]), args[i + 1]);
	if (!s.IsOK()) {
	return s;
	}
	}

	return Commander::Parse(args);
	}

	Status ParseParam(const std::string &option, const std::string &value) {
	if (option == "listening-port") {
	auto parse_result = ParseInt<int>(value, NumericRange<int>{1, PORT_LIMIT - 1}, 10);
	if (!parse_result) {
	return {Status::RedisParseErr, "listening-port should be number or out of range"};
	}

	port_ = *parse_result;
	} else if (option == "ip-address") {
	if (value == "") {
	return {Status::RedisParseErr, "ip-address should not be empty"};
	}
	ip_address_ = value;
	} else {
	return {Status::RedisParseErr, errUnknownOption};
	}

	return Status::OK();
	}

	Status Execute([[maybe_unused]] engine::Context &ctx, [[maybe_unused]] Server srv, Connection conn,
	std::string *output) override {
	if (port_ != 0) {
	conn->SetListeningPort(port_);
	}
	if (!ip_address_.empty()) {
	conn->SetAnnounceIP(ip_address_);
	}
	*output = redis::RESP_OK;
	return Status::OK();
	}

	private:
	int port_ = 0;
	std::string ip_address_;
	};

	class CommandFetchMeta : public Commander {
	public:
	Status Parse([[maybe_unused]] const std::vector<std::string> &args) override { return Status::OK(); }

	Status Execute([[maybe_unused]] engine::Context &ctx, Server srv, Connection conn,
	[[maybe_unused]] std::string *output) override {
	int repl_fd = conn->GetFD();
	std::string ip = conn->GetAnnounceIP();

	auto s = util::SockSetBlocking(repl_fd, 1);
	if (!s.IsOK()) {
	return s.Prefixed("failed to set blocking mode on socket");
	}

	conn->NeedNotFreeBufferEvent();
	conn->EnableFlag(redis::Connection::kCloseAsync);
	srv->stats.IncrFullSyncCount();

	// Feed-replica-meta thread
	auto t = GET_OR_RET(util::CreateThread("feed-repl-info", [srv, repl_fd, ip, bev = conn->GetBufferEvent()] {
	srv->IncrFetchFileThread();
	auto exit = MakeScopeExit([srv, bev] {
	bufferevent_free(bev);
	srv->DecrFetchFileThread();
	});

	std::string files;
	auto s = engine::Storage::ReplDataManager::GetFullReplDataInfo(srv->storage, &files);
	if (!s.IsOK()) {
	WARN("[replication] Failed to get full data file info: {}", s.Msg());
	s = util::SockSend(repl_fd, redis::Error({Status::RedisErrorNoPrefix, "can't create db checkpoint"}), bev);
	if (!s.IsOK()) {
	WARN("[replication] Failed to send error response: {}", s.Msg());
	}
	return;
	}
	// Send full data file info
	if (auto s = util::SockSend(repl_fd, files + CRLF, bev)) {
	INFO("[replication] Succeed sending full data file info to {}", ip);
	} else {
	WARN("[replication] Fail to send full data file info {}, error: {}", ip, s.Msg());
	}
	auto now_secs = static_cast<time_t>(util::GetTimeStamp());
	srv->storage->SetCheckpointAccessTimeSecs(now_secs);
	}));

	if (auto s = util::ThreadDetach(t); !s) {
	return s;
	}

	return Status::OK();
	}
	};

	class CommandFetchFile : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	files_str_ = args[1];
	return Status::OK();
	}

	Status Execute([[maybe_unused]] engine::Context &ctx, Server srv, Connection conn,
	[[maybe_unused]] std::string *output) override {
	std::vector<std::string> files = util::Split(files_str_, ",");

	int repl_fd = conn->GetFD();
	std::string ip = conn->GetAnnounceIP();

	auto s = util::SockSetBlocking(repl_fd, 1);
	if (!s.IsOK()) {
	return s.Prefixed("failed to set blocking mode on socket");
	}

	conn->NeedNotFreeBufferEvent(); // Feed-replica-file thread will close the replica bufferevent
	conn->EnableFlag(redis::Connection::kCloseAsync);

	auto t = GET_OR_RET(util::CreateThread("feed-repl-file", [srv, repl_fd, ip, files, bev = conn->GetBufferEvent()]() {
	auto exit = MakeScopeExit([bev] { bufferevent_free(bev); });
	srv->IncrFetchFileThread();

	for (const auto &file : files) {
	if (srv->IsStopped()) break;

	uint64_t file_size = 0, max_replication_bytes = 0;
	if (srv->GetConfig()->max_replication_mb > 0 && srv->GetFetchFileThreadNum() != 0) {
	max_replication_bytes = (srv->GetConfig()->max_replication_mb * MiB) / srv->GetFetchFileThreadNum();
	}
	auto start = std::chrono::high_resolution_clock::now();
	auto fd = UniqueFD(engine::Storage::ReplDataManager::OpenDataFile(srv->storage, file, &file_size));
	if (!fd) break;

	// Send file size and content
	auto s = util::SockSend(repl_fd, std::to_string(file_size) + CRLF, bev);
	if (s) {
	s = util::SockSendFile(repl_fd, *fd, file_size, bev);
	}
	if (s) {
	INFO("[replication] Succeed sending file {} to {}", file, ip);
	} else {
	WARN("[replication] Fail to send file {} to {}, error: {}", file, ip, s.Msg());
	break;
	}
	fd.Close();

	// Sleep if the speed of sending file is more than replication speed limit
	auto end = std::chrono::high_resolution_clock::now();
	uint64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
	if (max_replication_bytes > 0) {
	auto shortest = static_cast<uint64_t>(static_cast<double>(file_size) /
	static_cast<double>(max_replication_bytes) * (1000 * 1000));
	if (duration < shortest) {
	INFO("[replication] Need to sleep {} ms since of sending files too quickly", (shortest - duration) / 1000);
	usleep(shortest - duration);
	}
	}
	}
	auto now_secs = util::GetTimeStamp<std::chrono::seconds>();
	srv->storage->SetCheckpointAccessTimeSecs(now_secs);
	srv->DecrFetchFileThread();
	}));

	if (auto s = util::ThreadDetach(t); !s) {
	return s;
	}

	return Status::OK();
	}

	private:
	std::string files_str_;
	};

	class CommandDBName : public Commander {
	public:
	Status Parse([[maybe_unused]] const std::vector<std::string> &args) override { return Status::OK(); }

	Status Execute([[maybe_unused]] engine::Context &ctx, Server srv, Connection conn,
	[[maybe_unused]] std::string *output) override {
	conn->Reply(srv->storage->GetName() + CRLF);
	return Status::OK();
	}
	};

	class CommandWait : public Commander,
	private EventCallbackBase<CommandWait>,
	private EvbufCallbackBase<CommandWait, false> {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	auto num_replicas_result = ParseInt<int64_t>(args[1], 10);
	if (!num_replicas_result \|\| *num_replicas_result <= 0) {
	return {Status::RedisParseErr, "numreplicas should be a positive integer"};
	}

	num_replicas_ = *num_replicas_result;

	auto timeout_result = ParseInt<int64_t>(args[2], 10);
	if (!timeout_result \|\| *timeout_result < 0) {
	return {Status::RedisParseErr, "timeout should be a non-negative integer"};
	}

	timeout_ = timeout_result 1000;

	return Commander::Parse(args);
	}

	Status Execute([[maybe_unused]] engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	// Only master can execute WAIT command
	if (srv->IsSlave()) {
	return {Status::RedisExecErr, "WAIT command can only be executed on master"};
	}

	// Get current sequence number
	auto current_seq = srv->storage->LatestSeqNumber();

	// Check if we already have enough replicas at the current sequence
	size_t reached_replicas = srv->GetReplicasReachedSequence(current_seq);

	// If we already have enough replicas, return immediately
	if (reached_replicas >= num_replicas_) {
	*output = redis::Integer(reached_replicas);
	return Status::OK();
	}

	conn_ = conn;
	srv_ = srv;

	// Block the connection and wait for replicas to catch up
	srv->BlockOnWait(conn, current_seq, num_replicas_);

	// set callback to use the callbacks defined in this class
	SetCB(conn->GetBufferEvent());

	// Disable read event so the connection will not process any other commands
	// Disable write event so the connection will not send response
	bufferevent_disable(conn->GetBufferEvent(), EV_READ \| EV_WRITE);

	if (timeout_ > 0) {
	initTimer(current_seq, timeout_);
	}

	// The connection will be woken up by WakeupWaitConnections when enough replicas
	// have reached the target sequence
	return {Status::BlockingCmd};
	}

	void OnWrite(bufferevent *bev) {
	// Before unblocking, we must confirm that the wait condition has actually been met or that a timeout occurred.
	// Ideally, write callback is called after WAIT response is sent, but it may be called for previous commands.
	// so we need to check if the connection is still waiting.
	// For example, considering the following scenario:
	// 1. SET k1 v1
	// 2. WAIT 1 0
	// 3. SET k1 v2
	// After WAIT 1 0 is executed, the connection is blocked, and the write callback is called for SET k1 v1.
	size_t reached_replicas = srv_->GetReplicasReachedSequence(target_seq_);
	bool wait_condition_met = (reached_replicas >= num_replicas_);

	// The timer is reset in the TimerCB before the connection is woken up.
	// If the timer is null, it means we were woken up by a timeout.
	bool timed_out = (timeout_ > 0 && timer_ == nullptr);

	if (!wait_condition_met && !timed_out) {
	return; // This is a premature write, so we do nothing and keep the connection blocked.
	}

	if (timer_ != nullptr) {
	timer_.reset();
	}

	conn_->SetCB(bev);

	bufferevent_enable(bev, EV_READ);
	// We need to manually trigger the read event since we will stop processing commands
	// in connection after the blocking command, so there may have some commands to be processed.
	// Related issue: https://github.com/apache/kvrocks/issues/831
	bufferevent_trigger(bev, EV_READ, BEV_TRIG_IGNORE_WATERMARKS);
	}

	void TimerCB(int, int16_t) {
	timer_.reset();
	// Wake up the connection upon timeout.
	// WakeupWaitConnection will hold the lock of the connection during the execution,
	// holding the lock is necessary to avoid race condition that timeout and replication ack happen at the same time.
	srv_->WakeupWaitConnection(conn_, target_seq_);
	}

	void OnEvent(bufferevent *bev, int16_t events) {
	if (events & (BEV_EVENT_EOF \| BEV_EVENT_ERROR)) {
	if (timer_ != nullptr) {
	timer_.reset();
	}
	}

	conn_->OnEvent(bev, events);
	}

	private:
	// variables used for timeout only
	int64_t timeout_ = 0; // microseconds
	UniqueEvent timer_;
	rocksdb::SequenceNumber target_seq_ = 0;

	// variables used for all cases
	Server *srv_ = nullptr;
	uint64_t num_replicas_ = 0;
	Connection *conn_ = nullptr;

	void initTimer(rocksdb::SequenceNumber target_seq, int64_t timeout) {
	target_seq_ = target_seq;

	// init timer
	auto bev = conn_->GetBufferEvent();
	timer_.reset(NewTimer(bufferevent_get_base(bev)));
	int64_t timeout_second = timeout / 1000 / 1000;
	int64_t timeout_microsecond = timeout % (1000 * 1000);
	timeval tm = {timeout_second, static_cast<int>(timeout_microsecond)};
	evtimer_add(timer_.get(), &tm);
	}
	};

	REDIS_REGISTER_COMMANDS(Replication, MakeCmdAttr<CommandReplConf>("replconf", -3, "read-only no-script", NO_KEY),
	MakeCmdAttr<CommandPSync>("psync", -2, "read-only no-multi no-script admin", NO_KEY),
	MakeCmdAttr<CommandFetchMeta>("_fetch_meta", 1, "read-only no-multi no-script admin", NO_KEY),
	MakeCmdAttr<CommandFetchFile>("_fetch_file", 2, "read-only no-multi no-script admin", NO_KEY),
	MakeCmdAttr<CommandDBName>("_db_name", 1, "read-only no-multi", NO_KEY),
	MakeCmdAttr<CommandWait>("wait", 3, "read-only no-multi no-script blocking", NO_KEY), )

	} // namespace redis