Fix cft election

.azure-pipelines-templates/daily-matrix.yml

@@ -37,7 +37,7 @@ jobs:
       cmake_args: '${{ parameters.build.debug.cmake_args }} ${{ parameters.build.NoSGX.cmake_args }}'
       suffix: 'Instrumented'
       artifact_name: 'NoSGX_Instrumented'
-      ctest_filter: '-LE "benchmark|perf|long_test"'
+      ctest_filter: '-LE "benchmark|perf"'
       ctest_timeout: '300'
 
   - template: common.yml

.azure-pipelines-templates/matrix.yml

@@ -29,9 +29,9 @@ parameters:
 
   test:
     NoSGX:
-      ctest_args: '-LE "benchmark|perf|tlstest|long_test"'
+      ctest_args: '-LE "benchmark|perf|tlstest"'
     SGX:
-      ctest_args: '-LE "benchmark|perf|tlstest|long_test"'
+      ctest_args: '-LE "benchmark|perf|tlstest"'
     perf:
       ctest_args: '-L "benchmark|perf"'
 

CMakeLists.txt

@@ -623,6 +623,20 @@ if(BUILD_TESTS)
       4000
   )
 
+  add_e2e_test(
+    NAME rotation_test
+    PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/rotation.py
+    CONSENSUS cft
+    ADDITIONAL_ARGS --raft-election-timeout 4000
+  )
+
+  add_e2e_test(
+    NAME committable_suffix_test
+    PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/committable.py
+    CONSENSUS cft
+    ADDITIONAL_ARGS --raft-election-timeout 4000
+  )
+
   add_e2e_test(
     NAME lua_e2e_batched
     PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/e2e_batched.py

python/requirements.txt

@@ -1,5 +1,5 @@
 msgpack==1.0.0
-loguru==0.5.2
+loguru==0.5.3
 requests==2.24.0
 requests-http-signature==0.1.0
 websocket-client==0.57.0

src/consensus/aft/raft.h

@@ -74,6 +74,7 @@ namespace aft
 
     ReplicaState replica_state;
     std::chrono::milliseconds timeout_elapsed;
+    kv::Version election_index = 0;
 
     // BFT
     RequestsMap& pbft_requests_map;
@@ -184,6 +185,12 @@ namespace aft
       return replica_state == Follower;
     }
 
+    Index last_committable_index() const
+    {
+      return committable_indices.empty() ? state->commit_idx :
+                                           committable_indices.back();
+    }
+
     void enable_all_domains()
     {
       // When receiving append entries as a follower, all security domains will
@@ -281,6 +288,20 @@ namespace aft
       return {get_term_internal(state->commit_idx), state->commit_idx};
     }
 
+    std::optional<std::pair<Term, Index>> get_signable_commit_term_and_idx()
+    {
+      std::lock_guard<SpinLock> guard(state->lock);
+      if (state->commit_idx >= election_index)
+      {
+        return std::pair<Term, Index>{get_term_internal(state->commit_idx),
+                                      state->commit_idx};
+      }
+      else
+      {
+        return std::nullopt;
+      }
+    }
+
     Term get_term(Index idx)
     {
       if (consensus_type == ConsensusType::BFT && is_follower())
@@ -596,11 +617,12 @@ namespace aft
       }
 
       LOG_DEBUG_FMT(
-        "Received pt: {} pi: {} t: {} i: {}",
+        "Received pt: {} pi: {} t: {} i: {} toi: {}",
         r.prev_term,
         r.prev_idx,
         r.term,
-        r.idx);
+        r.idx,
+        r.term_of_idx);
 
       // Don't check that the sender node ID is valid. Accept anything that
       // passes the integrity check. This way, entries containing dynamic
