Fix cft election

CMakeLists.txt

@@ -529,7 +529,7 @@ if(BUILD_TESTS)
     NAME recovery_test_suite
     PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/e2e_suite.py
     CONSENSUS cft
-    LABEL suite
+    LABEL long_test
     ADDITIONAL_ARGS
       --test-duration
       150
@@ -544,7 +544,7 @@ if(BUILD_TESTS)
     NAME reconfiguration_test_suite
     PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/e2e_suite.py
     CONSENSUS cft
-    LABEL suite
+    LABEL long_test
     ADDITIONAL_ARGS
       --test-duration
       150
@@ -576,7 +576,7 @@ if(BUILD_TESTS)
     NAME full_test_suite
     PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/e2e_suite.py
     CONSENSUS cft
-    LABEL suite
+    LABEL long_test
     ADDITIONAL_ARGS
       --ledger-recovery-timeout
       20
@@ -589,6 +589,13 @@ 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 lua_e2e_batched
     PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/e2e_batched.py

src/consensus/aft/raft.h

@@ -184,6 +184,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
@@ -987,10 +993,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);
     }
@@ -1058,11 +1067,15 @@ namespace aft
       }
 
       // 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);
 
-      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)
       {
@@ -1199,12 +1212,12 @@ namespace aft
 
     void become_leader()
     {
-      // Discard any un-committed updates we may hold,
+      // 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(last_committable_index());
       }
       else
       {
@@ -1252,9 +1265,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);
@@ -1511,4 +1522,4 @@ namespace aft
       }
     }
   };
-}
+}

src/consensus/aft/raft_types.h

@@ -169,12 +169,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());
   }
 

src/consensus/aft/test/main.cpp

@@ -195,8 +195,9 @@ DOCTEST_TEST_CASE(
   DOCTEST_REQUIRE(get<0>(rv) == node_id1);
   auto rvc = get<1>(rv);
   DOCTEST_REQUIRE(rvc.term == 1);
-  DOCTEST_REQUIRE(rvc.last_commit_idx == 0);
-  DOCTEST_REQUIRE(rvc.last_commit_term == aft::ViewHistory::InvalidView);
+  DOCTEST_REQUIRE(rvc.last_committable_idx == 0);
+  DOCTEST_REQUIRE(
+    rvc.term_of_last_committable_idx == aft::ViewHistory::InvalidView);
 
   r1.recv_message(reinterpret_cast<uint8_t*>(&rvc), sizeof(rvc));
 
@@ -207,8 +208,9 @@ DOCTEST_TEST_CASE(
   DOCTEST_REQUIRE(get<0>(rv) == node_id2);
   rvc = get<1>(rv);
   DOCTEST_REQUIRE(rvc.term == 1);
-  DOCTEST_REQUIRE(rvc.last_commit_idx == 0);
-  DOCTEST_REQUIRE(rvc.last_commit_term == aft::ViewHistory::InvalidView);
+  DOCTEST_REQUIRE(rvc.last_committable_idx == 0);
+  DOCTEST_REQUIRE(
+    rvc.term_of_last_committable_idx == aft::ViewHistory::InvalidView);
 
   r2.recv_message(reinterpret_cast<uint8_t*>(&rvc), sizeof(rvc));
 
@@ -1042,12 +1044,9 @@ DOCTEST_TEST_CASE("Exceed append entries limit")
   DOCTEST_REQUIRE(r2.ledger->ledger.size() == individual_entries);
 }
 
-// Reproduces issue described here: https://github.com/microsoft/CCF/issues/521
-// Once this is fixed test will need to be modified since right now it
-// DOCTEST_CHECKs that the issue stands
 DOCTEST_TEST_CASE(
-  "Primary gets invalidated if it compacts right before a term change that it "
-  "doesn't participate in")
+  "Primary is isolated after advancing commit index but before secondaries are "
+  "notified")
 {
   auto kv_store0 = std::make_shared<StoreSig>(0);
   auto kv_store1 = std::make_shared<StoreSig>(1);
@@ -1279,8 +1278,8 @@ DOCTEST_TEST_CASE(
         ((aft::ChannelStubProxy*)r1.channels.get())->sent_request_vote,
         [](const auto& msg) {
           DOCTEST_REQUIRE(msg.term == 2);
-          DOCTEST_REQUIRE(msg.last_commit_idx == 1);
-          DOCTEST_REQUIRE(msg.last_commit_term == 1);
+          DOCTEST_REQUIRE(msg.term_of_last_committable_idx == 1);
+          DOCTEST_REQUIRE(msg.last_committable_idx == 2);
         }));
 
     DOCTEST_INFO("Node 2 votes for Node 1, Node 0 is suspended");
