[v23.3.x] Do not request votes if term changed in prevote phase

src/v/raft/consensus.cc

@@ -899,11 +899,26 @@ bool consensus::should_skip_vote(bool ignore_heartbeat) {
 }
 
 ss::future<bool> consensus::dispatch_prevote(bool leadership_transfer) {
-    auto pvstm_p = std::make_unique<prevote_stm>(this);
-    auto pvstm = pvstm_p.get();
+    vlog(
+      _ctxlog.info,
+      "starting pre-vote leader election, current term: {}, leadership "
+      "transfer: {}",
+      _term,
+      leadership_transfer);
+
     if (leadership_transfer) {
-        return ss::make_ready_future<bool>(true);
+        return _op_lock.with([this]() {
+            _vstate = vote_state::candidate;
+            if (_leader_id) {
+                _leader_id = std::nullopt;
+                trigger_leadership_notification();
+            }
+            return true;
+        });
     }
+    auto pvstm_p = std::make_unique<prevote_stm>(this);
+    auto pvstm = pvstm_p.get();
+
     return pvstm->prevote(leadership_transfer)
       .then_wrapped([this, pvstm_p = std::move(pvstm_p), pvstm](
                       ss::future<bool> prevote_f) mutable {
@@ -982,7 +997,16 @@ void consensus::dispatch_vote(bool leadership_transfer) {
       = ssx::spawn_with_gate_then(_bg, [this, leadership_transfer] {
             return dispatch_prevote(leadership_transfer)
               .then([this, leadership_transfer](bool ready) mutable {
-                  if (!ready) {
+                  vlog(
+                    _ctxlog.debug,
+                    "pre-vote phase success: {}, current term: {}, "
+                    "leadership transfer: {}",
+                    ready,
+                    _term,
+                    leadership_transfer);
+                  // if a current node is not longer candidate we should skip
+                  // proceeding to actual vote phase
+                  if (!ready || _vstate != vote_state::candidate) {
                       return ss::make_ready_future<>();
                   }
                   auto vstm = std::make_unique<vote_stm>(this);

src/v/raft/prevote_stm.cc

@@ -21,6 +21,7 @@
 #include "rpc/types.h"
 #include "ssx/semaphore.h"
 
+#include <seastar/core/future.hh>
 #include <seastar/util/bool_class.hh>
 
 #include <chrono>
@@ -81,6 +82,18 @@ prevote_stm::process_reply(vnode n, ss::future<result<vote_reply>> f) {
                 auto r = f.get0();
                 if (r.has_value()) {
                     auto v = r.value();
+                    if (v.term != _req.term) {
+                        vlog(
+                          _ctxlog.trace,
+                          "prevote ack: node {} has a higher term {}",
+                          n,
+                          v.term);
+                        voter_reply->second._is_failed = true;
+                        voter_reply->second._is_pending = false;
+                        if (v.term > _req.term) {
+                            _term_update = v.term;
+                        }
+                    }
                     _ptr->maybe_update_node_reply_timestamp(n);
                     if (v.log_ok) {
                         vlog(
@@ -182,20 +195,32 @@ ss::future<bool> prevote_stm::do_prevote() {
       [this](vnode id) { ssx::background = dispatch_prevote(id); });
 
     // process results
-    return process_replies().then([this]() {
-        const auto only_voter = _config->unique_voter_count() == 1
-                                && _config->is_voter(_ptr->self());
-        if (
-          _success && !only_voter
-          && _ptr->_node_priority_override == zero_voter_priority) {
-            vlog(
-              _ctxlog.debug,
-              "Ignoring successful pre-vote. Node priority too low: {}",
-              _ptr->_node_priority_override.value());
-            _success = false;
-        }
-        return _success;
-    });
+    return process_replies()
+      .then([this]() {
+          const auto only_voter = _config->unique_voter_count() == 1
+                                  && _config->is_voter(_ptr->self());
+          if (
+            _success && !only_voter
+            && _ptr->_node_priority_override == zero_voter_priority) {
+              vlog(
+                _ctxlog.debug,
+                "Ignoring successful pre-vote. Node priority too low: {}",
+                _ptr->_node_priority_override.value());
+              _success = false;
+          }
+      })
+      .then([this] {
+          if (_term_update) {
+              return update_term().then([this] { return _success; });
+          }
+          return ss::make_ready_future<bool>(_success);
+      });
+}
+
+ss::future<> prevote_stm::update_term() {
+    auto u = co_await _ptr->_op_lock.get_units();
+    _ptr->_term = std::max(_ptr->term(), _term_update.value());
+    _ptr->_vstate = consensus::vote_state::candidate;
 }
 
 ss::future<> prevote_stm::process_replies() {

src/v/raft/prevote_stm.h

@@ -11,6 +11,7 @@
 
 #pragma once
 
+#include "model/fundamental.h"
 #include "outcome.h"
 #include "raft/logger.h"
 #include "raft/types.h"
@@ -51,11 +52,13 @@ class prevote_stm {
     ss::future<result<vote_reply>> do_dispatch_prevote(vnode);
     ss::future<> process_reply(vnode n, ss::future<result<vote_reply>> f);
     ss::future<> process_replies();
+    ss::future<> update_term();
     // args
     consensus* _ptr;
     // make sure to always make a copy; never move() this struct
     vote_request _req;
     bool _success = false;
+    std::optional<model::term_id> _term_update;
     // for sequentiality/progress
     ssx::semaphore _sem;
     std::optional<raft::group_configuration> _config;