@@ -621,7 +643,7 @@ namespace aft
       else if (state->current_view > r.term)
       {
         // Reply false, since our term is later than the received term.
-        LOG_DEBUG_FMT(
+        LOG_INFO_FMT(
           "Recv append entries to {} from {} but our term is later ({} > {})",
           state->my_node_id,
           r.from_node,
@@ -775,11 +797,18 @@ namespace aft
           case kv::DeserialiseSuccess::PASS_SIGNATURE:
           {
             LOG_DEBUG_FMT("Deserialising signature at {}", i);
+            auto prev_lci = last_committable_index();
             committable_indices.push_back(i);
 
             if (sig_term)
             {
-              state->view_history.update(state->commit_idx + 1, sig_term);
+              // A signature for sig_term tells us that all transactions from
+              // the previous signature onwards (at least, if not further back)
+              // happened in sig_term. We reflect this in the history.
+              if (r.term_of_idx == aft::ViewHistory::InvalidView)
+                state->view_history.update(1, r.term);
+              else
+                state->view_history.update(prev_lci + 1, sig_term);
               commit_if_possible(r.leader_commit_idx);
             }
             if (consensus_type == ConsensusType::BFT)
@@ -808,7 +837,13 @@ namespace aft
       // After entries have been deserialised, we try to commit the leader's
       // commit index and update our term history accordingly
       commit_if_possible(r.leader_commit_idx);
-      state->view_history.update(state->commit_idx + 1, r.term_of_idx);
+
+      // The term may have changed, and we have not have seen a signature yet.
+      auto lci = last_committable_index();
+      if (r.term_of_idx == aft::ViewHistory::InvalidView)
+        state->view_history.update(1, r.term);
+      else
+        state->view_history.update(lci + 1, r.term_of_idx);
 
       send_append_entries_response(r.from_node, true);
     }
@@ -1168,10 +1203,13 @@ namespace aft
     {
       LOG_INFO_FMT("Send request vote from {} to {}", state->my_node_id, to);
 
+      auto last_committable_idx = last_committable_index();
+      CCF_ASSERT(last_committable_idx >= state->commit_idx, "lci < ci");
+
       RequestVote rv = {{raft_request_vote, state->my_node_id},
                         state->current_view,
-                        state->commit_idx,
-                        get_term_internal(state->commit_idx)};
+                        last_committable_idx,
+                        get_term_internal(last_committable_idx)};
 
       channels->send_authenticated(ccf::NodeMsgType::consensus_msg, to, rv);
     }
@@ -1238,12 +1276,17 @@ namespace aft
         return;
       }
 
-      // If the candidate's log is at least as up-to-date as ours, vote yes
-      auto last_commit_term = get_term_internal(state->commit_idx);
+      // If the candidate's committable log is at least as up-to-date as ours,
+      // vote yes
 
-      auto answer = (r.last_commit_term > last_commit_term) ||
-        ((r.last_commit_term == last_commit_term) &&
-         (r.last_commit_idx >= state->commit_idx));
+      auto last_committable_idx = last_committable_index();
+      auto term_of_last_committable_index =
+        get_term_internal(last_committable_idx);
+
+      auto answer =
+        (r.term_of_last_committable_idx > term_of_last_committable_index) ||
+        ((r.term_of_last_committable_idx == term_of_last_committable_index) &&
+         (r.last_committable_idx >= last_committable_idx));
 
       if (answer)
       {
@@ -1380,20 +1423,21 @@ namespace aft
 
     void become_leader()
     {
-      // Discard any un-committed updates we may hold,
+      election_index = last_committable_index();
+      LOG_DEBUG_FMT("Election index is {}", election_index);
+      // Discard any un-committable updates we may hold,
       // since we have no signature for them. Except at startup,
       // where we do not want to roll back the genesis transaction.
       if (state->commit_idx)
       {
-        rollback(state->commit_idx);
+        rollback(election_index);
       }
       else
       {
         // but we still want the KV to know which term we're in
         store->set_term(state->current_view);
       }
 
-      committable_indices.clear();
       replica_state = Leader;
       leader_id = state->my_node_id;
 
@@ -1433,9 +1477,7 @@ namespace aft
       voted_for = NoNode;
       votes_for_me.clear();
 
-      // Rollback unreplicated commits.
-      rollback(state->commit_idx);
-      committable_indices.clear();
+      rollback(last_committable_index());
 
       LOG_INFO_FMT(
         "Becoming follower {}: {}", state->my_node_id, state->current_view);
@@ -1512,6 +1554,11 @@ namespace aft
 
     void commit_if_possible(Index idx)
     {
+      LOG_DEBUG_FMT(
+        "Commit if possible {} (ci: {}) (ti {})",
+        idx,
+        state->commit_idx,
+        get_term_internal(idx));
       if (
         (idx > state->commit_idx) &&
         (get_term_internal(idx) <= state->current_view))
@@ -1695,4 +1742,4 @@ namespace aft
       }
     }
   };
-}
+}

src/consensus/aft/raft_consensus.h

@@ -72,6 +72,11 @@ namespace aft
       return aft->get_commit_term_and_idx();
     }
 
+    std::optional<std::pair<View, SeqNo>> get_signable_txid() override
+    {
+      return aft->get_signable_commit_term_and_idx();
+    }
+
     View get_view(SeqNo seqno) override
     {
       return aft->get_term(seqno);

src/consensus/aft/raft_types.h

@@ -186,12 +186,8 @@ namespace aft
   struct RequestVote : RaftHeader
   {
     Term term;
-    // last_log_idx in vanilla raft but last_commit_idx here to preserve
-    // verifiability
-    Index last_commit_idx;
-    // last_log_term in vanilla raft but last_commit_term here to preserve
-    // verifiability
-    Term last_commit_term;
+    Index last_committable_idx;
+    Term term_of_last_committable_idx;
   };
 
   struct RequestVoteResponse : RaftHeader

src/consensus/aft/test/driver.h

@@ -88,8 +88,8 @@ class RaftDriver
     aft::NodeId node_id, aft::NodeId tgt_node_id, aft::RequestVote rv)
   {
     std::ostringstream s;
-    s << "request_vote t: " << rv.term << ", lli: " << rv.last_commit_idx
-      << ", llt: " << rv.last_commit_term;
+    s << "request_vote t: " << rv.term << ", lci: " << rv.last_committable_idx
+      << ", tolci: " << rv.term_of_last_committable_idx;
     log(node_id, tgt_node_id, s.str());
   }