@@ -1305,9 +1304,9 @@ DOCTEST_TEST_CASE(
         nodes,
         ((aft::ChannelStubProxy*)r1.channels.get())->sent_append_entries,
         [](const auto& msg) {
-          DOCTEST_REQUIRE(msg.idx == 1);
+          DOCTEST_REQUIRE(msg.idx == 2);
           DOCTEST_REQUIRE(msg.term == 2);
-          DOCTEST_REQUIRE(msg.prev_idx == 1);
+          DOCTEST_REQUIRE(msg.prev_idx == 2);
           DOCTEST_REQUIRE(msg.prev_term == 1);
           DOCTEST_REQUIRE(msg.leader_commit_idx == 1);
         }));
@@ -1323,12 +1322,22 @@ DOCTEST_TEST_CASE(
           ->sent_append_entries_response));
   }
 
-  DOCTEST_INFO(
-    "Node 1 and Node 2 proceed to compact at idx 2, where Node 0 has "
-    "compacted for a previous term");
+  DOCTEST_CHECK(r0.get_term() == 1);
+  DOCTEST_CHECK(r0.get_commit_idx() == 2);
+  DOCTEST_CHECK(r0.get_last_idx() == 2);
+
+  DOCTEST_CHECK(r1.get_term() == 2);
+  DOCTEST_CHECK(r1.get_commit_idx() == 1);
+  DOCTEST_CHECK(r1.get_last_idx() == 2);
+
+  DOCTEST_CHECK(r2.get_term() == 2);
+  DOCTEST_CHECK(r2.get_commit_idx() == 1);
+  DOCTEST_CHECK(r2.get_last_idx() == 2);
+
+  DOCTEST_INFO("Node 1 resumes replication");
   {
-    DOCTEST_REQUIRE(r1.replicate(kv::BatchVector{{2, second_entry, true}}, 2));
-    DOCTEST_REQUIRE(r1.ledger->ledger.size() == 2);
+    DOCTEST_REQUIRE(r1.replicate(kv::BatchVector{{3, third_entry, true}}, 2));
+    DOCTEST_REQUIRE(r1.ledger->ledger.size() == 3);
     r1.periodic(ms(10));
     DOCTEST_REQUIRE(
       ((aft::ChannelStubProxy*)r1.channels.get())->sent_append_entries.size() ==
@@ -1346,20 +1355,23 @@ DOCTEST_TEST_CASE(
 
     DOCTEST_REQUIRE(
       1 ==
-      dispatch_all(
+      dispatch_all_and_DOCTEST_CHECK(
         nodes,
         ((aft::ChannelStubProxy*)r2.channels.get())
-          ->sent_append_entries_response));
+          ->sent_append_entries_response,
+        [](const auto& msg) { DOCTEST_REQUIRE(msg.success); }));
 
-    // Node 0 will not respond here since it received an append entries it can
-    // not process [prev_idex (1) < commit_idx (2)]
     DOCTEST_REQUIRE(
-      ((aft::ChannelStubProxy*)r0.channels.get())
-        ->sent_append_entries_response.size() == 0);
+      1 ==
+      dispatch_all_and_DOCTEST_CHECK(
+        nodes,
+        ((aft::ChannelStubProxy*)r0.channels.get())
+          ->sent_append_entries_response,
+        [](const auto& msg) { DOCTEST_REQUIRE(msg.success); }));
 
     DOCTEST_INFO("Another entry from Node 1 so that Node 2 can also compact");
-    DOCTEST_REQUIRE(r1.replicate(kv::BatchVector{{3, third_entry, true}}, 2));
-    DOCTEST_REQUIRE(r1.ledger->ledger.size() == 3);
+    DOCTEST_REQUIRE(r1.replicate(kv::BatchVector{{4, third_entry, true}}, 2));
+    DOCTEST_REQUIRE(r1.ledger->ledger.size() == 4);
     r1.periodic(ms(10));
     DOCTEST_REQUIRE(
       ((aft::ChannelStubProxy*)r1.channels.get())->sent_append_entries.size() ==
@@ -1375,31 +1387,16 @@ DOCTEST_TEST_CASE(
       ((aft::ChannelStubProxy*)r1.channels.get())->sent_append_entries.size() ==
       0);
 
-    // Node 0 will now have an ae response which will return false because
-    // its log for index 2 has the wrong term (ours: 1, theirs: 2)
-    DOCTEST_REQUIRE(
-      ((aft::ChannelStubProxy*)r0.channels.get())
-        ->sent_append_entries_response.size() == 1);
-    DOCTEST_REQUIRE(
-      1 ==
-      dispatch_all_and_DOCTEST_CHECK(
-        nodes,
-        ((aft::ChannelStubProxy*)r0.channels.get())
-          ->sent_append_entries_response,
-        [](const auto& msg) {
-          DOCTEST_REQUIRE(msg.last_log_idx == 2);
-          DOCTEST_REQUIRE(!msg.success);
-        }));
-
-    DOCTEST_REQUIRE(r1.ledger->ledger.size() == 3);
-    DOCTEST_REQUIRE(r2.ledger->ledger.size() == 3);
+    DOCTEST_CHECK(r0.get_term() == 2);
+    DOCTEST_CHECK(r0.get_commit_idx() == 3);
+    DOCTEST_CHECK(r0.get_last_idx() == 4);
 
     DOCTEST_CHECK(r1.get_term() == 2);
-    DOCTEST_CHECK(r1.get_commit_idx() == 2);
-    DOCTEST_CHECK(r1.get_last_idx() == 3);
+    DOCTEST_CHECK(r1.get_commit_idx() == 3);
+    DOCTEST_CHECK(r1.get_last_idx() == 4);
 
     DOCTEST_CHECK(r2.get_term() == 2);
-    DOCTEST_CHECK(r2.get_commit_idx() == 2);
-    DOCTEST_CHECK(r2.get_last_idx() == 3);
+    DOCTEST_CHECK(r2.get_commit_idx() == 3);
+    DOCTEST_CHECK(r2.get_last_idx() == 4);
   }
-}
+}

