diff --git a/CHANGELOG.md b/CHANGELOG.md index 3fad6684147f..61ba23d9848f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Changed - `ccf::SessionContext::caller_cert` is now immutable, and its SHA-256 digest is cached per session to avoid repeated hashing during user and member certificate authentication (#8164). +- Consensus current-state queries now return coherent light or full details snapshots. Full details include configuration and per-node acknowledgement data, while KV critical sections use a separate lock-order-safe consensus query API (#8184). ## [7.0.12] diff --git a/doc/schemas/node_openapi.json b/doc/schemas/node_openapi.json index 0bc4626534ef..3a23391070b0 100644 --- a/doc/schemas/node_openapi.json +++ b/doc/schemas/node_openapi.json @@ -134,6 +134,12 @@ "configs": { "$ref": "#/components/schemas/Configuration_array" }, + "committed_seqno": { + "$ref": "#/components/schemas/uint64" + }, + "committed_view": { + "$ref": "#/components/schemas/uint64" + }, "current_view": { "$ref": "#/components/schemas/uint64" }, @@ -157,11 +163,15 @@ }, "ticking": { "$ref": "#/components/schemas/boolean" + }, + "view_history": { + "$ref": "#/components/schemas/ConsensusViewHistory" } }, "required": [ "configs", "acks", + "view_history", "membership_state", "primary_id", "current_view", @@ -184,6 +194,17 @@ ], "type": "object" }, + "ConsensusViewHistory": { + "properties": { + "starts": { + "$ref": "#/components/schemas/uint64_array" + } + }, + "required": [ + "starts" + ], + "type": "object" + }, "ConsensusNodeConfig": { "properties": { "address": { @@ -899,6 +920,12 @@ "maximum": 18446744073709551615, "minimum": 0, "type": "integer" + }, + "uint64_array": { + "items": { + "$ref": "#/components/schemas/uint64" + }, + "type": "array" } }, "x-ccf-forwarding": { diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 286a3e0ebb8c..b7c13c4a3bfe 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -247,32 +247,11 @@ namespace aft ~Aft() override = default; - std::optional primary() override - { - return leader_id; - } - - ccf::NodeId id() override + ccf::NodeId id() { return state->node_id; } - bool is_primary() override - { - return state->leadership_state == ccf::kv::LeadershipState::Leader; - } - - bool is_candidate() override - { - return state->leadership_state == ccf::kv::LeadershipState::Candidate; - } - - bool can_replicate() override - { - std::unique_lock guard(state->lock); - return can_replicate_unsafe(); - } - /** * Returns true if the node is primary, max_uncommitted_tx_count is non-zero * and the number of transactions replicated but not yet committed exceeds @@ -303,16 +282,7 @@ namespace aft return Consensus::SignatureDisposition::CANT_REPLICATE; } - bool is_backup() override - { - return state->leadership_state == ccf::kv::LeadershipState::Follower; - } - - bool is_active() const - { - return state->membership_state == ccf::kv::MembershipState::Active; - } - + private: bool is_retired() const { return state->membership_state == ccf::kv::MembershipState::Retired; @@ -324,18 +294,13 @@ namespace aft state->retirement_phase == ccf::kv::RetirementPhase::RetiredCommitted; } - bool is_retired_completed() const - { - return state->membership_state == ccf::kv::MembershipState::Retired && - state->retirement_phase == ccf::kv::RetirementPhase::Completed; - } - + public: void set_retired_committed( ccf::SeqNo seqno, const std::vector& node_ids) override { for (const auto& node_id : node_ids) { - if (id() == node_id) + if (state->node_id == node_id) { CCF_ASSERT( state->membership_state == ccf::kv::MembershipState::Retired, @@ -464,41 +429,21 @@ namespace aft return state->last_idx; } - Index get_committed_seqno() override + std::vector get_view_history(Index idx) { - std::lock_guard guard(state->lock); - return get_commit_idx_unsafe(); - } - - Term get_view() override - { - std::lock_guard guard(state->lock); - return state->current_view; + return state->view_history.get_history_until(idx); } - std::pair get_committed_txid() override + ccf::TxStatus evaluate_tx_status( + ccf::View target_view, ccf::SeqNo target_seqno) override { std::lock_guard guard(state->lock); - ccf::SeqNo commit_idx = get_commit_idx_unsafe(); - return {get_term_internal(commit_idx), commit_idx}; - } + const auto local_view = get_term_internal(target_seqno); + const auto committed_seqno = get_commit_idx_unsafe(); + const auto committed_view = get_term_internal(committed_seqno); - Term get_view(Index idx) override - { - std::lock_guard guard(state->lock); - return get_term_internal(idx); - } - - std::vector get_view_history(Index idx) override - { - // This should only be called when the spin lock is held. - return state->view_history.get_history_until(idx); - } - - std::vector get_view_history_since(Index idx) override - { - // This should only be called when the spin lock is held. - return state->view_history.get_history_since(idx); + return ccf::evaluate_tx_status( + target_view, target_seqno, local_view, committed_view, committed_seqno); } // Same as ccfraft.tla GetServerSet/IsInServerSet @@ -579,28 +524,39 @@ namespace aft } } - Configuration::Nodes get_latest_configuration_unsafe() const override + void update_configuration( + Index idx, const Configuration::NodeChanges& changes) override { - if (configurations.empty()) + if (changes.empty()) { - return {}; + return; } - return configurations.back().nodes; - } - - Configuration::Nodes get_latest_configuration() override - { - std::lock_guard guard(state->lock); - return get_latest_configuration_unsafe(); + auto configuration = configurations.empty() ? Configuration::Nodes{} : + configurations.back().nodes; + for (const auto& [node_id, node_info] : changes) + { + if (node_info.has_value()) + { + configuration.try_emplace(node_id, node_info.value()); + } + else + { + configuration.erase(node_id); + } + } + add_configuration(idx, configuration); } - ccf::kv::ConsensusDetails get_details() override + private: + ccf::kv::ConsensusLightDetails get_light_details_unsafe() { - ccf::kv::ConsensusDetails details; - std::lock_guard guard(state->lock); + ccf::kv::ConsensusLightDetails details; details.primary_id = leader_id; details.current_view = state->current_view; + const auto committed_seqno = get_commit_idx_unsafe(); + details.committed_seqno = committed_seqno; + details.committed_view = get_term_internal(committed_seqno); details.ticking = ticking; details.leadership_state = state->leadership_state; details.membership_state = state->membership_state; @@ -608,16 +564,30 @@ namespace aft { details.retirement_phase = state->retirement_phase; } - for (auto const& conf : configurations) - { - details.configs.push_back(conf); - } + details.reconfiguration_type = ccf::ReconfigurationType::ONE_TRANSACTION; + return details; + } + + public: + ccf::kv::ConsensusLightDetails get_light_details() override + { + std::lock_guard guard(state->lock); + return get_light_details_unsafe(); + } + + ccf::kv::ConsensusDetails get_details() override + { + ccf::kv::ConsensusDetails details; + std::lock_guard guard(state->lock); + static_cast(details) = + get_light_details_unsafe(); + details.configs.assign(configurations.begin(), configurations.end()); + details.view_history.starts = state->view_history.get_history_until(); for (auto& [k, v] : all_other_nodes) { details.acks[k] = { v.match_idx, static_cast(v.last_ack_timeout.count())}; } - details.reconfiguration_type = ccf::ReconfigurationType::ONE_TRANSACTION; return details; } @@ -2607,7 +2577,10 @@ namespace aft } RAFT_DEBUG_FMT("Compacting..."); - store->compact(idx); + store->compact( + idx, + state->leadership_state == ccf::kv::LeadershipState::Leader, + {state->view_history.get_history_until(idx)}); ledger->commit(idx); if (commit_callbacks != nullptr) @@ -2650,7 +2623,9 @@ namespace aft if (changed) { create_and_remove_node_state(); - if (retired_node_cleanup && is_primary()) + if ( + retired_node_cleanup && + state->leadership_state == ccf::kv::LeadershipState::Leader) { retired_node_cleanup->cleanup(); } diff --git a/src/consensus/aft/raft_types.h b/src/consensus/aft/raft_types.h index 57b6b3bb03ad..c30f7490110d 100644 --- a/src/consensus/aft/raft_types.h +++ b/src/consensus/aft/raft_types.h @@ -26,7 +26,10 @@ namespace aft { public: virtual ~Store() = default; - virtual void compact(Index v) = 0; + virtual void compact( + Index v, + bool is_primary, + const ccf::kv::ConsensusViewHistory& view_history) = 0; virtual void rollback( const ccf::TxID& tx_id, Term term_of_next_version) = 0; virtual void initialise_term(Term t) = 0; @@ -45,12 +48,15 @@ namespace aft public: Adaptor(std::shared_ptr x) : x(x) {} - void compact(Index v) override + void compact( + Index v, + bool is_primary, + const ccf::kv::ConsensusViewHistory& view_history) override { auto p = x.lock(); if (p) { - p->compact(v); + p->compact(v, is_primary, view_history); } } diff --git a/src/consensus/aft/test/driver.h b/src/consensus/aft/test/driver.h index aa2693bef970..86af81f8c35e 100644 --- a/src/consensus/aft/test/driver.h +++ b/src/consensus/aft/test/driver.h @@ -74,10 +74,13 @@ struct LoggingStubStore_Mermaid : public aft::LoggingStubStoreConfig { using LoggingStubStoreConfig::LoggingStubStoreConfig; - void compact(aft::Index idx) override + void compact( + aft::Index idx, + bool is_primary, + const ccf::kv::ConsensusViewHistory& view_history) override { RAFT_DRIVER_PRINT("{}->>{}: [KV] compacting to {}", _id, _id, idx); - aft::LoggingStubStoreConfig::compact(idx); + aft::LoggingStubStoreConfig::compact(idx, is_primary, view_history); } void rollback(const ccf::TxID& tx_id, aft::Term t) override @@ -100,7 +103,7 @@ struct LoggingStubStore_Mermaid : public aft::LoggingStubStoreConfig }; using ms = std::chrono::milliseconds; -using TRaft = aft::Aft; +using TRaft = aft::TestAft; using Store = LoggingStubStore_Mermaid; using Adaptor = aft::Adaptor; @@ -633,7 +636,7 @@ class RaftDriver std::vector entries; for (ccf::kv::Version i = 1; i <= r.get_last_idx(); ++i) { - const auto t = r.get_view(i); + const auto t = r.get_details().view_history.view_at(i); auto s = fmt::format("{}.{}", t, i); if (i == r.get_committed_seqno()) { diff --git a/src/consensus/aft/test/logging_stub.h b/src/consensus/aft/test/logging_stub.h index 509893dfc922..f2bb55aac6aa 100644 --- a/src/consensus/aft/test/logging_stub.h +++ b/src/consensus/aft/test/logging_stub.h @@ -141,6 +141,80 @@ namespace aft void commit(Index idx) {} }; + // This is a shim presenting the old Raft API for test purposes, to avoid + // unnecessarily rewriting many test calling points + template + class TestAft : public Aft + { + public: + using Aft::Aft; + + ccf::View get_view(ccf::SeqNo seqno) + { + return this->get_details().view_history.view_at(seqno); + } + + std::optional primary() + { + return this->get_light_details().primary_id; + } + + bool is_primary() + { + return this->get_light_details().is_primary(); + } + + bool is_backup() + { + return this->get_light_details().is_backup(); + } + + bool is_candidate() + { + return this->get_light_details().is_candidate(); + } + + bool is_active() + { + return this->get_light_details().membership_state == + ccf::kv::MembershipState::Active; + } + + bool is_retired() + { + return this->get_light_details().membership_state == + ccf::kv::MembershipState::Retired; + } + + bool is_retired_completed() + { + const auto details = this->get_light_details(); + return details.membership_state == ccf::kv::MembershipState::Retired && + details.retirement_phase == ccf::kv::RetirementPhase::Completed; + } + + Index get_committed_seqno() + { + return this->get_light_details().committed_seqno; + } + + Term get_view() + { + return this->get_light_details().current_view; + } + + std::pair get_committed_txid() + { + const auto details = this->get_light_details(); + return {details.committed_view, details.committed_seqno}; + } + + Configuration::Nodes get_latest_configuration() + { + return this->get_details().configs.back().nodes; + } + }; + class ChannelStubProxy : public ccf::NodeToNode { public: @@ -305,29 +379,7 @@ namespace aft void call(ccf::kv::ConfigurableConsensus* consensus) override { - auto configuration = consensus->get_latest_configuration_unsafe(); - std::list itrs; - - // Remove and track retired nodes - for (auto it = configuration.begin(); it != configuration.end(); ++it) - { - if (new_configuration.find(it->first) == new_configuration.end()) - { - itrs.push_back(it); - } - } - for (auto it : itrs) - { - configuration.erase(it); - } - - // Add new node to configuration - for (const auto& [node_id, _] : new_configuration) - { - configuration[node_id] = {}; - } - - consensus->add_configuration(version, configuration); + consensus->add_configuration(version, new_configuration); } }; @@ -351,7 +403,11 @@ namespace aft set_retired_committed_hook = set_retired_committed_hook_; } - virtual void compact(Index i) {} + virtual void compact( + Index i, + bool is_primary, + const ccf::kv::ConsensusViewHistory& view_history) + {} virtual void rollback(const ccf::TxID& tx_id, Term t) {} @@ -482,7 +538,10 @@ namespace aft // compact and rollback emulate the behaviour of the retired_committed hook // in the real store through the retired_committed_entries vector, see // node_state.h, circa line 2147 - virtual void compact(Index i) override + virtual void compact( + Index i, + bool is_primary, + const ccf::kv::ConsensusViewHistory& view_history) override { for (auto& [version, configuration] : retired_committed_entries) { diff --git a/src/consensus/aft/test/main.cpp b/src/consensus/aft/test/main.cpp index 317ec0668bac..5144b9c81ef4 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -5,10 +5,32 @@ #define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES #define DOCTEST_CONFIG_IMPLEMENT +#include #include using ms = std::chrono::milliseconds; +DOCTEST_TEST_CASE("Consensus view history snapshot") +{ + ccf::kv::ConsensusViewHistory view_history{{1, 4, 4, 9}}; + + DOCTEST_REQUIRE(view_history.view_at(0) == ccf::VIEW_UNKNOWN); + DOCTEST_REQUIRE(view_history.view_at(1) == 1); + DOCTEST_REQUIRE(view_history.view_at(3) == 1); + DOCTEST_REQUIRE(view_history.view_at(4) == 3); + DOCTEST_REQUIRE(view_history.view_at(9) == 4); + + DOCTEST_REQUIRE(view_history.until(0).starts.empty()); + DOCTEST_REQUIRE( + view_history.until(4).starts == std::vector{1, 4, 4}); + DOCTEST_REQUIRE( + view_history.until(8).starts == std::vector{1, 4, 4}); + + DOCTEST_REQUIRE(view_history.since(0).empty()); + DOCTEST_REQUIRE(view_history.since(2) == std::vector{4, 4, 9}); + DOCTEST_REQUIRE(view_history.since(5).empty()); +} + DOCTEST_TEST_CASE("Single node startup" * doctest::test_suite("single")) { ccf::NodeId node_id = ccf::kv::test::PrimaryNodeId; @@ -45,6 +67,124 @@ DOCTEST_TEST_CASE("Single node startup" * doctest::test_suite("single")) DOCTEST_REQUIRE(r0.primary() == node_id); } +DOCTEST_TEST_CASE( + "Consensus details and primary state" * doctest::test_suite("single")) +{ + const auto node_id = ccf::kv::test::PrimaryNodeId; + const auto other_node_id = ccf::kv::test::FirstBackupNodeId; + auto kv_store = std::make_shared(node_id); + + TRaft raft( + raft_settings, + std::make_unique(kv_store), + std::make_unique(node_id), + std::make_shared(), + std::make_shared(node_id), + nullptr); + + ccf::kv::Configuration::Nodes config; + config.try_emplace(node_id); + config.try_emplace(other_node_id); + raft.add_configuration(0, config); + + const auto light_details = raft.get_light_details(); + DOCTEST_REQUIRE(!light_details.is_primary()); + DOCTEST_REQUIRE(!raft.is_primary()); + + const auto diagnostic_details = raft.get_details(); + DOCTEST_REQUIRE(diagnostic_details.configs.size() == 1); + DOCTEST_REQUIRE(diagnostic_details.configs.front().nodes == config); + DOCTEST_REQUIRE(diagnostic_details.acks.contains(other_node_id)); + + // Configuration deltas add new nodes without rewriting existing node info. + const auto new_node_id = ccf::kv::test::SecondBackupNodeId; + raft.update_configuration( + 1, + {{other_node_id, ccf::kv::Configuration::NodeInfo{"ignored", "1"}}, + {new_node_id, ccf::kv::Configuration::NodeInfo{"new", "2"}}}); + const auto updated_details = raft.get_details(); + DOCTEST_REQUIRE(updated_details.configs.size() == 2); + DOCTEST_REQUIRE( + updated_details.configs.back().nodes.at(other_node_id) == + config.at(other_node_id)); + DOCTEST_REQUIRE( + updated_details.configs.back().nodes.at(new_node_id) == + ccf::kv::Configuration::NodeInfo{"new", "2"}); + + raft.force_become_primary(); + const auto primary_details = raft.get_light_details(); + DOCTEST_REQUIRE(primary_details.is_primary()); + DOCTEST_REQUIRE(primary_details.primary_id == node_id); + DOCTEST_REQUIRE(raft.is_primary()); + + raft.become_follower(); + DOCTEST_REQUIRE(!raft.get_light_details().is_primary()); + DOCTEST_REQUIRE(!raft.is_primary()); +} + +DOCTEST_TEST_CASE( + "Concurrent public state reads during leadership transitions" * + doctest::test_suite("concurrency")) +{ + const auto node_id = ccf::kv::test::PrimaryNodeId; + const auto other_node_id = ccf::kv::test::FirstBackupNodeId; + auto kv_store = std::make_shared(node_id); + + TRaft raft( + raft_settings, + std::make_unique(kv_store), + std::make_unique(node_id), + std::make_shared(), + std::make_shared(node_id), + nullptr); + + aft::Configuration::Nodes config; + config.try_emplace(node_id); + config.try_emplace(other_node_id); + raft.add_configuration(0, config); + + std::atomic stop = false; + std::atomic observed = false; + constexpr size_t reader_thread_count = 8; + std::barrier start(reader_thread_count + 1); + std::thread driver([&]() { + start.arrive_and_wait(); + constexpr size_t transition_count = 2000; + for (size_t i = 0; i < transition_count; ++i) + { + raft.force_become_primary(); + raft.periodic(election_timeout); + } + stop.store(true, std::memory_order_release); + }); + + std::vector readers; + readers.reserve(reader_thread_count); + for (size_t i = 0; i < reader_thread_count; ++i) + { + readers.emplace_back([&]() { + start.arrive_and_wait(); + while (!stop.load(std::memory_order_acquire)) + { + const auto details = raft.get_light_details(); + observed.store(true, std::memory_order_release); + static_cast(details.primary_id.has_value()); + static_cast(details.is_primary()); + static_cast(details.is_candidate()); + static_cast(details.is_backup()); + raft.is_primary(); + } + }); + } + + driver.join(); + for (auto& reader : readers) + { + reader.join(); + } + DOCTEST_REQUIRE(observed.load(std::memory_order_acquire)); +} + DOCTEST_TEST_CASE("Single node commit" * doctest::test_suite("single")) { ccf::NodeId node_id = ccf::kv::test::PrimaryNodeId; diff --git a/src/consensus/aft/test/test_common.h b/src/consensus/aft/test/test_common.h index f9fc5f5ab3cf..2b211fa9ebe9 100644 --- a/src/consensus/aft/test/test_common.h +++ b/src/consensus/aft/test/test_common.h @@ -10,7 +10,7 @@ #include #include -using TRaft = aft::Aft; +using TRaft = aft::TestAft; using Store = aft::LoggingStubStore; using Adaptor = aft::Adaptor; diff --git a/src/endpoints/base_endpoint_registry.cpp b/src/endpoints/base_endpoint_registry.cpp index ac704a1438ef..84f3b6b3037a 100644 --- a/src/endpoints/base_endpoint_registry.cpp +++ b/src/endpoints/base_endpoint_registry.cpp @@ -32,14 +32,14 @@ namespace ccf reason = ccf::InvalidArgsReason::ViewSmallerThanOne; return ApiResult::InvalidArgs; } - auto latest_view = current_consensus->get_view(); + const auto details = current_consensus->get_details(); + const auto latest_view = details.current_view; if (since > latest_view) { // asking for something in the future return ApiResult::NotFound; } - const auto view_history = - current_consensus->get_view_history_since(since); + const auto view_history = details.view_history.since(since); for (ccf::View i = 0; i < view_history.size(); i++) { const auto view = i + since; @@ -96,9 +96,10 @@ namespace ccf { try { - const auto [v, s] = current_consensus->get_committed_txid(); - view = v; - seqno = s; + const auto details = current_consensus->get_light_details(); + const auto committed_txid = details.committed_txid(); + view = committed_txid.view; + seqno = committed_txid.seqno; return ApiResult::OK; } catch (const std::exception& e) @@ -206,7 +207,8 @@ namespace ccf auto* current_consensus = get_consensus(); if (current_consensus != nullptr) { - const auto v = current_consensus->get_view(seqno); + const auto v = + current_consensus->get_details().view_history.view_at(seqno); if (v != ccf::VIEW_UNKNOWN) { view = v; diff --git a/src/kv/kv_types.h b/src/kv/kv_types.h index 7a85981ad16d..2260e9077e71 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -72,6 +72,7 @@ namespace ccf::kv }; using Nodes = std::map; + using NodeChanges = std::map>; ccf::SeqNo idx = 0; Nodes nodes; @@ -153,16 +154,45 @@ namespace ccf::kv DECLARE_JSON_TYPE(Configuration); DECLARE_JSON_REQUIRED_FIELDS(Configuration, idx, nodes, rid); - struct ConsensusDetails + struct ConsensusViewHistory { - struct Ack + // Entry i stores the first sequence number in view i + 1. + std::vector starts; + + [[nodiscard]] ccf::View view_at(ccf::SeqNo seqno) const { - ccf::SeqNo seqno; - size_t last_received_ms; - }; + const auto it = std::upper_bound(starts.begin(), starts.end(), seqno); + if (it == starts.begin()) + { + return ccf::VIEW_UNKNOWN; + } - std::vector configs; - std::unordered_map acks; + return it - starts.begin(); + } + + [[nodiscard]] ConsensusViewHistory until(ccf::SeqNo seqno) const + { + return { + {starts.begin(), + std::upper_bound(starts.begin(), starts.end(), seqno)}}; + } + + [[nodiscard]] std::vector since(ccf::View view) const + { + if (view == 0 || view > starts.size()) + { + return {}; + } + + return {starts.begin() + view - 1, starts.end()}; + } + }; + + DECLARE_JSON_TYPE(ConsensusViewHistory); + DECLARE_JSON_REQUIRED_FIELDS(ConsensusViewHistory, starts); + + struct ConsensusLightDetails + { MembershipState membership_state{}; std::optional leadership_state = std::nullopt; std::optional retirement_phase = std::nullopt; @@ -171,38 +201,77 @@ namespace ccf::kv std::optional reconfiguration_type = std::nullopt; std::optional primary_id = std::nullopt; ccf::View current_view = 0; + ccf::View committed_view = 0; + ccf::SeqNo committed_seqno = 0; bool ticking = false; + + [[nodiscard]] bool is_primary() const + { + return leadership_state == LeadershipState::Leader; + } + + [[nodiscard]] bool is_backup() const + { + return leadership_state == LeadershipState::Follower; + } + + [[nodiscard]] bool is_candidate() const + { + return leadership_state == LeadershipState::Candidate; + } + + [[nodiscard]] bool can_replicate() const + { + return is_primary() && + !(membership_state == MembershipState::Retired && + retirement_phase == RetirementPhase::RetiredCommitted); + } + + [[nodiscard]] ccf::TxID committed_txid() const + { + return {committed_view, committed_seqno}; + } + }; + + struct ConsensusDetails : ConsensusLightDetails + { + struct Ack + { + ccf::SeqNo seqno; + size_t last_received_ms; + }; + + std::vector configs; + std::unordered_map acks; + ConsensusViewHistory view_history; }; DECLARE_JSON_TYPE(ConsensusDetails::Ack); DECLARE_JSON_REQUIRED_FIELDS(ConsensusDetails::Ack, seqno, last_received_ms); - DECLARE_JSON_TYPE_WITH_OPTIONAL_FIELDS(ConsensusDetails); + DECLARE_JSON_TYPE_WITH_OPTIONAL_FIELDS(ConsensusLightDetails); DECLARE_JSON_REQUIRED_FIELDS( - ConsensusDetails, - configs, - acks, - membership_state, - primary_id, - current_view, - ticking); + ConsensusLightDetails, membership_state, primary_id, current_view, ticking); DECLARE_JSON_OPTIONAL_FIELDS( - ConsensusDetails, + ConsensusLightDetails, reconfiguration_type, learners, leadership_state, - retirement_phase); + retirement_phase, + committed_view, + committed_seqno); + + DECLARE_JSON_TYPE_WITH_BASE(ConsensusDetails, ConsensusLightDetails); + DECLARE_JSON_REQUIRED_FIELDS(ConsensusDetails, configs, acks, view_history); class ConfigurableConsensus { public: virtual ~ConfigurableConsensus() = default; virtual void add_configuration( - ccf::SeqNo seqno, const Configuration::Nodes& conf) = 0; - virtual Configuration::Nodes get_latest_configuration() = 0; - [[nodiscard]] virtual Configuration::Nodes get_latest_configuration_unsafe() - const = 0; - virtual ConsensusDetails get_details() = 0; + ccf::SeqNo seqno, const Configuration::Nodes& configuration) = 0; + virtual void update_configuration( + ccf::SeqNo seqno, const Configuration::NodeChanges& changes) = 0; }; using BatchVector = std::vector&, ccf::SeqNo) = 0; virtual bool replicate(const BatchVector& entries, ccf::View view) = 0; - virtual std::pair get_committed_txid() = 0; - virtual ccf::View get_view(ccf::SeqNo seqno) = 0; - virtual ccf::View get_view() = 0; - virtual std::vector get_view_history( - ccf::SeqNo seqno = std::numeric_limits::max()) = 0; - virtual std::vector get_view_history_since( - ccf::SeqNo seqno) = 0; - virtual ccf::SeqNo get_committed_seqno() = 0; - virtual std::optional primary() = 0; + virtual ConsensusLightDetails get_light_details() = 0; + virtual ConsensusDetails get_details() = 0; + + virtual ccf::TxStatus evaluate_tx_status( + ccf::View target_view, ccf::SeqNo target_seqno) = 0; virtual void recv_message( const NodeId& from, const uint8_t* data, size_t size) = 0; @@ -469,16 +529,6 @@ namespace ccf::kv {} virtual void nominate_successor() {} - - ccf::TxStatus evaluate_tx_status( - ccf::View target_view, ccf::SeqNo target_seqno) - { - const auto local_view = get_view(target_seqno); - const auto [committed_view, committed_seqno] = get_committed_txid(); - - return ccf::evaluate_tx_status( - target_view, target_seqno, local_view, committed_view, committed_seqno); - } }; struct PendingTxInfo @@ -588,7 +638,10 @@ namespace ccf::kv virtual bool record_committable(ccf::kv::Version v) = 0; virtual bool should_schedule_snapshot(ccf::kv::Version v) = 0; - virtual void commit(ccf::kv::Version v, bool generate_snapshot) = 0; + virtual void commit( + ccf::kv::Version v, + bool generate_snapshot, + const ConsensusViewHistory& view_history) = 0; virtual void rollback(ccf::kv::Version v) = 0; }; using SnapshotterPtr = std::shared_ptr; @@ -736,7 +789,7 @@ namespace ccf::kv virtual bool check_rollback_count(Version count) = 0; virtual std::unique_ptr snapshot_unsafe_maps( - Version v) = 0; + Version v, ConsensusViewHistory view_history) = 0; virtual void lock_maps() = 0; virtual void unlock_maps() = 0; virtual std::vector serialise_snapshot( diff --git a/src/kv/store.h b/src/kv/store.h index dce78e55f41d..3a15786ca6a0 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -348,7 +348,8 @@ namespace ccf::kv } } - std::unique_ptr snapshot_unsafe_maps(Version v) override + std::unique_ptr snapshot_unsafe_maps( + Version v, ConsensusViewHistory view_history) override { auto cv = compacted_version(); if (v < cv) @@ -384,10 +385,9 @@ namespace ccf::kv snapshot->add_hash_at_snapshot(h->get_raw_leaf(v)); } - auto c = get_consensus(); - if (c) + if (!view_history.starts.empty()) { - snapshot->add_view_history(c->get_view_history(v)); + snapshot->add_view_history(std::move(view_history.starts)); } } @@ -567,6 +567,12 @@ namespace ccf::kv } void compact(Version v) override + { + compact(v, false, {}); + } + + void compact( + Version v, bool is_primary, const ConsensusViewHistory& view_history) { // This is called when the store will never be rolled back to any // state before the specified version. @@ -574,9 +580,7 @@ namespace ccf::kv if (snapshotter) { - auto c = get_consensus(); - bool generate_snapshot = c && c->is_primary(); - snapshotter->commit(v, generate_snapshot); + snapshotter->commit(v, is_primary, view_history); } if (chunker) @@ -961,10 +965,10 @@ namespace ccf::kv { 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. + // but tries to commit after the local store has moved to a new term. LOG_DEBUG_FMT( "Want to commit for term {} but term is {}", txid.view, diff --git a/src/kv/test/kv_bench.cpp b/src/kv/test/kv_bench.cpp index d772315890b3..b2a619000e08 100644 --- a/src/kv/test/kv_bench.cpp +++ b/src/kv/test/kv_bench.cpp @@ -194,7 +194,7 @@ static void ser_snap(picobench::state& s) std::unique_ptr snap = nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&kv_store); - snap = kv_store.snapshot_unsafe_maps(tx.commit_version()); + snap = kv_store.snapshot_unsafe_maps(tx.commit_version(), {}); } kv_store.serialise_snapshot(std::move(snap)); s.stop_timer(); @@ -232,7 +232,7 @@ static void des_snap(picobench::state& s) std::unique_ptr snap = nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&kv_store); - snap = kv_store.snapshot_unsafe_maps(tx.commit_version()); + snap = kv_store.snapshot_unsafe_maps(tx.commit_version(), {}); } auto serialised_snap = kv_store.serialise_snapshot(std::move(snap)); diff --git a/src/kv/test/kv_dynamic_tables.cpp b/src/kv/test/kv_dynamic_tables.cpp index 4f22565b2777..c922d8f71315 100644 --- a/src/kv/test/kv_dynamic_tables.cpp +++ b/src/kv/test/kv_dynamic_tables.cpp @@ -539,7 +539,7 @@ TEST_CASE("Dynamic map snapshot serialisation" * doctest::test_suite("dynamic")) std::unique_ptr snapshot = nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&store); - snapshot = store.snapshot_unsafe_maps(snapshot_version); + snapshot = store.snapshot_unsafe_maps(snapshot_version, {}); } auto serialised_snapshot = store.serialise_snapshot(std::move(snapshot)); diff --git a/src/kv/test/kv_snapshot.cpp b/src/kv/test/kv_snapshot.cpp index 08ed6f959e98..8b33abfb6dde 100644 --- a/src/kv/test/kv_snapshot.cpp +++ b/src/kv/test/kv_snapshot.cpp @@ -57,7 +57,7 @@ TEST_CASE("Simple snapshot" * doctest::test_suite("snapshot")) nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&store); - first_snapshot = store.snapshot_unsafe_maps(first_snapshot_version); + first_snapshot = store.snapshot_unsafe_maps(first_snapshot_version, {}); } auto first_serialised_snapshot = store.serialise_snapshot(std::move(first_snapshot)); @@ -118,7 +118,7 @@ TEST_CASE("Simple snapshot" * doctest::test_suite("snapshot")) nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&store); - second_snapshot = store.snapshot_unsafe_maps(second_snapshot_version); + second_snapshot = store.snapshot_unsafe_maps(second_snapshot_version, {}); } auto second_serialised_snapshot = store.serialise_snapshot(std::move(second_snapshot)); @@ -278,7 +278,7 @@ TEST_CASE( std::unique_ptr snapshot = nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&store); - snapshot = store.snapshot_unsafe_maps(snapshot_version); + snapshot = store.snapshot_unsafe_maps(snapshot_version, {}); } auto serialised_snapshot = store.serialise_snapshot(std::move(snapshot)); @@ -389,7 +389,7 @@ TEST_CASE("Commit hooks with snapshot" * doctest::test_suite("snapshot")) std::unique_ptr snapshot = nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&store); - snapshot = store.snapshot_unsafe_maps(snapshot_version); + snapshot = store.snapshot_unsafe_maps(snapshot_version, {}); } auto serialised_snapshot = store.serialise_snapshot(std::move(snapshot)); @@ -522,7 +522,7 @@ TEST_CASE("Commit hooks with snapshot" * doctest::test_suite("snapshot")) snapshot_version = tx.commit_version(); { ccf::kv::ScopedStoreMapsLock maps_lock(&store); - snapshot = store.snapshot_unsafe_maps(snapshot_version); + snapshot = store.snapshot_unsafe_maps(snapshot_version, {}); } serialised_snapshot = store.serialise_snapshot(std::move(snapshot)); diff --git a/src/kv/test/stub_consensus.h b/src/kv/test/stub_consensus.h index 5d40a1558183..a09a7340fcab 100644 --- a/src/kv/test/stub_consensus.h +++ b/src/kv/test/stub_consensus.h @@ -38,28 +38,27 @@ namespace ccf::kv::test State state; NodeId local_id; - StubConsensus() : replica(), state(Backup), local_id(PrimaryNodeId) {} + explicit StubConsensus(State state_ = Primary) : + replica(), + state(state_), + local_id(PrimaryNodeId) + {} - virtual NodeId id() override + virtual NodeId id() { return local_id; } - virtual bool is_primary() override + virtual bool is_primary() { return state == Primary; } - virtual bool is_candidate() override + virtual bool is_candidate() { return state == Candidate; } - virtual bool can_replicate() override - { - return state == Primary; - } - virtual bool is_at_max_capacity() override { return false; @@ -77,7 +76,7 @@ namespace ccf::kv::test } } - virtual bool is_backup() override + virtual bool is_backup() { return state == Backup; } @@ -107,6 +106,11 @@ namespace ccf::kv::test bool replicate(const BatchVector& entries, ccf::View view) override { + if (state != Primary) + { + return false; + } + for (const auto& entry : entries) { replica.push_back(entry); @@ -162,39 +166,35 @@ namespace ccf::kv::test replica.clear(); } - std::pair get_committed_txid() override + virtual std::pair get_committed_txid() { return {committed_txid.view, committed_txid.seqno}; } - ccf::SeqNo get_committed_seqno() override + virtual ccf::SeqNo get_committed_seqno() { return committed_txid.seqno; } - std::optional primary() override + virtual std::optional primary() { return PrimaryNodeId; } - ccf::View get_view(ccf::SeqNo seqno) override - { - return view_history.view_at(seqno); - } - - ccf::View get_view() override + virtual ccf::View get_view() { return current_view; } - std::vector get_view_history(ccf::SeqNo seqno) override + ccf::TxStatus evaluate_tx_status( + ccf::View target_view, ccf::SeqNo target_seqno) override { - return view_history.get_history_until(seqno); - } - - std::vector get_view_history_since(ccf::SeqNo seqno) override - { - return view_history.get_history_since(seqno); + return ccf::evaluate_tx_status( + target_view, + target_seqno, + view_history.view_at(target_seqno), + committed_txid.view, + committed_txid.seqno); } void recv_message( @@ -202,22 +202,38 @@ namespace ccf::kv::test {} void add_configuration( - ccf::SeqNo seqno, const Configuration::Nodes& conf) override + ccf::SeqNo seqno, const Configuration::Nodes& configuration) override {} - Configuration::Nodes get_latest_configuration_unsafe() const override + void update_configuration( + ccf::SeqNo seqno, const Configuration::NodeChanges& changes) override + {} + + virtual Configuration::Nodes get_latest_configuration() { return {}; } - Configuration::Nodes get_latest_configuration() override + ConsensusLightDetails get_light_details() override { - return {}; + ConsensusLightDetails details; + details.membership_state = MembershipState::Active; + details.leadership_state = state == Primary ? LeadershipState::Leader : + state == Candidate ? LeadershipState::Candidate : + LeadershipState::Follower; + details.primary_id = PrimaryNodeId; + details.current_view = current_view; + details.committed_view = committed_txid.view; + details.committed_seqno = committed_txid.seqno; + return details; } ConsensusDetails get_details() override { - return ConsensusDetails{{}, {}, MembershipState::Active}; + ConsensusDetails details; + static_cast(details) = get_light_details(); + details.view_history.starts = view_history.get_history_until(); + return details; } void set_last_signature_at(ccf::SeqNo seqno) @@ -229,47 +245,12 @@ namespace ccf::kv::test class BackupStubConsensus : public StubConsensus { public: - BackupStubConsensus() : StubConsensus() {} - - bool is_primary() override - { - return false; - } - - bool replicate(const BatchVector& entries, ccf::View view) override - { - return false; - } - - bool can_replicate() override - { - return false; - } - - Consensus::SignatureDisposition get_signature_disposition() override - { - return Consensus::SignatureDisposition::CANT_REPLICATE; - } + BackupStubConsensus() : StubConsensus(Backup) {} }; class PrimaryStubConsensus : public StubConsensus { public: - PrimaryStubConsensus() : StubConsensus() {} - - bool is_primary() override - { - return true; - } - - bool can_replicate() override - { - return true; - } - - Consensus::SignatureDisposition get_signature_disposition() override - { - return Consensus::SignatureDisposition::CAN_SIGN; - } + PrimaryStubConsensus() : StubConsensus(Primary) {} }; } diff --git a/src/node/historical_queries.h b/src/node/historical_queries.h index 55fd7b7c81f0..00cee8f69c5b 100644 --- a/src/node/historical_queries.h +++ b/src/node/historical_queries.h @@ -1348,7 +1348,8 @@ namespace ccf::historical return false; } - const auto actual_view = consensus->get_view(seqno); + const auto actual_view = + consensus->get_details().view_history.view_at(seqno); if (actual_view != tx_id.view) { LOG_FAIL_FMT( diff --git a/src/node/hooks.h b/src/node/hooks.h index abf10379d96c..3bc6922f06ef 100644 --- a/src/node/hooks.h +++ b/src/node/hooks.h @@ -12,16 +12,10 @@ namespace ccf { - struct NodeAddr - { - std::string hostname; - std::string port; - }; - class ConfigurationChangeHook : public ccf::kv::ConsensusHook { ccf::kv::Version version; - std::map> cfg_delta; + ccf::kv::Configuration::NodeChanges cfg_delta; public: ConfigurationChangeHook(ccf::kv::Version version_, const Nodes::Write& w) : @@ -48,7 +42,8 @@ namespace ccf } case NodeStatus::TRUSTED: { - cfg_delta.try_emplace(node_id, NodeAddr{host, port}); + cfg_delta.try_emplace( + node_id, ccf::kv::Configuration::NodeInfo{host, port}); break; } case NodeStatus::RETIRED: @@ -69,22 +64,7 @@ namespace ccf void call(ccf::kv::ConfigurableConsensus* consensus) override { - auto configuration = consensus->get_latest_configuration_unsafe(); - for (const auto& [node_id, opt_ni] : cfg_delta) - { - if (opt_ni.has_value()) - { - configuration.try_emplace(node_id, opt_ni->hostname, opt_ni->port); - } - else - { - configuration.erase(node_id); - } - } - if (!cfg_delta.empty()) - { - consensus->add_configuration(version, configuration); - } + consensus->update_configuration(version, cfg_delta); } }; } diff --git a/src/node/jwt_key_auto_refresh.h b/src/node/jwt_key_auto_refresh.h index f1b67abf575d..e6d54cbd5334 100644 --- a/src/node/jwt_key_auto_refresh.h +++ b/src/node/jwt_key_auto_refresh.h @@ -112,7 +112,7 @@ namespace ccf return; } - if (!self_sp->consensus->can_replicate()) + if (!self_sp->consensus->get_light_details().can_replicate()) { LOG_DEBUG_FMT("JWT key auto-refresh: Node is not primary, skipping"); } @@ -150,7 +150,7 @@ namespace ccf return; } - if (!self_sp->consensus->can_replicate()) + if (!self_sp->consensus->get_light_details().can_replicate()) { LOG_DEBUG_FMT( "JWT key one-off refresh: Node is not primary, skipping"); diff --git a/src/node/node_state.h b/src/node/node_state.h index eb4f64e7a911..9e77d68af974 100644 --- a/src/node/node_state.h +++ b/src/node/node_state.h @@ -287,7 +287,8 @@ namespace ccf std::string primary_address; std::vector service_cert; { - auto primary_id = owner->consensus->primary(); + const auto primary_id = + owner->consensus->get_light_details().primary_id; if (!primary_id.has_value()) { LOG_INFO_FMT( @@ -2258,7 +2259,7 @@ namespace ccf LOG_INFO_FMT( "Try end private recovery at {}. Is primary: {}", recovery_v, - consensus->is_primary()); + consensus->get_light_details().is_primary()); if (recovery_v != recovery_store->current_version()) { @@ -2287,7 +2288,7 @@ namespace ccf snapshotter->set_snapshot_generation(true); // Open the service - if (consensus->can_replicate()) + if (consensus->get_light_details().can_replicate()) { LOG_INFO_FMT( "Try end private recovery at {}. Trigger service opening", @@ -2699,8 +2700,8 @@ namespace ccf if (sm.check(NodeStartupState::partOfNetwork)) { - const auto tx_id = consensus->get_committed_txid(); - indexer->update_strategies(elapsed, {tx_id.first, tx_id.second}); + const auto details = consensus->get_light_details(); + indexer->update_strategies(elapsed, details.committed_txid()); } n2n_channels->tick(elapsed); @@ -2753,7 +2754,7 @@ namespace ccf (sm.check(NodeStartupState::partOfNetwork) || sm.check(NodeStartupState::partOfPublicNetwork) || sm.check(NodeStartupState::readingPrivateLedger)) && - consensus->is_primary()); + consensus->get_light_details().is_primary()); } bool can_replicate() override @@ -2762,12 +2763,12 @@ namespace ccf (sm.check(NodeStartupState::partOfNetwork) || sm.check(NodeStartupState::partOfPublicNetwork) || sm.check(NodeStartupState::readingPrivateLedger)) && - consensus->can_replicate()); + consensus->get_light_details().can_replicate()); } std::optional get_primary() override { - return consensus->primary(); + return consensus->get_light_details().primary_id; } [[nodiscard]] bool is_in_initialised_state() const override @@ -3675,7 +3676,7 @@ namespace ccf // backup, schedule a fetch task if ( config.snapshots.backup_fetch.enabled && consensus != nullptr && - !consensus->is_primary()) + !consensus->get_light_details().is_primary()) { ccf::tasks::Task task_to_schedule = nullptr; { diff --git a/src/node/recovery_decision_protocol.cpp b/src/node/recovery_decision_protocol.cpp index 4a2d56eac381..1a8333e59f34 100644 --- a/src/node/recovery_decision_protocol.cpp +++ b/src/node/recovery_decision_protocol.cpp @@ -723,9 +723,10 @@ namespace ccf ccf::TxID RecoveryDecisionProtocolSubsystem::get_last_recovered_signed_txid() { auto recovery_seqno = node_state->last_recovered_signed_idx; - auto recovery_view = node_state->consensus->get_view(recovery_seqno); - // get_view returns VIEW_UNKNOWN=InvalidView if the view is not in the view - // history (too old or too new) + auto recovery_view = + node_state->consensus->get_details().view_history.view_at(recovery_seqno); + // view_at returns VIEW_UNKNOWN if the sequence number is not in the view + // history (too old or too new). if (recovery_view == ccf::VIEW_UNKNOWN) { throw std::logic_error(fmt::format( diff --git a/src/node/rpc/frontend.h b/src/node/rpc/frontend.h index ec65f95ab601..52873fa70f55 100644 --- a/src/node/rpc/frontend.h +++ b/src/node/rpc/frontend.h @@ -241,7 +241,8 @@ namespace ccf } { - const auto primary_id = current_consensus->primary(); + const auto primary_id = + current_consensus->get_light_details().primary_id; if (seeking_primary && primary_id.has_value()) { target_node_its.push_back(nodes.find(primary_id.value())); @@ -309,8 +310,8 @@ namespace ccf case (ccf::endpoints::RedirectionStrategy::ToPrimary): { - const bool is_primary = - current_consensus != nullptr && current_consensus->can_replicate(); + const bool is_primary = current_consensus != nullptr && + current_consensus->get_light_details().can_replicate(); if (!is_primary) { @@ -344,8 +345,8 @@ namespace ccf case (ccf::endpoints::RedirectionStrategy::ToBackup): { - const bool is_backup = - current_consensus != nullptr && !current_consensus->can_replicate(); + const bool is_backup = current_consensus != nullptr && + current_consensus->get_light_details().is_backup(); if (!is_backup) { @@ -419,7 +420,8 @@ namespace ccf { if (current_consensus != nullptr) { - auto current_view = current_consensus->get_view(); + const auto current_view = + current_consensus->get_light_details().current_view; auto session_ctx = ctx->get_session_context(); if (!session_ctx->active_view.has_value()) { @@ -570,7 +572,7 @@ namespace ccf return; } - auto primary_id = current_consensus->primary(); + const auto primary_id = current_consensus->get_light_details().primary_id; if (!primary_id.has_value()) { ctx->set_error( @@ -794,7 +796,7 @@ namespace ccf else { bool is_primary = current_consensus == nullptr || - current_consensus->can_replicate(); + current_consensus->get_light_details().can_replicate(); const bool forwardable = current_consensus != nullptr; if (!is_primary && forwardable) @@ -960,7 +962,7 @@ namespace ccf if ( current_consensus != nullptr && - current_consensus->can_replicate() && + current_consensus->get_light_details().can_replicate() && current_history != nullptr) { current_history->try_emit_signature(); diff --git a/src/node/rpc/node_frontend.h b/src/node/rpc/node_frontend.h index f0bca921fd55..00a910d62d10 100644 --- a/src/node/rpc/node_frontend.h +++ b/src/node/rpc/node_frontend.h @@ -538,7 +538,8 @@ namespace ccf if ( current_consensus != nullptr && !this->node_operation.can_replicate()) { - auto primary_id = current_consensus->primary(); + const auto primary_id = + current_consensus->get_light_details().primary_id; if (primary_id.has_value()) { const auto address = node::get_redirect_address_for_node( @@ -897,8 +898,10 @@ namespace ccf auto* current_consensus = get_consensus(); if (current_consensus != nullptr) { - out.current_view = current_consensus->get_view(); - auto primary_id = current_consensus->primary(); + const auto consensus_details = + current_consensus->get_light_details(); + out.current_view = consensus_details.current_view; + const auto& primary_id = consensus_details.primary_id; if (primary_id.has_value()) { out.primary_id = primary_id.value(); @@ -1010,7 +1013,8 @@ namespace ccf bool is_primary = false; if (current_consensus != nullptr) { - is_primary = current_consensus->primary() == nid; + is_primary = + current_consensus->get_light_details().primary_id == nid; } out.nodes.push_back( @@ -1204,7 +1208,8 @@ namespace ccf auto* current_consensus = get_consensus(); if (current_consensus != nullptr) { - auto primary = current_consensus->primary(); + const auto primary = + current_consensus->get_light_details().primary_id; if (primary.has_value() && primary.value() == node_id) { is_primary = true; @@ -1236,7 +1241,8 @@ namespace ccf auto* current_consensus = get_consensus(); if (current_consensus != nullptr) { - auto primary = current_consensus->primary(); + const auto primary = + current_consensus->get_light_details().primary_id; if (primary.has_value() && primary.value() == node_id) { is_primary = true; @@ -1290,7 +1296,8 @@ namespace ccf auto* current_consensus = get_consensus(); if (current_consensus != nullptr) { - auto primary_id = current_consensus->primary(); + const auto primary_id = + current_consensus->get_light_details().primary_id; if (!primary_id.has_value()) { return make_error( @@ -1350,7 +1357,8 @@ namespace ccf return; } - auto primary_id = current_consensus->primary(); + const auto primary_id = + current_consensus->get_light_details().primary_id; if (!primary_id.has_value()) { args.rpc_ctx->set_error( @@ -1419,9 +1427,12 @@ namespace ccf auto* current_consensus = get_consensus(); if (current_consensus != nullptr) { - auto cfg = current_consensus->get_latest_configuration(); + const auto details = current_consensus->get_details(); + const auto cfg = details.configs.empty() ? + ccf::kv::Configuration::Nodes{} : + details.configs.back().nodes; ConsensusConfig cc; - for (auto& [nid, ninfo] : cfg) + for (const auto& [nid, ninfo] : cfg) { cc.emplace( nid.value(), @@ -1714,7 +1725,8 @@ namespace ccf // All errors are server errors since the client is the server. auto* current_consensus = get_consensus(); - auto primary_id = current_consensus->primary(); + const auto primary_id = + current_consensus->get_light_details().primary_id; if (!primary_id.has_value()) { LOG_FAIL_FMT("JWT key auto-refresh: primary unknown"); diff --git a/src/node/snapshotter.h b/src/node/snapshotter.h index 5b437c2b1901..0547edff5b65 100644 --- a/src/node/snapshotter.h +++ b/src/node/snapshotter.h @@ -598,7 +598,10 @@ namespace ccf }); } - void schedule_snapshot(::consensus::Index idx, TimePoint timestamp) + void schedule_snapshot( + ::consensus::Index idx, + TimePoint timestamp, + const ccf::kv::ConsensusViewHistory& view_history) { // Called with lock held (from commit()). static uint32_t generation_count = 0; @@ -623,13 +626,16 @@ namespace ccf info.tasks->add_action(std::make_shared( shared_from_this(), - store->snapshot_unsafe_maps(idx), + store->snapshot_unsafe_maps(idx, view_history.until(idx)), generation, timestamp, info.serialised)); } - void commit(::consensus::Index idx, bool generate_snapshot) override + void commit( + ::consensus::Index idx, + bool generate_snapshot, + const ccf::kv::ConsensusViewHistory& view_history) override { // If generate_snapshot is true, takes a snapshot of the key value store // at the last snapshottable index before idx, and schedule snapshot @@ -673,11 +679,11 @@ namespace ccf LOG_FAIL_FMT( "Could not find scheduled snapshot time for idx {}", next.idx); scheduled_snapshot_times[next.idx] = timestamp; - schedule_snapshot(next.idx, timestamp); + schedule_snapshot(next.idx, timestamp, view_history); } else { - schedule_snapshot(next.idx, snapshot_time->second); + schedule_snapshot(next.idx, snapshot_time->second, view_history); } next.done = true; } diff --git a/src/node/test/history.cpp b/src/node/test/history.cpp index 39c955488238..e81469891d81 100644 --- a/src/node/test/history.cpp +++ b/src/node/test/history.cpp @@ -286,11 +286,6 @@ class CompactingConsensus : public ccf::kv::test::StubConsensus { return ccf::kv::test::PrimaryNodeId; } - - ccf::View get_view(ccf::kv::Version version) override - { - return 2; - } }; class TestPendingTx : public ccf::kv::PendingTx @@ -460,11 +455,6 @@ class RollbackConsensus : public ccf::kv::test::StubConsensus return ccf::kv::test::PrimaryNodeId; } - ccf::View get_view(ccf::SeqNo seqno) override - { - return 2; - } - ccf::View get_view() override { return 2; diff --git a/src/node/test/snapshot.cpp b/src/node/test/snapshot.cpp index 40aae786ccec..968cf08b5c2e 100644 --- a/src/node/test/snapshot.cpp +++ b/src/node/test/snapshot.cpp @@ -113,7 +113,10 @@ TEST_CASE("Snapshot with merkle tree" * doctest::test_suite("snapshot")) nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&source_store); - snapshot = source_store.snapshot_unsafe_maps(snapshot_version - 1); + snapshot = source_store.snapshot_unsafe_maps( + snapshot_version - 1, + {source_consensus->view_history.get_history_until( + snapshot_version - 1)}); } auto serialised_snapshot = source_store.serialise_snapshot(std::move(snapshot)); @@ -135,7 +138,9 @@ TEST_CASE("Snapshot with merkle tree" * doctest::test_suite("snapshot")) nullptr; { ccf::kv::ScopedStoreMapsLock maps_lock(&source_store); - snapshot = source_store.snapshot_unsafe_maps(snapshot_version); + snapshot = source_store.snapshot_unsafe_maps( + snapshot_version, + {source_consensus->view_history.get_history_until(snapshot_version)}); } auto serialised_snapshot = source_store.serialise_snapshot(std::move(snapshot)); diff --git a/src/node/test/snapshotter.cpp b/src/node/test/snapshotter.cpp index 158d5f8fa045..11dcad0b6fe6 100644 --- a/src/node/test/snapshotter.cpp +++ b/src/node/test/snapshotter.cpp @@ -428,7 +428,8 @@ TEST_CASE("Regular snapshotting") { REQUIRE_FALSE(record_signature(history, snapshotter, snapshot_idx - 1)); commit_idx = snapshot_idx - 1; - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE_THROWS_AS( @@ -445,7 +446,8 @@ TEST_CASE("Regular snapshotting") // Note: even if commit_idx > snapshot_tx_interval, the snapshot is // generated at snapshot_idx commit_idx = snapshot_idx + 1; - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); run_one_task(); // Snapshot evidence is committed to the KV, but the snapshot is not @@ -460,7 +462,8 @@ TEST_CASE("Regular snapshotting") record_snapshot_evidence(snapshotter, snapshot_idx, snapshot_evidence_idx); commit_idx = snapshot_idx + 2; REQUIRE_FALSE(record_signature(history, snapshotter, commit_idx)); - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); // The persist action runs on the task system once commit evidence is // durable run_one_task(); @@ -474,7 +477,8 @@ TEST_CASE("Regular snapshotting") INFO("Subsequent commit before next snapshot idx has no effect"); { commit_idx = snapshot_idx + 2; - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE( latest_committed_snapshot_idx(snapshot_dir.path) == @@ -490,7 +494,8 @@ TEST_CASE("Regular snapshotting") REQUIRE(record_signature(history, snapshotter, snapshot_idx)); // Note: Commit exactly on snapshot idx commit_idx = snapshot_idx; - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE(read_latest_snapshot_evidence(network.tables) == snapshot_idx); @@ -507,7 +512,8 @@ TEST_CASE("Regular snapshotting") commit_idx = snapshot_idx + 2; REQUIRE_FALSE(record_signature(history, snapshotter, commit_idx)); - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE(latest_committed_snapshot_idx(snapshot_dir.path) == snapshot_idx); REQUIRE( @@ -545,7 +551,8 @@ TEST_CASE("Rollback before snapshot is committed") { snapshot_idx = snapshot_tx_interval; REQUIRE(record_signature(history, snapshotter, snapshot_idx)); - snapshotter->commit(snapshot_idx, true); + snapshotter->commit( + snapshot_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE(read_latest_snapshot_evidence(network.tables) == snapshot_idx); @@ -559,12 +566,14 @@ TEST_CASE("Rollback before snapshot is committed") // ... More transactions are committed, passing the idx at which the // evidence was originally committed - snapshotter->commit(snapshot_tx_interval + 1, true); + snapshotter->commit( + snapshot_tx_interval + 1, true, consensus->get_details().view_history); // Snapshot previously generated is not committed REQUIRE_FALSE(latest_committed_snapshot_idx(snapshot_dir.path).has_value()); - snapshotter->commit(snapshot_tx_interval + 2, true); + snapshotter->commit( + snapshot_tx_interval + 2, true, consensus->get_details().view_history); REQUIRE_FALSE(latest_committed_snapshot_idx(snapshot_dir.path).has_value()); } @@ -574,7 +583,8 @@ TEST_CASE("Rollback before snapshot is committed") size_t new_snapshot_idx = network.tables->current_version(); REQUIRE(record_signature(history, snapshotter, new_snapshot_idx)); - snapshotter->commit(new_snapshot_idx, true); + snapshotter->commit( + new_snapshot_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE(read_latest_snapshot_evidence(network.tables) == new_snapshot_idx); @@ -586,7 +596,8 @@ TEST_CASE("Rollback before snapshot is committed") record_snapshot_evidence( snapshotter, new_snapshot_idx, new_snapshot_idx + 1); REQUIRE_FALSE(record_signature(history, snapshotter, commit_idx)); - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE( latest_committed_snapshot_idx(snapshot_dir.path) == new_snapshot_idx); @@ -601,7 +612,8 @@ TEST_CASE("Rollback before snapshot is committed") ccf::kv::AbstractStore::StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE); REQUIRE(record_signature(history, snapshotter, new_snapshot_idx)); - snapshotter->commit(new_snapshot_idx, true); + snapshotter->commit( + new_snapshot_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE(read_latest_snapshot_evidence(network.tables) == new_snapshot_idx); @@ -618,7 +630,8 @@ TEST_CASE("Rollback before snapshot is committed") record_snapshot_evidence( snapshotter, new_snapshot_idx, new_snapshot_idx + 1); REQUIRE_FALSE(record_signature(history, snapshotter, commit_idx)); - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); run_one_task(); REQUIRE( latest_committed_snapshot_idx(snapshot_dir.path) == new_snapshot_idx); @@ -676,7 +689,8 @@ TEST_CASE("Snapshot status updates preserve future queued snapshot") REQUIRE_FALSE( record_signature(history, snapshotter, network.tables->current_version())); - snapshotter->commit(2 * snapshot_tx_interval, true); + snapshotter->commit( + 2 * snapshot_tx_interval, true, consensus->get_details().view_history); run_one_task(); // The snapshot was generated at the expected idx, as confirmed by the @@ -766,7 +780,8 @@ TEST_CASE("Rekey ledger while snapshot is in progress") tx.commit(); REQUIRE(record_signature(history, snapshotter, snapshot_idx)); - snapshotter->commit(snapshot_idx, true); + snapshotter->commit( + snapshot_idx, true, consensus->get_details().view_history); // Do not schedule task just yet so that we can interleave ledger rekey } @@ -790,7 +805,8 @@ TEST_CASE("Rekey ledger while snapshot is in progress") record_snapshot_evidence(snapshotter, snapshot_idx, snapshot_idx + 1); auto commit_idx = snapshot_idx + 2; REQUIRE_FALSE(record_signature(history, snapshotter, commit_idx)); - snapshotter->commit(commit_idx, true); + snapshotter->commit( + commit_idx, true, consensus->get_details().view_history); // The persist action runs on the task system, writing the serialised // snapshot bytes to disk.