diff --git a/CMakeLists.txt b/CMakeLists.txt index 29298eb683ce..154c9debf7a4 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 9056c818e5ff..3c5f917526cc 100644 --- a/src/consensus/aft/raft.h +++ b/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 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( diff --git a/src/consensus/aft/test/committable_suffix.cpp b/src/consensus/aft/test/committable_suffix.cpp index 3571bedea2e5..6a35c2f8888a 100644 --- a/src/consensus/aft/test/committable_suffix.cpp +++ b/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); diff --git a/src/consensus/aft/test/driver.h b/src/consensus/aft/test/driver.h index aa2693bef970..e556d64d1fa2 100644 --- a/src/consensus/aft/test/driver.h +++ b/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>(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) diff --git a/src/consensus/aft/test/main.cpp b/src/consensus/aft/test/main.cpp index 317ec0668bac..6b1d50841439 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -77,7 +77,7 @@ DOCTEST_TEST_CASE("Single node commit" * doctest::test_suite("single")) entry->push_back(2); entry->push_back(3); - r0.replicate(ccf::kv::BatchVector{{i, entry, true, hooks}}, 1); + r0.replicate(ccf::kv::BatchVector{{ccf::TxID{1, i}, entry, true, hooks}}); DOCTEST_REQUIRE(r0.get_last_idx() == i); DOCTEST_REQUIRE(r0.get_committed_seqno() == i); } @@ -429,11 +429,11 @@ DOCTEST_TEST_CASE( std::vector entry = {1, 2, 3}; auto data = std::make_shared>(entry); DOCTEST_REQUIRE_FALSE( - r1.replicate(ccf::kv::BatchVector{{1, data, true, hooks}}, 1)); + r1.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 1}, data, true, hooks}})); DOCTEST_INFO("Tell the leader to replicate a message"); DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{1, data, true, hooks}}, 1)); + r0.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 1}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 1); // The test ledger adds its own header. Confirm that the expected data is @@ -549,7 +549,7 @@ DOCTEST_TEST_CASE("Multiple nodes late join" * doctest::test_suite("multiple")) std::vector first_entry = {1, 2, 3}; auto data = std::make_shared>(first_entry); DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{1, data, true, hooks}}, 1)); + r0.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 1}, data, true, hooks}})); r0.periodic(request_timeout); DOCTEST_REQUIRE( @@ -662,10 +662,10 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) std::vector second_entry = {2, 2, 2}; auto data_2 = std::make_shared>(second_entry); - DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{1, data_1, true, hooks}}, 1)); - DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{2, data_2, true, hooks}}, 1)); + DOCTEST_REQUIRE(r0.replicate( + ccf::kv::BatchVector{{ccf::TxID{1, 1}, data_1, true, hooks}})); + DOCTEST_REQUIRE(r0.replicate( + ccf::kv::BatchVector{{ccf::TxID{1, 2}, data_2, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 2); r0.periodic(request_timeout); DOCTEST_REQUIRE(r0c->messages.size() == 1); @@ -687,7 +687,7 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) std::vector third_entry = {3, 3, 3}; auto data = std::make_shared>(third_entry); DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{3, data, true, hooks}}, 1)); + r0.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 3}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 3); // Simulate that the append entries was not deserialised successfully @@ -720,7 +720,7 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) std::vector fourth_entry = {4, 4, 4}; auto data = std::make_shared>(fourth_entry); DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{4, data, true, hooks}}, 1)); + r0.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 4}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 4); r0.periodic(request_timeout); DOCTEST_REQUIRE(r0c->messages.size() == 1); @@ -734,7 +734,7 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) std::vector fifth_entry = {5, 5, 5}; auto data = std::make_shared>(fifth_entry); DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{5, data, true, hooks}}, 1)); + r0.replicate(ccf::kv::BatchVector{{ccf::TxID{1, 5}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 5); r0.periodic(request_timeout); DOCTEST_REQUIRE(r0c->messages.size() == 1); @@ -763,8 +763,8 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) { std::vector entry_6 = {6, 6, 6}; auto data = std::make_shared>(entry_6); - DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{6, data, true, hooks}}, 1)); + DOCTEST_REQUIRE(r0.replicate( + ccf::kv::BatchVector{{ccf::TxID{1, 6}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 6); } const auto last_correct_version = r0.ledger->ledger.size(); @@ -773,8 +773,8 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) { std::vector entry_7 = {7, 7, 7}; auto data = std::make_shared>(entry_7); - DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{7, data, true, hooks}}, 1)); + DOCTEST_REQUIRE(r0.replicate( + ccf::kv::BatchVector{{ccf::TxID{1, 7}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 7); dead_branch = r0.ledger->ledger.back(); } @@ -795,8 +795,8 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) { std::vector entry_7b = {7, 7, 'b'}; auto data = std::make_shared>(entry_7b); - DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{7, data, true, hooks}}, 4)); + DOCTEST_REQUIRE(r0.replicate( + ccf::kv::BatchVector{{ccf::TxID{4, 7}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 7); live_branch = r0.ledger->ledger.back(); } @@ -804,8 +804,8 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) { std::vector entry_8 = {8, 8, 8}; auto data = std::make_shared>(entry_8); - DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{8, data, true, hooks}}, 4)); + DOCTEST_REQUIRE(r0.replicate( + ccf::kv::BatchVector{{ccf::TxID{4, 8}, data, true, hooks}})); DOCTEST_REQUIRE(r0.ledger->ledger.size() == 8); DOCTEST_REQUIRE(r0.ledger->ledger.size() > last_correct_version); } @@ -937,7 +937,7 @@ DOCTEST_TEST_CASE("Exceed append entries limit") for (size_t i = 1; i <= static_cast(num_big_entries); ++i) { DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{i, data, true, hooks}}, 1)); + r0.replicate(ccf::kv::BatchVector{{ccf::TxID{1, i}, data, true, hooks}})); const auto received_ae = dispatch_all_and_DOCTEST_CHECK( nodes, node_id0, r0c->messages, [](const auto& msg) { @@ -955,8 +955,8 @@ DOCTEST_TEST_CASE("Exceed append entries limit") i <= static_cast(individual_entries); ++i) { - DOCTEST_REQUIRE( - r0.replicate(ccf::kv::BatchVector{{i, smaller_data, true, hooks}}, 1)); + DOCTEST_REQUIRE(r0.replicate( + ccf::kv::BatchVector{{ccf::TxID{1, i}, smaller_data, true, hooks}})); dispatch_all(nodes, node_id0, r0c->messages); } diff --git a/src/consensus/aft/test/view_straddling_transactions.cpp b/src/consensus/aft/test/view_straddling_transactions.cpp new file mode 100644 index 000000000000..a60b13326163 --- /dev/null +++ b/src/consensus/aft/test/view_straddling_transactions.cpp @@ -0,0 +1,487 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the Apache 2.0 License. + +#include "kv/store.h" +#include "kv/test/null_encryptor.h" +#include "test_common.h" + +#include +#include +#include +#include + +namespace +{ + using TestMap = ccf::kv::Map; + using Raft = aft::Aft; + + class BaselinePendingTx : public ccf::kv::PendingTx + { + ccf::TxID txid; + ccf::kv::Store& store; + TestMap& table; + + public: + BaselinePendingTx( + ccf::TxID txid_, ccf::kv::Store& store_, TestMap& table_) : + txid(txid_), + store(store_), + table(table_) + {} + + ccf::kv::PendingTxInfo call() override + { + auto tx = store.create_reserved_tx(txid); + tx.rw(table)->put(0, 1); + return tx.commit_reserved(); + } + }; + + struct CommitPause + { + std::mutex lock; + std::condition_variable paused_cv; + std::condition_variable resume_cv; + bool paused = false; + bool resume = false; + + void pause() + { + { + std::lock_guard guard(lock); + paused = true; + } + paused_cv.notify_one(); + + std::unique_lock guard(lock); + resume_cv.wait(guard, [this]() { return resume; }); + } + + void wait_until_paused() + { + std::unique_lock guard(lock); + paused_cv.wait(guard, [this]() { return paused; }); + } + + void release() + { + { + std::lock_guard guard(lock); + resume = true; + } + resume_cv.notify_one(); + } + }; + + static std::optional read_value( + ccf::kv::Store& store, TestMap& table, size_t key) + { + auto tx = store.create_read_only_tx(); + return tx.ro(table)->get(key); + } + + struct Fixture + { + const ccf::NodeId node_id = ccf::kv::test::PrimaryNodeId; + std::shared_ptr store = std::make_shared(); + TestMap table{"public:table"}; + std::shared_ptr raft; + ccf::View initial_view = 0; + + Fixture() + { + store->set_encryptor(std::make_shared()); + raft = std::make_shared( + raft_settings, + std::make_unique>(store), + std::make_unique(node_id), + std::make_shared(), + std::make_shared(node_id), + nullptr); + store->set_consensus(raft); + + ccf::kv::Configuration::Nodes configuration; + configuration.try_emplace(node_id); + raft->add_configuration(0, configuration); + raft->force_become_primary(); + initial_view = raft->get_view(); + + const auto baseline_txid = store->next_txid(); + REQUIRE( + store->commit( + baseline_txid, + std::make_unique(baseline_txid, *store, table), + true) == ccf::kv::CommitResult::SUCCESS); + REQUIRE(store->current_txid() == ccf::TxID(initial_view, 1)); + REQUIRE(raft->get_committed_seqno() == 1); + REQUIRE(raft->ledger->ledger.size() == 1); + } + + void step_down() + { + const auto next_view = raft->get_view() + 1; + raft->become_aware_of_new_term(next_view); + } + + ccf::View reelect() + { + step_down(); + raft->force_become_primary(); + return raft->get_view(); + } + }; + + static ccf::kv::BatchVector::value_type make_entry( + const ccf::TxID& tx_id, bool globally_committable = false) + { + return { + tx_id, + std::make_shared>(16, tx_id.seqno), + globally_committable, + std::make_shared()}; + } +} + +TEST_CASE( + "Long-lived transaction is rolled back after leadership loss" * + doctest::test_suite("view_straddling_transactions")) +{ + Fixture fixture; + + INFO("Start applying a local transaction in the initial view"); + auto stale_tx = fixture.store->create_tx(); + stale_tx.rw(fixture.table)->put(1, 2); + + CommitPause pause; + std::optional stale_result; + std::thread stale_worker([&]() { + stale_result = + stale_tx.commit(ccf::empty_claims(), [&pause](const auto&, const auto&) { + pause.pause(); + }); + }); + pause.wait_until_paused(); + REQUIRE(stale_tx.get_txid() == ccf::TxID(fixture.initial_view, 2)); + + INFO("Step down after the transaction has been assigned its old-view TxID"); + fixture.step_down(); + + INFO("AFT rejects the transaction and rolls Store back to the baseline"); + pause.release(); + stale_worker.join(); + REQUIRE(stale_result.has_value()); + REQUIRE(stale_result.value() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + REQUIRE(fixture.store->current_txid() == ccf::TxID(fixture.initial_view, 1)); + REQUIRE_FALSE(read_value(*fixture.store, fixture.table, 1).has_value()); + REQUIRE(fixture.raft->get_last_idx() == 1); + REQUIRE(fixture.raft->ledger->ledger.size() == 1); + + INFO("Win a later election and replicate the next transaction normally"); + fixture.raft->force_become_primary(); + const auto fresh_view = fixture.raft->get_view(); + auto fresh_tx = fixture.store->create_tx(); + fresh_tx.rw(fixture.table)->put(2, 3); + REQUIRE(fresh_tx.commit() == ccf::kv::CommitResult::SUCCESS); + REQUIRE(fixture.store->current_txid() == ccf::TxID(fresh_view, 2)); + REQUIRE(read_value(*fixture.store, fixture.table, 2) == 3); + REQUIRE(fixture.raft->get_last_idx() == 2); + REQUIRE(fixture.raft->ledger->ledger.size() == 2); +} + +TEST_CASE( + "Read-only transaction can finish after re-election" * + doctest::test_suite("view_straddling_transactions")) +{ + Fixture fixture; + + INFO("Read the baseline in the initial view"); + auto read_tx = fixture.store->create_tx(); + REQUIRE(read_tx.ro(fixture.table)->get(0) == 1); + const auto read_txid = fixture.store->current_txid(); + + INFO("Lose leadership and win a later election before finishing the read"); + fixture.reelect(); + + INFO("The read remains valid at the TxID where it observed state"); + REQUIRE(read_tx.commit() == ccf::kv::CommitResult::SUCCESS); + REQUIRE(read_tx.get_txid() == read_txid); + CHECK(fixture.store->current_txid() == read_txid); + CHECK(fixture.raft->get_last_idx() == 1); + CHECK(fixture.raft->ledger->ledger.size() == 1); +} + +TEST_CASE( + "Transaction begun before re-election can commit in the new view" * + doctest::test_suite("view_straddling_transactions")) +{ + Fixture fixture; + + INFO("Read state and prepare writes in the initial view"); + auto tx = fixture.store->create_tx(); + auto handle = tx.rw(fixture.table); + REQUIRE(handle->get(0) == 1); + handle->put(1, 2); + + INFO("Win a later election before assigning the transaction a TxID"); + const auto reelection_view = fixture.reelect(); + + INFO("Revalidate the read and assign the write a TxID in the new view"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + REQUIRE(tx.get_txid() == ccf::TxID(reelection_view, 2)); + CHECK(fixture.store->current_txid() == ccf::TxID(reelection_view, 2)); + CHECK(read_value(*fixture.store, fixture.table, 1) == 2); + CHECK(fixture.raft->get_last_idx() == 2); + CHECK(fixture.raft->ledger->ledger.size() == 2); +} + +TEST_CASE( + "Transaction conflicts when re-election rolls back its read snapshot" * + doctest::test_suite("view_straddling_transactions")) +{ + Fixture fixture; + + INFO("Read state and prepare writes in the initial view"); + auto tx = fixture.store->create_tx(); + auto handle = tx.rw(fixture.table); + REQUIRE(handle->get(0) == 1); + handle->put(1, 2); + + INFO("Create an unreplicated local suffix after the transaction's read"); + REQUIRE(fixture.store->next_txid() == ccf::TxID(fixture.initial_view, 2)); + auto suffix_tx = fixture.store->create_tx(); + suffix_tx.rw(fixture.table)->put(0, 9); + REQUIRE(suffix_tx.commit() == ccf::kv::CommitResult::SUCCESS); + REQUIRE(suffix_tx.get_txid() == ccf::TxID(fixture.initial_view, 3)); + REQUIRE(fixture.raft->get_last_idx() == 1); + + INFO("Win a later election, rolling the local suffix back"); + fixture.reelect(); + REQUIRE(read_value(*fixture.store, fixture.table, 0) == 1); + + INFO("The transaction's pre-rollback change set is no longer valid"); + CHECK(tx.commit() == ccf::kv::CommitResult::FAIL_CONFLICT); + CHECK(fixture.store->current_txid() == ccf::TxID(fixture.initial_view, 1)); + CHECK_FALSE(read_value(*fixture.store, fixture.table, 1).has_value()); + CHECK(fixture.raft->get_last_idx() == 1); + CHECK(fixture.raft->ledger->ledger.size() == 1); +} + +TEST_CASE( + "Assigned old-view transaction is rolled back after re-election" * + doctest::test_suite("view_straddling_transactions")) +{ + Fixture fixture; + + INFO("Apply a transaction and assign its TxID in the initial view"); + auto stale_tx = fixture.store->create_tx(); + stale_tx.rw(fixture.table)->put(1, 2); + CommitPause pause; + std::optional stale_result; + std::thread stale_worker([&]() { + stale_result = + stale_tx.commit(ccf::empty_claims(), [&pause](const auto&, const auto&) { + pause.pause(); + }); + }); + pause.wait_until_paused(); + REQUIRE(stale_tx.get_txid() == ccf::TxID(fixture.initial_view, 2)); + + INFO("Lose leadership and win a later election before replication"); + const auto reelection_view = fixture.reelect(); + + INFO("AFT rejects the assigned old-view transaction"); + pause.release(); + stale_worker.join(); + REQUIRE(stale_result.has_value()); + REQUIRE(stale_result.value() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + CHECK(fixture.store->current_txid() == ccf::TxID(fixture.initial_view, 1)); + CHECK_FALSE(read_value(*fixture.store, fixture.table, 1).has_value()); + CHECK(fixture.raft->get_last_idx() == 1); + CHECK(fixture.raft->ledger->ledger.size() == 1); + + INFO("Replicate a fresh transaction at the next index in the new view"); + auto fresh_tx = fixture.store->create_tx(); + fresh_tx.rw(fixture.table)->put(2, 3); + CHECK(fresh_tx.commit() == ccf::kv::CommitResult::SUCCESS); + CHECK(fixture.store->current_txid() == ccf::TxID(reelection_view, 2)); + CHECK(read_value(*fixture.store, fixture.table, 2) == 3); + CHECK(fixture.raft->get_last_idx() == 2); + CHECK(fixture.raft->ledger->ledger.size() == 2); +} + +TEST_CASE( + "Rolled-back stale transaction cannot invalidate current-view work" * + doctest::test_suite("view_straddling_transactions")) +{ + Fixture fixture; + + INFO( + "Assign an old-view transaction seqno 2, then pause before Store::commit"); + auto stale_tx = fixture.store->create_tx(); + stale_tx.rw(fixture.table)->put(1, 2); + CommitPause stale_pause; + std::optional stale_result; + std::thread stale_worker([&]() { + stale_result = stale_tx.commit( + ccf::empty_claims(), + [&stale_pause](const auto&, const auto&) { stale_pause.pause(); }); + }); + stale_pause.wait_until_paused(); + REQUIRE(stale_tx.get_txid() == ccf::TxID(fixture.initial_view, 2)); + + INFO("Win a later election, reclaiming seqno 2"); + const auto reelection_view = fixture.reelect(); + + INFO("Assign current-view seqno 2, then pause before Store::commit"); + auto current_head = fixture.store->create_tx(); + current_head.rw(fixture.table)->put(2, 3); + CommitPause current_pause; + std::optional current_result; + std::thread current_worker([&]() { + current_result = current_head.commit( + ccf::empty_claims(), + [¤t_pause](const auto&, const auto&) { current_pause.pause(); }); + }); + current_pause.wait_until_paused(); + REQUIRE(current_head.get_txid() == ccf::TxID(reelection_view, 2)); + + INFO("Queue current-view seqno 3 behind the missing seqno 2"); + auto current_suffix = fixture.store->create_tx(); + current_suffix.rw(fixture.table)->put(3, 4); + REQUIRE(current_suffix.commit() == ccf::kv::CommitResult::SUCCESS); + REQUIRE(current_suffix.get_txid() == ccf::TxID(reelection_view, 3)); + + INFO("Resume old 2.2; Store must reject its invalidated local application"); + stale_pause.release(); + stale_worker.join(); + REQUIRE(stale_result.has_value()); + CHECK(stale_result.value() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + + INFO("Resume current 3.2, which can now replicate with pending 3.3"); + current_pause.release(); + current_worker.join(); + REQUIRE(current_result.has_value()); + CHECK(current_result.value() == ccf::kv::CommitResult::SUCCESS); + + CHECK(fixture.store->current_txid() == ccf::TxID(reelection_view, 3)); + CHECK_FALSE(read_value(*fixture.store, fixture.table, 1).has_value()); + CHECK(read_value(*fixture.store, fixture.table, 2) == 3); + CHECK(read_value(*fixture.store, fixture.table, 3) == 4); + CHECK(fixture.raft->get_last_idx() == 3); + CHECK(fixture.raft->ledger->ledger.size() == 3); + + INFO("Replicate a fresh new-view transaction at seqno 4"); + auto fresh_tx = fixture.store->create_tx(); + fresh_tx.rw(fixture.table)->put(4, 5); + CHECK(fresh_tx.commit() == ccf::kv::CommitResult::SUCCESS); + CHECK(fixture.store->current_txid() == ccf::TxID(reelection_view, 4)); + CHECK(read_value(*fixture.store, fixture.table, 4) == 5); + CHECK(fixture.raft->get_last_idx() == 4); + CHECK(fixture.raft->ledger->ledger.size() == 4); +} + +TEST_CASE( + "Rolled-back stale suffix cannot block a current-view prefix" * + doctest::test_suite("view_straddling_transactions")) +{ + Fixture fixture; + + INFO("Reserve old-view seqno 2, leaving a hole before stale seqno 3"); + REQUIRE(fixture.store->next_txid() == ccf::TxID(fixture.initial_view, 2)); + + auto stale_suffix = fixture.store->create_tx(); + stale_suffix.rw(fixture.table)->put(1, 2); + CommitPause stale_pause; + std::optional stale_result; + std::thread stale_worker([&]() { + stale_result = stale_suffix.commit( + ccf::empty_claims(), + [&stale_pause](const auto&, const auto&) { stale_pause.pause(); }); + }); + stale_pause.wait_until_paused(); + REQUIRE(stale_suffix.get_txid() == ccf::TxID(fixture.initial_view, 3)); + + INFO("Win a later election, rolling back the old reservation and write"); + const auto reelection_view = fixture.reelect(); + + INFO( + "Apply the current-view transaction at seqno 2, then pause before " + "Store::commit"); + auto current_tx = fixture.store->create_tx(); + current_tx.rw(fixture.table)->put(2, 3); + CommitPause current_pause; + std::optional current_result; + std::thread current_worker([&]() { + current_result = current_tx.commit( + ccf::empty_claims(), + [¤t_pause](const auto&, const auto&) { current_pause.pause(); }); + }); + current_pause.wait_until_paused(); + REQUIRE(current_tx.get_txid() == ccf::TxID(reelection_view, 2)); + + INFO("Resume stale 2.3; Store must reject its invalidated local application"); + stale_pause.release(); + stale_worker.join(); + REQUIRE(stale_result.has_value()); + CHECK(stale_result.value() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + + INFO("Resume seqno 2; AFT can accept the current-view transaction"); + current_pause.release(); + current_worker.join(); + + REQUIRE(current_result.has_value()); + CHECK(current_result.value() == ccf::kv::CommitResult::SUCCESS); + CHECK(fixture.store->current_txid() == ccf::TxID(reelection_view, 2)); + CHECK_FALSE(read_value(*fixture.store, fixture.table, 1).has_value()); + CHECK(read_value(*fixture.store, fixture.table, 2) == 3); + CHECK(fixture.raft->get_last_idx() == 2); + CHECK(fixture.raft->ledger->ledger.size() == 2); + + INFO("Replicate the next new-view transaction at seqno 3"); + auto fresh_tx = fixture.store->create_tx(); + fresh_tx.rw(fixture.table)->put(3, 4); + CHECK(fresh_tx.commit() == ccf::kv::CommitResult::SUCCESS); + CHECK(fixture.store->current_txid() == ccf::TxID(reelection_view, 3)); + CHECK(read_value(*fixture.store, fixture.table, 3) == 4); + CHECK(fixture.raft->get_last_idx() == 3); + CHECK(fixture.raft->ledger->ledger.size() == 3); +} + +TEST_CASE( + "AFT accepts a current-view prefix before a stale suffix" * + doctest::test_suite("view_straddling_transactions")) +{ + const ccf::NodeId node_id = ccf::kv::test::PrimaryNodeId; + auto store = std::make_shared(node_id); + Raft raft( + raft_settings, + std::make_unique>(store), + std::make_unique(node_id), + std::make_shared(), + std::make_shared(node_id), + nullptr); + + ccf::kv::Configuration::Nodes configuration; + configuration.try_emplace(node_id); + raft.add_configuration(0, configuration); + raft.force_become_primary(); + + const auto initial_view = raft.get_view(); + REQUIRE(raft.replicate({make_entry({initial_view, 1}, true)})); + REQUIRE(raft.get_committed_seqno() == 1); + + raft.become_aware_of_new_term(initial_view + 1); + raft.force_become_primary(); + const auto current_view = raft.get_view(); + + INFO("Submit current 3.2 followed by stale 2.3 in one candidate batch"); + CHECK(raft.replicate( + {make_entry({current_view, 2}), make_entry({initial_view, 3})})); + CHECK(raft.get_last_idx() == 2); + CHECK(raft.ledger->ledger.size() == 2); + + INFO("The next current-view entry can reuse seqno 3"); + CHECK(raft.replicate({make_entry({current_view, 3})})); + CHECK(raft.get_last_idx() == 3); + CHECK(raft.ledger->ledger.size() == 3); +} \ No newline at end of file diff --git a/src/indexing/test/common.h b/src/indexing/test/common.h index 9c65339b4458..fc14d0830122 100644 --- a/src/indexing/test/common.h +++ b/src/indexing/test/common.h @@ -87,17 +87,17 @@ class AllCommittableWrapper : public TConsensus public: using TConsensus::TConsensus; - bool replicate(const ccf::kv::BatchVector& entries_, ccf::View view) override + size_t replicate(const ccf::kv::BatchVector& entries_) override { // Rather than building a history that produces real signatures, we just // overwrite the entries here to say that everything is committable ccf::kv::BatchVector entries(entries_); - for (auto& [seqno, data, committable, hooks] : entries) + for (auto& [tx_id, data, committable, hooks] : entries) { committable = true; } - return TConsensus::replicate(entries, view); + return TConsensus::replicate(entries); } }; diff --git a/src/kv/apply_changes.h b/src/kv/apply_changes.h index dc64c5dd3dc1..a387b8dbb4ab 100644 --- a/src/kv/apply_changes.h +++ b/src/kv/apply_changes.h @@ -16,17 +16,16 @@ namespace ccf::kv using MapCollection = std::map>; // Atomically checks for conflicts then applies the writes in the given change - // sets to their underlying Maps. Calls f() at most once, iff the writes are - // applied, to retrieve a unique Version for the write set and return the max - // version which can have a conflict with the transaction. + // sets to their underlying Maps. Calls tx_id_resolver() at most once, iff the + // writes are applied, to retrieve a unique TxID for the write set. + // Returns std::nullopt on conflict, a default TxID on successful application + // with no writes, or the assigned TxID on successful application with writes. - using VersionLastNewMap = Version; - using VersionResolver = std::function( - bool tx_contains_new_map)>; + using TxIDResolver = std::function; - static inline std::optional apply_changes( + static inline std::optional apply_changes( OrderedChanges& changes, - VersionResolver version_resolver_fn, + TxIDResolver tx_id_resolver, ccf::kv::ConsensusHookPtrs& hooks, const MapCollection& new_maps, const std::optional& new_maps_conflict_version, @@ -37,7 +36,7 @@ namespace ccf::kv // and possibly committed, and then all maps with pending writes are // unlocked. This is to prevent transactions from being committed in an // interleaved fashion. - Version version = NoVersion; + ccf::TxID tx_id; bool has_writes = false; std::map> views; @@ -117,9 +116,7 @@ namespace ccf::kv if (ok && has_writes) { // Get the version number to be used for this commit. - ccf::kv::Version version_last_new_map = 0; - std::tie(version, version_last_new_map) = - version_resolver_fn(!new_maps.empty()); + tx_id = tx_id_resolver(); // Transfer ownership of these new maps to their target stores, iff we // have writes to them @@ -128,13 +125,13 @@ namespace ccf::kv const auto it = views.find(map_name); if (it != views.end() && it->second->has_writes()) { - map_ptr->get_store()->add_dynamic_map(version, map_ptr); + map_ptr->get_store()->add_dynamic_map(tx_id.seqno, map_ptr); } } for (auto& [view_name, view_ptr] : views) { - view_ptr->commit(version, track_deletes_on_missing_keys); + view_ptr->commit(tx_id.seqno, track_deletes_on_missing_keys); } // Collect ConsensusHooks @@ -158,6 +155,6 @@ namespace ccf::kv return std::nullopt; } - return version; + return tx_id; } } diff --git a/src/kv/committable_tx.h b/src/kv/committable_tx.h index e47877e983bd..bf50ec6d09a3 100644 --- a/src/kv/committable_tx.h +++ b/src/kv/committable_tx.h @@ -28,10 +28,17 @@ namespace ccf::kv }; protected: + // Indicates that this transaction's changes have been applied to the local + // KV. Monotonic and gating, to form a crude linear type - some functions + // are available only pre-commit, some only post-commit. bool committed = false; - bool success = false; - Version version = NoVersion; + // The TxID at which this transaction was applied to the local KV. A + // successful transaction that never acquired a map handle has no changes + // and no TxID; its legacy commit metadata is represented by NoVersion and + // VIEW_UNKNOWN. A committed transaction with changes but no applied TxID + // was aborted. + std::optional applied_txid = std::nullopt; TxFlags flags = 0; SerialisedEntryFlags entry_flags = 0; @@ -78,7 +85,9 @@ namespace ccf::kv SizeKvStoreSerialiser size_serialiser( e, - TxID{pimpl->commit_view, NoVersion}, + // Used as IV for encrypted serialisation, but does not affect the + // projected size. + ccf::TxID{0, 0}, EntryType::WriteSetWithCommitEvidenceAndClaims, entry_flags, // Both digests are fixed-size, so their values do not affect the @@ -103,7 +112,7 @@ namespace ccf::kv throw std::logic_error("Transaction not yet committed"); } - if (!success) + if (!applied_txid.has_value()) { throw std::logic_error("Transaction aborted"); } @@ -124,7 +133,7 @@ namespace ccf::kv throw KvSerialiserException("No encryptor set"); } - commit_evidence = e->get_commit_evidence({pimpl->commit_view, version}); + commit_evidence = e->get_commit_evidence(*applied_txid); LOG_TRACE_FMT("Commit evidence: {}", commit_evidence); ccf::crypto::Sha256Hash tx_commit_evidence_digest(commit_evidence); commit_evidence_digest = tx_commit_evidence_digest; @@ -136,7 +145,7 @@ namespace ccf::kv RawKvStoreSerialiser serialiser( e, - {pimpl->commit_view, version}, + *applied_txid, EntryType::WriteSetWithCommitEvidenceAndClaims, entry_flags, tx_commit_evidence_digest, @@ -170,8 +179,6 @@ namespace ccf::kv */ CommitResult commit( const ccf::ClaimsDigest& claims = ccf::empty_claims(), - std::function(bool has_new_map)> - version_resolver = nullptr, WriteSetObserver write_set_observer = nullptr) { if (committed) @@ -182,7 +189,6 @@ namespace ccf::kv if (all_changes.empty()) { committed = true; - success = true; return CommitResult::SUCCESS; } @@ -216,13 +222,11 @@ namespace ccf::kv std::optional new_maps_conflict_version = std::nullopt; bool track_deletes_on_missing_keys = false; - auto c = apply_changes( + auto txid_resolver = [&]() { return this->pimpl->store->next_txid(); }; + + applied_txid = apply_changes( all_changes, - version_resolver == nullptr ? - [&](bool has_new_map) { - return pimpl->store->next_version(has_new_map); - } : - version_resolver, + txid_resolver, hooks, pimpl->created_maps, new_maps_conflict_version, @@ -233,9 +237,7 @@ namespace ccf::kv this->pimpl->store->unlock_map_set(); } - success = c.has_value(); - - if (!success) + if (!applied_txid.has_value()) { // This Tx is now in a dead state. Caller should create a new Tx and try // again. @@ -244,14 +246,13 @@ namespace ccf::kv } committed = true; - version = c.value(); if (tx_flag_enabled(TxFlag::LEDGER_CHUNK_AT_NEXT_SIGNATURE)) { auto chunker = pimpl->store->get_chunker(); if (chunker) { - chunker->force_end_of_chunk(version); + chunker->force_end_of_chunk(applied_txid->seqno); } } @@ -262,9 +263,10 @@ namespace ccf::kv unset_tx_flag(TxFlag::SNAPSHOT_AT_NEXT_SIGNATURE); } - if (version == NoVersion) + if (applied_txid->seqno == NoVersion) { // Read-only transaction + applied_txid = pimpl->read_txid; return CommitResult::SUCCESS; } @@ -296,7 +298,7 @@ namespace ccf::kv auto claims_ = claims; return pimpl->store->commit( - {pimpl->commit_view, version}, + *applied_txid, std::make_unique( std::move(data), std::move(claims_), @@ -331,12 +333,17 @@ namespace ccf::kv throw std::logic_error("Transaction not yet committed"); } - if (!success) + if (!applied_txid.has_value()) { + if (all_changes.empty()) + { + return NoVersion; + } + throw std::logic_error("Transaction aborted"); } - return version; + return applied_txid->seqno; } /** Get term in which this transaction was committed. @@ -346,19 +353,24 @@ namespace ccf::kv * * @return Commit term */ - [[nodiscard]] Version commit_term() const + [[nodiscard]] Term commit_term() const { if (!committed) { throw std::logic_error("Transaction not yet committed"); } - if (!success) + if (!applied_txid.has_value()) { + if (all_changes.empty()) + { + return ccf::VIEW_UNKNOWN; + } + throw std::logic_error("Transaction aborted"); } - return pimpl->commit_view; + return applied_txid->view; } [[nodiscard]] std::optional get_txid() const @@ -368,32 +380,16 @@ namespace ccf::kv throw std::logic_error("Transaction not yet committed"); } - if (!pimpl->read_txid.has_value()) - { - // Transaction did not get a handle on any map. - return std::nullopt; - } - - // A committed tx is read-only (i.e. no write to any map) if it was not - // assigned a version when it was committed - if (version == NoVersion) - { - // Read-only transaction - return pimpl->read_txid; - } - - // Write transaction - return TxID(pimpl->commit_view, version); + return applied_txid; } - void set_read_txid(const TxID& tx_id, Term commit_view_) + void set_read_txid(const TxID& tx_id) { if (pimpl->read_txid.has_value()) { throw std::logic_error("Read TxID already set"); } pimpl->read_txid = tx_id; - pimpl->commit_view = commit_view_; } void set_root_at_read_version(const ccf::crypto::Sha256Hash& r) @@ -428,19 +424,19 @@ namespace ccf::kv { private: Version rollback_count = 0; + const TxID reserved_txid; public: ReservedTx( AbstractStore* _store, Term read_term, - const TxID& reserved_tx_id, + const TxID& reserved_txid_, Version rollback_count_) : CommittableTx(_store), - rollback_count(rollback_count_) + rollback_count(rollback_count_), + reserved_txid(reserved_txid_) { - version = reserved_tx_id.seqno; - pimpl->commit_view = reserved_tx_id.view; - pimpl->read_txid = TxID(read_term, reserved_tx_id.seqno - 1); + pimpl->read_txid = TxID(read_term, reserved_txid.seqno - 1); } // Used by frontend to commit reserved transactions @@ -458,15 +454,15 @@ namespace ccf::kv std::vector hooks; bool track_deletes_on_missing_keys = false; - auto c = apply_changes( + applied_txid = apply_changes( all_changes, - [this](bool) { return std::make_tuple(version, version - 1); }, + [this]() { return reserved_txid; }, hooks, pimpl->created_maps, - version, + reserved_txid.seqno, track_deletes_on_missing_keys, rollback_count); - success = c.has_value(); + const auto success = applied_txid.has_value(); if (!success) { @@ -486,18 +482,18 @@ namespace ccf::kv // This is a signature and, if the ledger chunking or snapshot flags are // enabled, we want the host to create a chunk when it sees this entry. // version_lock held by Store::commit - if (pimpl->store->should_create_ledger_chunk_unsafe(version)) + if (pimpl->store->should_create_ledger_chunk_unsafe(applied_txid->seqno)) { entry_flags |= EntryFlags::FORCE_LEDGER_CHUNK_AFTER; LOG_DEBUG_FMT( "Ending ledger chunk with signature at {}.{}", - pimpl->commit_view, - version); + applied_txid->view, + applied_txid->seqno); auto chunker = pimpl->store->get_chunker(); if (chunker) { - chunker->produced_chunk_at(version); + chunker->produced_chunk_at(applied_txid->seqno); } } diff --git a/src/kv/kv_types.h b/src/kv/kv_types.h index 96de7bc58572..ab08ff646b07 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -207,7 +207,7 @@ namespace ccf::kv }; using BatchVector = std::vector>, bool, std::shared_ptr>>; @@ -401,8 +401,7 @@ namespace ccf::kv virtual ccf::crypto::Sha256Hash get_replicated_state_root() = 0; virtual std::tuple< ccf::TxID /* TxID of last transaction seen by history */, - ccf::crypto::Sha256Hash /* root as of TxID */, - ccf::kv::Term /* term_of_next_version */> + ccf::crypto::Sha256Hash /* root as of TxID */> get_replicated_state_txid_and_root() = 0; virtual std::vector get_proof(Version v) = 0; virtual bool verify_proof(const std::vector& proof) = 0; @@ -410,11 +409,8 @@ namespace ccf::kv const std::vector& hash_at_snapshot) = 0; virtual std::vector get_raw_leaf(uint64_t index) = 0; virtual void append(const std::vector& data) = 0; - virtual void append_entry( - const ccf::crypto::Sha256Hash& digest, - std::optional expected_term = std::nullopt) = 0; - virtual void rollback( - const ccf::TxID& tx_id, ccf::kv::Term term_of_next_version_) = 0; + virtual void append_entry(const ccf::crypto::Sha256Hash& digest) = 0; + virtual void rollback(const ccf::TxID& tx_id) = 0; virtual void compact(Version v) = 0; virtual void set_term(ccf::kv::Term) = 0; virtual std::vector serialise_tree(size_t to) = 0; @@ -452,7 +448,7 @@ namespace ccf::kv virtual void init_as_backup( ccf::SeqNo, ccf::View, const std::vector&, ccf::SeqNo) = 0; - virtual bool replicate(const BatchVector& entries, ccf::View view) = 0; + virtual size_t replicate(const BatchVector& entries) = 0; virtual std::pair get_committed_txid() = 0; virtual ccf::View get_view(ccf::SeqNo seqno) = 0; @@ -708,13 +704,10 @@ namespace ccf::kv virtual void lock_map_set() = 0; virtual void unlock_map_set() = 0; - virtual Version next_version() = 0; - virtual std::tuple next_version(bool commit_new_map) = 0; virtual ccf::TxID next_txid() = 0; virtual Version current_version() = 0; virtual ccf::TxID current_txid() = 0; - virtual std::pair current_txid_and_commit_term() = 0; virtual Version compacted_version() = 0; virtual Term commit_view() = 0; diff --git a/src/kv/store.h b/src/kv/store.h index 2a4175280836..e4a90125ebaf 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -44,7 +44,6 @@ namespace ccf::kv ccf::pal::Mutex version_lock; std::atomic version = 0; - Version last_new_map = ccf::kv::NoVersion; std::atomic compacted = 0; // Calls to Store::commit are made atomic by taking this lock. @@ -66,7 +65,9 @@ namespace ccf::kv Version rollback_count = 0; - std::unordered_map, bool>> + std::unordered_map< + Version, + std::tuple, bool>> pending_txs; public: @@ -79,7 +80,6 @@ namespace ccf::kv pending_txs.clear(); version = 0; - last_new_map = ccf::kv::NoVersion; compacted = 0; term_of_next_version = 0; term_of_last_version = 0; @@ -141,7 +141,7 @@ namespace ccf::kv auto c = apply_changes( changes, - [v](bool) { return std::make_tuple(v, v - 1); }, + [term, v]() { return ccf::TxID(term, v); }, hooks, new_maps, std::nullopt, @@ -556,7 +556,7 @@ namespace ccf::kv bool track_deletes_on_missing_keys = false; auto r = apply_changes( changes, - [](bool) { return std::make_tuple(NoVersion, NoVersion); }, + [term, v]() { return ccf::TxID(term, v); }, hooks, new_maps, std::nullopt, @@ -691,7 +691,7 @@ namespace ccf::kv auto h = get_history(); if (h) { - h->rollback(tx_id, term_of_next_version); + h->rollback(tx_id); } if (tx_id.seqno >= version) @@ -939,13 +939,6 @@ namespace ccf::kv return current_txid_unsafe(); } - std::pair current_txid_and_commit_term() override - { - // Must lock in case the version or commit term is being incremented. - std::lock_guard vguard(version_lock); - return {current_txid_unsafe(), term_of_next_version}; - } - Version compacted_version() override { return compacted; @@ -977,25 +970,20 @@ namespace ccf::kv BatchVector batch; Version previous_last_replicated = 0; - Version next_last_replicated = 0; Version previous_rollback_count = 0; - ccf::View replication_view = 0; - std::vector, bool>> - contiguous_pending_txs; + std::vector contiguous_pending_txs; auto h = get_history(); { std::lock_guard vguard(version_lock); - if (txid.view != term_of_next_version && get_consensus()->is_primary()) + + if (txid.view != term_of_next_version) { - // This can happen when a transaction started before a view change, - // but tries to commit after the view change is complete. LOG_DEBUG_FMT( - "Want to commit for term {} but term is {}", - txid.view, + "Discarding transaction {} after Store moved to view {}", + txid.to_str(), term_of_next_version); - return CommitResult::FAIL_NO_REPLICATE; } @@ -1004,9 +992,24 @@ namespace ccf::kv last_committable = txid.seqno; } - pending_txs.insert( - {txid.seqno, - std::make_tuple(std::move(pending_tx), globally_committable)}); + auto [it, inserted] = pending_txs.try_emplace( + txid.seqno, txid, std::move(pending_tx), globally_committable); + + if (!inserted) + { + // Extremely unexpected case: Something went very wrong with TxID + // assignment, but still fail report here rather than persisting the + // confusion + const auto& existing_txid = std::get<0>(it->second); + + LOG_FAIL_FMT( + "Conflicting pending transactions at seqno {}: {} and {}", + txid.seqno, + existing_txid.to_str(), + txid.to_str()); + + return CommitResult::FAIL_NO_REPLICATE; + } LOG_TRACE_FMT("Inserting pending tx at {}", txid.seqno); @@ -1017,12 +1020,11 @@ namespace ccf::kv { LOG_TRACE_FMT( "Couldn't find {} = {} + {}, giving up on batch while committing " - "{}.{}", + "{}", last_replicated + offset, last_replicated, offset, - txid.view, - txid.seqno); + txid.to_str()); break; } @@ -1032,9 +1034,6 @@ namespace ccf::kv previous_rollback_count = rollback_count; previous_last_replicated = last_replicated; - next_last_replicated = last_replicated + contiguous_pending_txs.size(); - - replication_view = term_of_next_version; } // Release version lock @@ -1044,7 +1043,8 @@ namespace ccf::kv } size_t offset = 1; - for (auto& [pending_tx_, committable_] : contiguous_pending_txs) + for (auto& [pending_txid_, pending_tx_, committable_] : + contiguous_pending_txs) { auto [success_, data_, claims_digest_, commit_evidence_digest_, hooks_] = @@ -1067,18 +1067,15 @@ namespace ccf::kv if (success_ != CommitResult::SUCCESS) { LOG_FAIL_FMT( - "Unexpected failure reason {} during commit of {}.{}", + "Unexpected failure reason {} during commit of {}", static_cast(success_), - txid.view, - txid.seqno); + txid.to_str()); } if (h) { - h->append_entry( - ccf::entry_leaf( - *data_shared, commit_evidence_digest_, claims_digest_), - replication_view); + h->append_entry(ccf::entry_leaf( + *data_shared, commit_evidence_digest_, claims_digest_)); } if (chunker) @@ -1087,31 +1084,32 @@ namespace ccf::kv } LOG_DEBUG_FMT( - "Batching {} ({}) during commit of {}.{}", + "Batching {} ({}) during commit of {}", previous_last_replicated + offset, data_shared->size(), - txid.view, - txid.seqno); + txid.to_str()); batch.emplace_back( - previous_last_replicated + offset, - data_shared, - committable_, - hooks_shared); + pending_txid_, data_shared, committable_, hooks_shared); offset++; } - if (c->replicate(batch, replication_view)) + const auto replicated_count = c->replicate(batch); + if (replicated_count > 0) { std::lock_guard vguard(version_lock); if ( last_replicated == previous_last_replicated && previous_rollback_count == rollback_count) { - last_replicated = next_last_replicated; + last_replicated += replicated_count; + } + + if (last_replicated >= txid.seqno) + { + return CommitResult::SUCCESS; } - return CommitResult::SUCCESS; } LOG_DEBUG_FMT("Failed to replicate"); @@ -1176,26 +1174,6 @@ namespace ccf::kv return rollback_count == count; } - std::tuple next_version(bool commit_new_map) override - { - std::lock_guard vguard(version_lock); - Version v = next_version_unsafe(); - - auto previous_last_new_map = last_new_map; - if (commit_new_map) - { - last_new_map = v; - } - - return std::make_tuple(v, previous_last_new_map); - } - - Version next_version() override - { - std::lock_guard vguard(version_lock); - return next_version_unsafe(); - } - TxID next_txid() override { std::lock_guard vguard(version_lock); diff --git a/src/kv/test/kv_contention.cpp b/src/kv/test/kv_contention.cpp index 3bdde2a260f7..690a1462b62b 100644 --- a/src/kv/test/kv_contention.cpp +++ b/src/kv/test/kv_contention.cpp @@ -22,7 +22,7 @@ class SlowStubConsensus : public ccf::kv::test::StubConsensus public: using ccf::kv::test::StubConsensus::StubConsensus; - bool replicate(const ccf::kv::BatchVector& entries, ccf::View view) override + size_t replicate(const ccf::kv::BatchVector& entries) override { if (rand() % 2 == 0) { @@ -30,7 +30,7 @@ class SlowStubConsensus : public ccf::kv::test::StubConsensus std::this_thread::sleep_for(std::chrono::milliseconds(delay)); } - return ccf::kv::test::StubConsensus::replicate(entries, view); + return ccf::kv::test::StubConsensus::replicate(entries); } }; diff --git a/src/kv/test/kv_test.cpp b/src/kv/test/kv_test.cpp index a356b968e6d1..5df739621b04 100644 --- a/src/kv/test/kv_test.cpp +++ b/src/kv/test/kv_test.cpp @@ -3027,6 +3027,8 @@ TEST_CASE("Reported TxID after commit") // Committed transaction was not assigned a TxID because it was empty REQUIRE_FALSE(tx.get_txid().has_value()); + REQUIRE_EQ(tx.commit_version(), ccf::kv::NoVersion); + REQUIRE_EQ(tx.commit_term(), ccf::VIEW_UNKNOWN); } INFO("Simple read-only tx"); diff --git a/src/kv/test/stub_consensus.h b/src/kv/test/stub_consensus.h index 5d40a1558183..579630789c55 100644 --- a/src/kv/test/stub_consensus.h +++ b/src/kv/test/stub_consensus.h @@ -105,25 +105,24 @@ namespace ccf::kv::test state = Backup; } - bool replicate(const BatchVector& entries, ccf::View view) override + size_t replicate(const BatchVector& entries) override { for (const auto& entry : entries) { replica.push_back(entry); - const auto& [v, data, committable, hooks] = entry; + const auto& [tx_id, data, committable, hooks] = entry; - // Simplification: all entries are replicated in the same term - view_history.update(v, view); + view_history.update(tx_id.seqno, tx_id.view); if (committable) { // All committable indices are instantly committed - committed_txid = {view, v}; + committed_txid = tx_id; } + current_view = tx_id.view; } - current_view = view; - return true; + return entries.size(); } std::optional> get_latest_data() @@ -236,9 +235,9 @@ namespace ccf::kv::test return false; } - bool replicate(const BatchVector& entries, ccf::View view) override + size_t replicate(const BatchVector& entries) override { - return false; + return 0; } bool can_replicate() override diff --git a/src/kv/tx.cpp b/src/kv/tx.cpp index 3432868cc2dc..9ecbba0b453e 100644 --- a/src/kv/tx.cpp +++ b/src/kv/tx.cpp @@ -58,9 +58,7 @@ namespace ccf::kv // rather than earlier, at Tx construction. This is to minimise the // window during which concurrent transactions can write to the same map // and cause this transaction to conflict on commit. - auto p = pimpl->store->current_txid_and_commit_term(); - read_txid = p.first; - pimpl->commit_view = p.second; + read_txid = pimpl->store->current_txid(); } auto abstract_map = pimpl->store->get_map(read_txid->seqno, map_name); diff --git a/src/kv/tx_pimpl.h b/src/kv/tx_pimpl.h index 1f14e91297e4..e42f866be060 100644 --- a/src/kv/tx_pimpl.h +++ b/src/kv/tx_pimpl.h @@ -21,7 +21,6 @@ namespace ccf::kv // Note: read_txid version is set to NoVersion for the first transaction in // the service, before anything has been applied to the KV. std::optional read_txid = std::nullopt; - ccf::View commit_view = ccf::VIEW_UNKNOWN; std::map> created_maps; }; diff --git a/src/node/history.h b/src/node/history.h index db80959d33e6..a18791bcbfd0 100644 --- a/src/node/history.h +++ b/src/node/history.h @@ -119,7 +119,6 @@ namespace ccf protected: ccf::kv::Version version = 0; ccf::kv::Term term_of_last_version = 0; - ccf::kv::Term term_of_next_version = 0; public: NullTxHistory( @@ -133,10 +132,7 @@ namespace ccf version++; } - void append_entry( - const ccf::crypto::Sha256Hash& /*digest*/, - std::optional /*term_of_next_version_*/ = - std::nullopt) override + void append_entry(const ccf::crypto::Sha256Hash& /*digest*/) override { version++; } @@ -149,14 +145,12 @@ namespace ccf void set_term(ccf::kv::Term t) override { term_of_last_version = t; - term_of_next_version = t; } - void rollback(const ccf::TxID& tx_id, ccf::kv::Term commit_term_) override + void rollback(const ccf::TxID& tx_id) override { version = tx_id.seqno; term_of_last_version = tx_id.view; - term_of_next_version = commit_term_; } void compact(ccf::kv::Version /*v*/) override {} @@ -201,13 +195,12 @@ namespace ccf return ccf::crypto::Sha256Hash(std::to_string(version)); } - std::tuple + std::tuple get_replicated_state_txid_and_root() override { return { {term_of_last_version, version}, - ccf::crypto::Sha256Hash(std::to_string(version)), - term_of_next_version}; + ccf::crypto::Sha256Hash(std::to_string(version))}; } std::vector get_proof(ccf::kv::Version /*v*/) override @@ -575,7 +568,6 @@ namespace ccf ccf::pal::Mutex state_lock; ccf::kv::Term term_of_last_version = 0; - ccf::kv::Term term_of_next_version{}; std::optional endorsed_cert = std::nullopt; @@ -757,15 +749,14 @@ namespace ccf return replicated_state_tree.get_root(); } - std::tuple + std::tuple get_replicated_state_txid_and_root() override { std::lock_guard guard(state_lock); return { {term_of_last_version, static_cast(replicated_state_tree.end_index())}, - replicated_state_tree.get_root(), - term_of_next_version}; + replicated_state_tree.get_root()}; } bool verify_root_signatures(ccf::kv::Version version) override @@ -891,16 +882,13 @@ namespace ccf // term std::lock_guard guard(state_lock); term_of_last_version = t; - term_of_next_version = t; } - void rollback( - const ccf::TxID& tx_id, ccf::kv::Term term_of_next_version_) override + void rollback(const ccf::TxID& tx_id) override { std::lock_guard guard(state_lock); LOG_TRACE_FMT("Rollback to {}.{}", tx_id.view, tx_id.seqno); term_of_last_version = tx_id.view; - term_of_next_version = term_of_next_version_; replicated_state_tree.retract(tx_id.seqno); log_hash(replicated_state_tree.get_root(), ROLLBACK); } @@ -1003,20 +991,10 @@ namespace ccf replicated_state_tree.append(rh); } - void append_entry( - const ccf::crypto::Sha256Hash& digest, - std::optional expected_term_of_next_version = - std::nullopt) override + void append_entry(const ccf::crypto::Sha256Hash& digest) override { log_hash(digest, APPEND); std::lock_guard guard(state_lock); - if (expected_term_of_next_version.has_value()) - { - if (expected_term_of_next_version.value() != term_of_next_version) - { - return; - } - } replicated_state_tree.append(digest); } diff --git a/src/node/node_state.h b/src/node/node_state.h index eb4f64e7a911..38eff1274625 100644 --- a/src/node/node_state.h +++ b/src/node/node_state.h @@ -2432,7 +2432,7 @@ namespace ccf // version can advance before its Merkle history is updated during commit, // so these must be captured together from the history. auto* h = dynamic_cast(history.get()); - const auto& [txid, root, _] = h->get_replicated_state_txid_and_root(); + const auto& [txid, root] = h->get_replicated_state_txid_and_root(); recovery_v = txid.seqno; recovery_root = root; diff --git a/src/node/rpc/frontend.h b/src/node/rpc/frontend.h index 01a7d28af1a3..ca1e51cc82df 100644 --- a/src/node/rpc/frontend.h +++ b/src/node/rpc/frontend.h @@ -900,8 +900,7 @@ namespace ccf }; } - ccf::kv::CommitResult result = - tx.commit(ctx->claims, nullptr, ws_observer); + ccf::kv::CommitResult result = tx.commit(ctx->claims, ws_observer); switch (result) { @@ -1125,9 +1124,9 @@ namespace ccf // should only ever be used for the proposal creation endpoint and // nothing else. Many bad things could happen otherwise (e.g. breaking // session consistency). - const auto& [txid, root, term_of_next_version] = + const auto& [txid, root] = current_history->get_replicated_state_txid_and_root(); - tx.set_read_txid(txid, term_of_next_version); + tx.set_read_txid(txid); tx.set_root_at_read_version(root); } } diff --git a/src/node/snapshotter.h b/src/node/snapshotter.h index 5b437c2b1901..03a0d93703bc 100644 --- a/src/node/snapshotter.h +++ b/src/node/snapshotter.h @@ -277,7 +277,7 @@ namespace ccf commit_evidence = commit_evidence_; }; - auto rc = tx.commit(cd, nullptr, capture_ws_digest_and_commit_evidence); + auto rc = tx.commit(cd, capture_ws_digest_and_commit_evidence); if (rc != ccf::kv::CommitResult::SUCCESS) { LOG_FAIL_FMT( diff --git a/src/node/test/historical_queries.cpp b/src/node/test/historical_queries.cpp index 47a8561da949..db0d252f0fd3 100644 --- a/src/node/test/historical_queries.cpp +++ b/src/node/test/historical_queries.cpp @@ -223,18 +223,18 @@ std::map> construct_host_ledger( std::map> ledger; auto next_ledger_entry = consensus->pop_oldest_entry(); - auto version = std::get<0>(next_ledger_entry.value()); + auto version = std::get<0>(next_ledger_entry.value()).seqno; while (next_ledger_entry.has_value()) { - const auto ib = ledger.insert(std::make_pair( - std::get<0>(next_ledger_entry.value()), - *std::get<1>(next_ledger_entry.value()))); + const auto ib = ledger.insert( + std::make_pair(version, *std::get<1>(next_ledger_entry.value()))); REQUIRE(ib.second); next_ledger_entry = consensus->pop_oldest_entry(); if (next_ledger_entry.has_value()) { - REQUIRE(version + 1 == std::get<0>(next_ledger_entry.value())); - version = std::get<0>(next_ledger_entry.value()); + const auto next_version = std::get<0>(next_ledger_entry.value()).seqno; + REQUIRE(version + 1 == next_version); + version = next_version; } } diff --git a/src/node/test/history.cpp b/src/node/test/history.cpp index 02eb4e7e5aa5..584b8ed832fd 100644 --- a/src/node/test/history.cpp +++ b/src/node/test/history.cpp @@ -39,15 +39,19 @@ class DummyConsensus : public ccf::kv::test::StubConsensus DummyConsensus(ccf::kv::Store* store_) : store(store_) {} - bool replicate(const ccf::kv::BatchVector& entries, ccf::View view) override + size_t replicate(const ccf::kv::BatchVector& entries) override { if (store) { REQUIRE(entries.size() == 1); - return store->deserialize(*std::get<1>(entries[0]))->apply() != - ccf::kv::ApplyResult::FAIL; + if ( + store->deserialize(*std::get<1>(entries[0]))->apply() != + ccf::kv::ApplyResult::FAIL) + { + return 1; + } } - return true; + return 0; } std::pair get_committed_txid() override @@ -255,15 +259,15 @@ class CompactingConsensus : public ccf::kv::test::StubConsensus CompactingConsensus(ccf::kv::Store* store_) : store(store_) {} - bool replicate(const ccf::kv::BatchVector& entries, ccf::View view) override + size_t replicate(const ccf::kv::BatchVector& entries) override { - for (auto& [version, data, committable, hooks] : entries) + for (auto& [tx_id, data, committable, hooks] : entries) { count++; if (committable) - store->compact(version); + store->compact(tx_id.seqno); } - return true; + return entries.size(); } std::pair get_committed_txid() override @@ -429,15 +433,15 @@ class RollbackConsensus : public ccf::kv::test::StubConsensus rollback_to(rollback_to_) {} - bool replicate(const ccf::kv::BatchVector& entries, ccf::View view) override + size_t replicate(const ccf::kv::BatchVector& entries) override { - for (auto& [version, data, committable, hook] : entries) + for (auto& [tx_id, data, committable, hook] : entries) { count++; - if (version == rollback_at) - store->rollback({view, rollback_to}, store->commit_view()); + if (tx_id.seqno == rollback_at) + store->rollback(tx_id, store->commit_view()); } - return true; + return entries.size(); } std::pair get_committed_txid() override