Raft: Assign a term during commit rather than execution, to fix view-straddling transaction behaviour

CMakeLists.txt

@@ -716,8 +716,9 @@ if(BUILD_TESTS)
       ${CMAKE_CURRENT_SOURCE_DIR}/src/consensus/aft/test/main.cpp
       ${CMAKE_CURRENT_SOURCE_DIR}/src/consensus/aft/test/view_history.cpp
       ${CMAKE_CURRENT_SOURCE_DIR}/src/consensus/aft/test/committable_suffix.cpp
+      ${CMAKE_CURRENT_SOURCE_DIR}/src/consensus/aft/test/view_straddling_transactions.cpp
     )
-    target_link_libraries(raft_test PRIVATE ccfcrypto ccf_tasks)
+    target_link_libraries(raft_test PRIVATE ccfcrypto ccf_kv ccf_tasks)
 
     add_unit_test(
       raft_enclave_test

src/consensus/aft/raft.h

@@ -621,118 +621,128 @@ namespace aft
       return details;
     }
 
-    bool replicate(const ccf::kv::BatchVector& entries, Term term) override
+    size_t replicate(const ccf::kv::BatchVector& entries) override
     {
       std::lock_guard<ccf::pal::Mutex> guard(state->lock);
 
+      size_t replicated_count = 0;
+
       if (state->leadership_state != ccf::kv::LeadershipState::Leader)
       {
         RAFT_DEBUG_FMT(
           "Failed to replicate {} items: not leader", entries.size());
-        rollback(state->last_idx);
-        return false;
       }
-
-      if (term != state->current_view)
-      {
-        RAFT_DEBUG_FMT(
-          "Failed to replicate {} items at term {}, current term is {}",
-          entries.size(),
-          term,
-          state->current_view);
-        return false;
-      }
-
-      if (is_retired_committed())
+      else if (is_retired_committed())
       {
         RAFT_DEBUG_FMT(
           "Failed to replicate {} items: node retirement is complete",
           entries.size());
-        rollback(state->last_idx);
-        return false;
       }
-
-      RAFT_DEBUG_FMT("Replicating {} entries", entries.size());
-
-      for (const auto& [index, data, is_globally_committable, hooks] : entries)
+      else
       {
-        bool globally_committable = is_globally_committable;
+        RAFT_DEBUG_FMT("Replicating {} entries", entries.size());
 
-        if (index != state->last_idx + 1)
+        for (const auto& [tx_id, data, is_globally_committable, hooks] :
+             entries)
         {
-          return false;
-        }
+          bool globally_committable = is_globally_committable;
 
-        RAFT_DEBUG_FMT(
-          "Replicated on leader {}: {}{} ({} hooks)",
-          state->node_id,
-          index,
-          (globally_committable ? " committable" : ""),
-          hooks->size());
+          if (tx_id.seqno != state->last_idx + 1)
+          {
+            RAFT_DEBUG_FMT(
+              "Received non-contiguous batch: {} != {} + 1",
+              tx_id.seqno,
+              state->last_idx);
+            break;
+          }
+
+          if (tx_id.view != state->current_view)
+          {
+            RAFT_DEBUG_FMT(
+              "Failed to replicate item at {}.{}, current term is {}",
+              tx_id.view,
+              tx_id.seqno,
+              state->current_view);
+            break;
+          }
+
+          RAFT_DEBUG_FMT(
+            "Replicated on leader {}: {}{} ({} hooks)",
+            state->node_id,
+            tx_id.seqno,
+            (globally_committable ? " committable" : ""),
+            hooks->size());
 
 #ifdef CCF_RAFT_TRACING
-        nlohmann::json j = {};
-        j["function"] = "replicate";
-        j["state"] = *state;
-        COMMITTABLE_INDICES(j["state"], state);
-        j["view"] = term;
-        j["seqno"] = index;
-        j["globally_committable"] = globally_committable;
-        RAFT_TRACE_JSON_OUT(j);
+          nlohmann::json j = {};
+          j["function"] = "replicate";
+          j["state"] = *state;
+          COMMITTABLE_INDICES(j["state"], state);
+          j["seqno"] = tx_id.seqno;
+          j["globally_committable"] = globally_committable;
+          RAFT_TRACE_JSON_OUT(j);
 #endif
 
-        for (auto& hook : *hooks)
-        {
-          hook->call(this);
-        }
-
-        if (globally_committable)
-        {
-          RAFT_DEBUG_FMT(
-            "membership: {} leadership: {}",
-            state->membership_state,
-            state->leadership_state);
-          if (
-            state->membership_state == ccf::kv::MembershipState::Retired &&
-            state->retirement_phase == ccf::kv::RetirementPhase::Ordered)
+          for (auto& hook : *hooks)
           {
-            become_retired(index, ccf::kv::RetirementPhase::Signed);
+            hook->call(this);
           }
-          state->committable_indices.push_back(index);
-          start_ticking_if_necessary();
 
-          // Reset should_sign here - whenever we see a committable entry we
-          // don't need to produce _another_ signature
-          should_sign = false;
-        }
+          if (globally_committable)
+          {
+            RAFT_DEBUG_FMT(
+              "membership: {} leadership: {}",
+              state->membership_state,
+              state->leadership_state);
+            if (
+              state->membership_state == ccf::kv::MembershipState::Retired &&
+              state->retirement_phase == ccf::kv::RetirementPhase::Ordered)
+            {
+              become_retired(tx_id.seqno, ccf::kv::RetirementPhase::Signed);
+            }
+            state->committable_indices.push_back(tx_id.seqno);
+            start_ticking_if_necessary();
 
-        state->last_idx = index;
-        ledger->put_entry(
-          *data, globally_committable, state->current_view, index);
-        entry_size_not_limited += data->size();
-        entry_count++;
+            // Reset should_sign here - whenever we see a committable entry we
+            // don't need to produce _another_ signature
+            should_sign = false;
+          }
 
-        state->view_history.update(index, state->current_view);
-        if (entry_size_not_limited >= append_entries_size_limit)
-        {
-          update_batch_size();
-          entry_count = 0;
-          entry_size_not_limited = 0;
-          for (const auto& it : all_other_nodes)
+          state->last_idx = tx_id.seqno;
+          ledger->put_entry(
+            *data, globally_committable, tx_id.view, tx_id.seqno);
+          entry_size_not_limited += data->size();
+          entry_count++;
+
+          state->view_history.update(tx_id.seqno, state->current_view);
+          if (entry_size_not_limited >= append_entries_size_limit)
           {
-            RAFT_DEBUG_FMT("Sending updates to follower {}", it.first);
-            send_append_entries(it.first, it.second.sent_idx + 1);
+            update_batch_size();
+            entry_count = 0;
+            entry_size_not_limited = 0;
+            for (const auto& it : all_other_nodes)
+            {
+              RAFT_DEBUG_FMT("Sending updates to follower {}", it.first);
+              send_append_entries(it.first, it.second.sent_idx + 1);
+            }
           }
+
+          replicated_count++;
+        }
+
+        // Try to advance commit at once if there are no other nodes.
+        if (other_nodes_in_active_configs().size() == 0)
+        {
+          update_commit();
         }
       }
 
-      // Try to advance commit at once if there are no other nodes.
-      if (other_nodes_in_active_configs().size() == 0)
+      if (replicated_count != entries.size())
       {
-        update_commit();
+        rollback(state->last_idx);
       }
 
-      return true;
+      return replicated_count;
     }
 
     void recv_message(

src/consensus/aft/test/committable_suffix.cpp

@@ -214,7 +214,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit")
   DOCTEST_INFO("Entry at 1.1 is received by all nodes");
   {
     auto entry = make_ledger_entry(1, 1);
-    rA.replicate(ccf::kv::BatchVector{{1, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 1}, entry, true, hooks}});
     DOCTEST_REQUIRE(rA.get_last_idx() == 1);
     DOCTEST_REQUIRE(rA.get_committed_seqno() == 0);
     // Size limit was reached, so periodic is not needed
@@ -244,21 +244,21 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit")
     "committed");
   {
     auto entry = make_ledger_entry(1, 2);
-    rA.replicate(ccf::kv::BatchVector{{2, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 2}, entry, true, hooks}});
     DOCTEST_REQUIRE(rA.get_last_idx() == 2);
     DOCTEST_REQUIRE(rA.get_committed_seqno() == 1);
     // Size limit was reached, so periodic is not needed
     // rA.periodic(request_timeout);
 
     entry = make_ledger_entry(1, 3);
