Notify user code about follower loss

include/libnuraft/callback.hxx

@@ -207,6 +207,13 @@ public:
          */
         ResignationFromLeader = 27,
 
+        /**
+         * When a peer RPC errors count exceeds raft_server::limits.warning_limit_, or
+         * a peer doesn't respond for a long time (raft_params::leadership_expiry_),
+         * the peer is considered lost.
+         * ctx: null.
+         */
+        FollowerLost = 100,
     };
 
     struct Param {

include/libnuraft/peer.hxx

@@ -302,6 +302,10 @@ public:
     ptr<req_msg> get_rsv_msg() const { return rsv_msg_; }
     rpc_handler get_rsv_msg_handler() const { return rsv_msg_handler_; }
 
+    bool IsLost() const { return lost_by_leader_; }
+    void SetLost() { lost_by_leader_ = true; }
+    void SetRecovered() { lost_by_leader_ = false; }
+
 private:
     void handle_rpc_result(ptr<peer> myself,
                            ptr<rpc_client> my_rpc_client,
@@ -498,6 +502,8 @@ private:
      */
     std::atomic<bool> abandoned_;
 
+    std::atomic<bool> lost_by_leader_ {false};
+
     /**
      * Reserved message that should be sent next time.
      */

include/libnuraft/raft_server.hxx

@@ -886,7 +886,8 @@ protected:
     int32 get_quorum_for_election();
     int32 get_quorum_for_commit();
     int32 get_leadership_expiry();
-    size_t get_not_responding_peers();
+    std::vector<ptr<peer>> get_not_responding_peers();
+    size_t get_not_responding_peers_count();
     size_t get_num_stale_peers();
 
     ptr<resp_msg> handle_append_entries(req_msg& req);

src/handle_append_entries.cxx

@@ -162,7 +162,7 @@ bool raft_server::request_append_entries(ptr<peer> p) {
          chk_timer.timeout_and_reset() ) {
         // If auto adjust mode is on for 2-node cluster, and
         // the follower is not responding, adjust the quorum.
-        size_t num_not_responding_peers = get_not_responding_peers();
+        size_t num_not_responding_peers = get_not_responding_peers_count();
         size_t cur_quorum_size = get_quorum_for_commit();
         size_t num_stale_peers = get_num_stale_peers();
         if (cur_quorum_size >= 1) {
@@ -1191,7 +1191,7 @@ ulong raft_server::get_expected_committed_log_idx() {
 
     size_t quorum_idx = get_quorum_for_commit();
     if (ctx_->get_params()->use_full_consensus_among_healthy_members_) {
-        size_t not_responding_peers = get_not_responding_peers();
+        size_t not_responding_peers = get_not_responding_peers_count();
         if (not_responding_peers < voting_members - quorum_idx) {
             // If full consensus option is on, commit should be
             // agreed by all healthy members, and the number of

src/raft_server.cxx

@@ -608,7 +608,25 @@ int32 raft_server::get_leadership_expiry() {
     return expiry;
 }
 
-size_t raft_server::get_not_responding_peers() {
+std::vector<ptr<peer>> raft_server::get_not_responding_peers() {
+    ptr<raft_params> params = ctx_->get_params();
+    const auto expiry =
+        params->heart_beat_interval_ * raft_server::raft_limits_.response_limit_;
+    std::vector<ptr<peer>> rs;
+    for (const auto& [id, peer_ptr]: peers_) {
+        if (!is_regular_member(peer_ptr)) {
+            continue;
+        }
+        const auto resp_elapsed_ms =
+            static_cast<int32>(peer_ptr->get_resp_timer_us() / 1000);
+        if (resp_elapsed_ms > expiry) {
+            rs.push_back(peer_ptr);
+        }
+    }
+    return rs;
+}
+
+size_t raft_server::get_not_responding_peers_count() {
     // Check if quorum nodes are not responding
     // (i.e., don't respond 20x heartbeat time long).
     size_t num_not_resp_nodes = 0;
@@ -804,6 +822,14 @@ void raft_server::handle_peer_resp(ptr<resp_msg>& resp, ptr<rpc_exception>& err)
         } else if (rpc_errs == raft_server::raft_limits_.warning_limit_) {
             p_wn("too verbose RPC error on peer (%d), "
                  "will suppress it from now", peer_id);
+            if (!pp || !pp->IsLost()) {
+                if (pp) {
+                    pp->SetLost();
+                }
+                cb_func::Param param(id_, leader_, peer_id);
+                const auto rc = ctx_->cb_func_.call(cb_func::FollowerLost, &param);
+                assert(rc == cb_func::ReturnCode::Ok);
+            }
         }
 
         if (pp && pp->is_leave_flag_set()) {
@@ -837,6 +863,9 @@ void raft_server::handle_peer_resp(ptr<resp_msg>& resp, ptr<rpc_exception>& err)
             peer* pp = entry->second.get();
             int rpc_errs = pp->get_rpc_errs();
             if (rpc_errs >= raft_server::raft_limits_.warning_limit_) {
+                if (pp) {
+                    pp->SetRecovered();
+                }
                 p_wn("recovered from RPC failure from peer %d, %d errors",
                      resp->get_src(), rpc_errs);
             }
@@ -1085,7 +1114,7 @@ bool raft_server::check_leadership_validity() {
     int32 num_voting_members = get_num_voting_members();
 
     int leadership_expiry = get_leadership_expiry();
-    int32 nr_peers = (int32)get_not_responding_peers();
+    int32 nr_peers = (int32)get_not_responding_peers_count();
     if (leadership_expiry < 0) {
         // Negative expiry: leadership will never expire.
         nr_peers = 0;
@@ -1102,6 +1131,18 @@ bool raft_server::check_leadership_validity() {
              get_leadership_expiry(),
              min_quorum_size);
 
+        const auto nr_peers_list = get_not_responding_peers();
+        assert(nr_peers_list.size() == nr_peers);
+        for (auto& peer : nr_peers_list) {
+          if (peer->IsLost()) {
+            continue;
+          }
+          peer->SetLost();
+          cb_func::Param param(id_, leader_, peer->get_id());
+          const auto rc = ctx_->cb_func_.call(cb_func::FollowerLost, &param);
+          assert(rc == cb_func::ReturnCode::Ok);
+        }
+
         // NOTE:
         //   For a cluster where the number of members is the same
         //   as the size of quorum, we should not expire leadership,