src/node/rpc/member_frontend.h

@@ -793,7 +793,7 @@ namespace ccf
       };
       make_endpoint("read", HTTP_POST, json_adapter(read))
         // This can be executed locally, but can't currently take ReadOnlyTx due
-        // to restristions in our lua wrappers
+        // to restrictions in our lua wrappers
         .set_forwarding_required(ForwardingRequired::Sometimes)
         .set_auto_schema<KVRead>()
         .install();

src/node/rpc/node_frontend.h

@@ -354,6 +354,31 @@ namespace ccf
       make_read_only_endpoint("primary", HTTP_HEAD, is_primary)
         .set_forwarding_required(ForwardingRequired::Never)
         .install();
+
+      auto consensus_config = [this](CommandEndpointContext& args) {
+        // Query node for configurations, separate current from pending
+        if (consensus != nullptr)
+        {
+          auto cfg = consensus->get_latest_configuration();
+          nlohmann::json c;
+          for (auto& [nid, ninfo] : cfg)
+          {
+            nlohmann::json n;
+            n["address"] = fmt::format("{}:{}", ninfo.hostname, ninfo.port);
+            c[fmt::format("{}", nid)] = n;
+          }
+          args.rpc_ctx->set_response_body(c.dump());
+        }
+        else
+        {
+          args.rpc_ctx->set_response_status(HTTP_STATUS_NOT_FOUND);
+          args.rpc_ctx->set_response_body("No configured consensus");
+        }
+      };
+
+      make_command_endpoint("config", HTTP_GET, consensus_config)
+        .set_forwarding_required(ForwardingRequired::Never)
+        .install();
     }
   };
 

tests/infra/network.py

@@ -608,7 +608,7 @@ def find_primary_and_any_backup(self, timeout=3):
         backup = random.choice(backups)
         return primary, backup
 
-    def wait_for_all_nodes_to_catch_up(self, primary, timeout=3):
+    def wait_for_all_nodes_to_catch_up(self, primary, timeout=10):
         """
         Wait for all nodes to have joined the network and globally replicated
         all transactions globally executed on the primary (including transactions