-    rA.replicate(ccf::kv::BatchVector{{3, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 3}, entry, true, hooks}});
     DOCTEST_REQUIRE(rA.get_last_idx() == 3);
     DOCTEST_REQUIRE(rA.get_committed_seqno() == 1);
     // Size limit was reached, so periodic is not needed
     // rA.periodic(request_timeout);
 
     entry = make_ledger_entry(1, 4);
-    rA.replicate(ccf::kv::BatchVector{{4, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 4}, entry, true, hooks}});
     DOCTEST_REQUIRE(rA.get_last_idx() == 4);
     DOCTEST_REQUIRE(rA.get_committed_seqno() == 1);
     // Size limit was reached, so periodic is not needed
@@ -292,7 +292,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit")
     "committed");
   {
     auto entry = make_ledger_entry(1, 5);
-    rA.replicate(ccf::kv::BatchVector{{5, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 5}, entry, true, hooks}});
     DOCTEST_REQUIRE(rA.get_last_idx() == 5);
     // Size limit was reached, so periodic is not needed
     // rB.periodic(request_timeout);
@@ -367,11 +367,11 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit")
   DOCTEST_INFO("Node B writes some entries, though they are lost");
   {
     auto entry = make_ledger_entry(2, 6);
-    rB.replicate(ccf::kv::BatchVector{{6, entry, true, hooks}}, 2);
+    rB.replicate(ccf::kv::BatchVector{{ccf::TxID{2, 6}, entry, true, hooks}});
     DOCTEST_REQUIRE(rB.get_last_idx() == 6);
 
     entry = make_ledger_entry(2, 7);
-    rB.replicate(ccf::kv::BatchVector{{7, entry, true, hooks}}, 2);
+    rB.replicate(ccf::kv::BatchVector{{ccf::TxID{2, 7}, entry, true, hooks}});
     DOCTEST_REQUIRE(rB.get_last_idx() == 7);
 
     // Size limit was reached, so periodic is not needed
@@ -426,15 +426,15 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit")
   DOCTEST_REQUIRE("Node C produces 3.5, 3.6, and 3.7");
   {
     auto entry = make_ledger_entry(3, 5);
-    rC.replicate(ccf::kv::BatchVector{{5, entry, true, hooks}}, 3);
+    rC.replicate(ccf::kv::BatchVector{{ccf::TxID{3, 5}, entry, true, hooks}});
     DOCTEST_REQUIRE(rC.get_last_idx() == 5);
 
     entry = make_ledger_entry(3, 6);
-    rC.replicate(ccf::kv::BatchVector{{6, entry, true, hooks}}, 3);
+    rC.replicate(ccf::kv::BatchVector{{ccf::TxID{3, 6}, entry, true, hooks}});
     DOCTEST_REQUIRE(rC.get_last_idx() == 6);
 
     entry = make_ledger_entry(3, 7);
-    rC.replicate(ccf::kv::BatchVector{{7, entry, true, hooks}}, 3);
+    rC.replicate(ccf::kv::BatchVector{{ccf::TxID{3, 7}, entry, true, hooks}});
     DOCTEST_REQUIRE(rC.get_last_idx() == 7);
 
     // The early AppendEntries that describe this are lost
@@ -638,8 +638,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase)
     for (auto idx = start_idx + 1; idx <= start_idx + num_entries; ++idx)
     {
       auto entry = make_ledger_entry(primary.get_view(), idx);
-      primary.replicate(
-        ccf::kv::BatchVector{{idx, entry, true, hooks}}, primary.get_view());
+      primary.replicate(ccf::kv::BatchVector{
+        {ccf::TxID{primary.get_view(), idx}, entry, true, hooks}});
     }
 
     // All related AppendEntries are lost
@@ -668,9 +668,9 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase)
     // be committed
 
     auto entry = make_ledger_entry(1, 1);
