Fixes Raft nodes not to erase their votes on leader step down

src/v/raft/consensus.cc

@@ -84,7 +84,6 @@ void consensus::setup_metrics() {
 }
 
 void consensus::do_step_down() {
-    _voted_for = {};
     _hbeat = clock_type::now();
     _vstate = vote_state::follower;
 }
@@ -700,6 +699,7 @@ void consensus::read_voted_for() {
      * Initial values
      */
     _voted_for = model::node_id{};
+    _last_election = model::term_id(-1);
     _term = model::term_id(0);
 
     /*
@@ -800,8 +800,9 @@ ss::future<vote_reply> consensus::do_vote(vote_request&& r) {
           r.node_id,
           r.term,
           _term);
+        reply.term = r.term;
         _term = r.term;
-        reply.term = _term;
+        _voted_for = {};
         do_step_down();
         // even tough we step down we do not want to update the hbeat as it
         // would cause subsequent votes to fail (_hbeat is updated by the
@@ -814,33 +815,48 @@ ss::future<vote_reply> consensus::do_vote(vote_request&& r) {
         return ss::make_ready_future<vote_reply>(reply);
     }
 
-    auto f = ss::make_ready_future<>();
-
     if (_voted_for() < 0) {
-        _voted_for = model::node_id(r.node_id);
-        vlog(_ctxlog.trace, "Voting for {} in term {}", r.node_id, _term);
-        f = f.then([this] {
-            return write_voted_for({_voted_for, _term})
-              .handle_exception([this](const std::exception_ptr& e) {
+        if (_last_election == _term) {
+            // another fiber is trying to vote at the current term
+            reply.granted = false;
+            return ss::make_ready_future<vote_reply>(std::move(reply));
+        }
+
+        _last_election = _term;
+
+        return write_voted_for({r.node_id, _term})
+          .then_wrapped([this,
+                         reply = std::move(reply),
+                         voting_started_at = _term,
+                         r = std::move(r)](ss::future<> f) {
+              bool granted = false;
+
+              if (f.failed()) {
                   vlog(
                     _ctxlog.warn,
                     "Unable to persist raft group state, vote not granted "
                     "- {}",
-                    e);
-                  _voted_for = {};
-              });
-        });
-    }
-
-    // vote for the same term, same server_id
-    reply.granted = (r.node_id == _voted_for);
-    if (reply.granted) {
-        _hbeat = clock_type::now();
-    }
+                    f.get_exception());
+              } else {
+                  if (voting_started_at == _term) {
+                      _voted_for = r.node_id;
+                      granted = true;
+                  }
+              }
 
-    return f.then([reply = std::move(reply)] {
+              vote_reply response;
+              response.term = reply.term;
+              response.log_ok = reply.log_ok;
+              response.granted = granted;
+              return ss::make_ready_future<vote_reply>(std::move(response));
+          });
+    } else {
+        reply.granted = (r.node_id == _voted_for);
+        if (reply.granted) {
+            _hbeat = clock_type::now();
+        }
         return ss::make_ready_future<vote_reply>(std::move(reply));
-    });
+    }
 }
 
 ss::future<append_entries_reply>
@@ -883,6 +899,7 @@ consensus::do_append_entries(append_entries_request&& r) {
           r.meta.term,
           _term);
         _term = r.meta.term;
+        _voted_for = {};
         return do_append_entries(std::move(r));
     }
 
@@ -1130,6 +1147,7 @@ consensus::do_install_snapshot(install_snapshot_request&& r) {
     // request received from new leader
     if (r.term > _term) {
         _term = r.term;
+        _voted_for = {};
         do_step_down();
         return do_install_snapshot(std::move(r));
     }

src/v/raft/consensus.h

@@ -159,6 +159,7 @@ class consensus {
             // term backward
             if (term > _term) {
                 _term = term;
+                _voted_for = {};
                 do_step_down();
             }
         });
@@ -307,6 +308,7 @@ class consensus {
 
     // read at `ss::future<> start()`
     model::node_id _voted_for;
+    model::term_id _last_election;
     std::optional<model::node_id> _leader_id;
     bool _transferring_leadership{false};
 

src/v/raft/vote_stm.cc

@@ -86,6 +86,7 @@ ss::future<> vote_stm::vote(bool leadership_transfer) {
           _ptr->trigger_leadership_notification();
           // 5.2.1.2
           _ptr->_term += model::term_id(1);
+          _ptr->_voted_for = {};
 
           // vote is the only method under _op_sem
           _ptr->config().for_each_voter(
@@ -186,15 +187,26 @@ ss::future<> vote_stm::wait() { return _vote_bg.close(); }
 
 void vote_stm::update_vote_state(ss::semaphore_units<> u) {
     // use reply term to update voter term
+    bool is_term_behind = false;
+
     for (auto& [_, r] : _replies) {
         if (r.value && r.value->has_value()) {
             auto term = r.value->value().term;
             if (term > _ptr->_term) {
+                vlog(
+                  _ctxlog.info, "Vote failed - received larger term: {}", term);
                 _ptr->_term = term;
+                _ptr->_voted_for = {};
+                is_term_behind = true;
             }
         }
     }
 
+    if (is_term_behind) {
+        _ptr->_vstate = consensus::vote_state::follower;
+        return;
+    }
+
     if (_ptr->_vstate != consensus::vote_state::candidate) {
         vlog(_ctxlog.info, "No longer a candidate, ignoring vote replies");
         return;