fix: after full synchronization fails, full synchronization will be s…

…kipped directly(OpenAtomFoundation#2433) (OpenAtomFoundation#2439)

Co-authored-by: liuchengyu <[email protected]>

include/pika_rm.h

@@ -115,7 +115,7 @@ class SyncSlaveDB : public SyncDB {
   void SetMasterSessionId(int32_t session_id);
   void SetLocalIp(const std::string& local_ip);
   void StopRsync();
-  void ActivateRsync();
+  pstd::Status ActivateRsync();
   bool IsRsyncRunning() {return rsync_cli_->IsRunning();}
 
  private:

src/pika_rm.cc

@@ -502,15 +502,18 @@ void SyncSlaveDB::StopRsync() {
   rsync_cli_->Stop();
 }
 
-void SyncSlaveDB::ActivateRsync() {
+pstd::Status SyncSlaveDB::ActivateRsync() {
+  Status s = Status::OK();
   if (!rsync_cli_->IsIdle()) {
-    return;
+    return s;
   }
   LOG(WARNING) << "ActivateRsync ...";
   if (rsync_cli_->Init()) {
     rsync_cli_->Start();
+    return s;
   } else {
     SetReplState(ReplState::kError);
+    return Status::Error("rsync client init failed!");;
   }
 }
 
@@ -960,7 +963,13 @@ Status PikaReplicaManager::RunSyncSlaveDBStateMachine() {
     } else if (s_db->State() == ReplState::kWaitReply) {
       continue;
     } else if (s_db->State() == ReplState::kWaitDBSync) {
-      s_db->ActivateRsync();
+      Status s = s_db->ActivateRsync();
+      if (!s.ok()) {
+        g_pika_server->SetForceFullSync(true);
+        LOG(WARNING) << "Slave DB: " << s_db->DBName() << " rsync failed! full synchronization will be retried later";
+        continue;
+      }
+
       std::shared_ptr<DB> db =
           g_pika_server->GetDB(p_info.db_name_);
       if (db) {

src/rsync_client.cc

@@ -66,6 +66,7 @@ bool RsyncClient::Init() {
   if (!ret) {
     LOG(WARNING) << "RsyncClient recover failed";
     client_thread_->StopThread();
+    state_.store(IDLE);
     return false;
   }
   finished_work_cnt_.store(0);