-    rA.replicate(ccf::kv::BatchVector{{1, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 1}, entry, true, hooks}});
     entry = make_ledger_entry(1, 2);
-    rA.replicate(ccf::kv::BatchVector{{2, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 2}, entry, true, hooks}});
     DOCTEST_REQUIRE(rA.get_last_idx() == 2);
     DOCTEST_REQUIRE(rA.get_committed_seqno() == 0);
     // Size limit was reached, so periodic is not needed
@@ -703,18 +703,18 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase)
     // Node A produces 2 additional entries that A and B have, and 2 additional
     // entries that are only present on A
     entry = make_ledger_entry(1, 3);
-    rA.replicate(ccf::kv::BatchVector{{3, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 3}, entry, true, hooks}});
 
     entry = make_ledger_entry(1, 4);
-    rA.replicate(ccf::kv::BatchVector{{4, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 4}, entry, true, hooks}});
     keep_messages_for(node_idB, channelsA->messages);
     DOCTEST_REQUIRE(2 == dispatch_all(nodes, node_idA));
 
     entry = make_ledger_entry(1, 5);
-    rA.replicate(ccf::kv::BatchVector{{5, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 5}, entry, true, hooks}});
 
     entry = make_ledger_entry(1, 6);
-    rA.replicate(ccf::kv::BatchVector{{6, entry, true, hooks}}, 1);
+    rA.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 6}, entry, true, hooks}});
     channelsA->messages.clear();
     channelsB->messages.clear();
 
@@ -990,8 +990,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase)
         const auto view = rPrimary.get_view();
         const auto seqno = rPrimary.get_last_idx() + 1;
         auto final_entry = make_ledger_entry(view, seqno);
-        rPrimary.replicate(
-          ccf::kv::BatchVector{{seqno, final_entry, true, hooks}}, view);
+        rPrimary.replicate(ccf::kv::BatchVector{
+          {ccf::TxID{view, seqno}, final_entry, true, hooks}});
 
         rPrimary.periodic(request_timeout);
         keep_earliest_append_entries_for_each_target(channelsPrimary->messages);

src/consensus/aft/test/driver.h

@@ -188,7 +188,8 @@ class RaftDriver
 
     auto s = nlohmann::json(aft::ReplicatedData{type, data}).dump();
     auto d = std::make_shared<std::vector<uint8_t>>(s.begin(), s.end());
-    raft->replicate(ccf::kv::BatchVector{{idx, d, committable, hooks}}, term);
+    raft->replicate(
+      ccf::kv::BatchVector{{ccf::TxID{term, idx}, d, committable, hooks}});
   }
 
   void add_node(ccf::NodeId node_id)