From 3678ecebac5aee6c8750f7c7926ff0d251306ef9 Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Thu, 20 Aug 2026 14:34:37 +0000 Subject: [PATCH 1/9] Agent Host changes for agents/issue-8184-implementation --- CHANGELOG.md | 1 + doc/schemas/node_openapi.json | 6 + src/consensus/aft/raft.h | 112 ++++++---------- src/consensus/aft/raft_types.h | 6 +- src/consensus/aft/test/committable_suffix.cpp | 106 +++++++-------- src/consensus/aft/test/driver.h | 36 ++--- src/consensus/aft/test/logging_stub.h | 4 +- src/consensus/aft/test/main.cpp | 126 ++++++++++++++++-- src/endpoints/base_endpoint_registry.cpp | 10 +- src/kv/kv_types.h | 86 +++++++----- src/kv/store.h | 10 +- src/kv/test/stub_consensus.h | 63 +++------ src/node/node_state.h | 18 +-- src/node/rpc/frontend.h | 8 +- src/node/rpc/node_frontend.h | 32 +++-- 15 files changed, 363 insertions(+), 261 deletions(-) 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..9afa960682fc 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" }, diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 286a3e0ebb8c..de48133d579d 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -21,6 +21,7 @@ #include "service/tables/signatures.h" #include +#include #include #include #include @@ -201,6 +202,11 @@ namespace aft // pre-deserialisation, without an additional header. static constexpr size_t max_terms_per_append_entries = 1; + void set_leadership_state(ccf::kv::LeadershipState new_state) + { + std::atomic_ref(state->leadership_state).store(new_state); + } + public: static constexpr size_t append_entries_size_limit = 20000; std::unique_ptr ledger; @@ -247,26 +253,11 @@ namespace aft ~Aft() override = default; - std::optional primary() override - { - return leader_id; - } - ccf::NodeId id() override { 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); @@ -303,16 +294,13 @@ namespace aft return Consensus::SignatureDisposition::CANT_REPLICATE; } - bool is_backup() override - { - return state->leadership_state == ccf::kv::LeadershipState::Follower; - } - - bool is_active() const + bool is_primary() const { - return state->membership_state == ccf::kv::MembershipState::Active; + return std::atomic_ref(state->leadership_state).load() == + ccf::kv::LeadershipState::Leader; } + private: bool is_retired() const { return state->membership_state == ccf::kv::MembershipState::Retired; @@ -324,12 +312,7 @@ 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 { @@ -464,25 +447,6 @@ namespace aft return state->last_idx; } - Index get_committed_seqno() override - { - 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; - } - - std::pair get_committed_txid() override - { - std::lock_guard guard(state->lock); - ccf::SeqNo commit_idx = get_commit_idx_unsafe(); - return {get_term_internal(commit_idx), commit_idx}; - } - Term get_view(Index idx) override { std::lock_guard guard(state->lock); @@ -589,18 +553,15 @@ namespace aft return configurations.back().nodes; } - Configuration::Nodes get_latest_configuration() override - { - std::lock_guard guard(state->lock); - return get_latest_configuration_unsafe(); - } - - 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 +569,29 @@ 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()); 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; } @@ -2102,7 +2076,7 @@ namespace aft return; } - state->leadership_state = ccf::kv::LeadershipState::PreVoteCandidate; + set_leadership_state(ccf::kv::LeadershipState::PreVoteCandidate); leader_id.reset(); reset_votes_for_me(); @@ -2147,7 +2121,7 @@ namespace aft return; } - state->leadership_state = ccf::kv::LeadershipState::Candidate; + set_leadership_state(ccf::kv::LeadershipState::Candidate); leader_id.reset(); voted_for = state->node_id; @@ -2204,7 +2178,7 @@ namespace aft store->initialise_term(state->current_view); } - state->leadership_state = ccf::kv::LeadershipState::Leader; + set_leadership_state(ccf::kv::LeadershipState::Leader); leader_id = state->node_id; should_sign = true; @@ -2258,7 +2232,7 @@ namespace aft restart_election_timeout(); reset_last_ack_timeouts(); - state->leadership_state = ccf::kv::LeadershipState::Follower; + set_leadership_state(ccf::kv::LeadershipState::Follower); RAFT_INFO_FMT( "Becoming follower {}: {}.{}", state->node_id, @@ -2381,7 +2355,7 @@ namespace aft nominate_successor(); leader_id.reset(); - state->leadership_state = ccf::kv::LeadershipState::None; + set_leadership_state(ccf::kv::LeadershipState::None); } state->membership_state = ccf::kv::MembershipState::Retired; @@ -2607,7 +2581,7 @@ namespace aft } RAFT_DEBUG_FMT("Compacting..."); - store->compact(idx); + store->compact(idx, is_primary()); ledger->commit(idx); if (commit_callbacks != nullptr) @@ -2650,7 +2624,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..4c1c23d745f7 100644 --- a/src/consensus/aft/raft_types.h +++ b/src/consensus/aft/raft_types.h @@ -26,7 +26,7 @@ namespace aft { public: virtual ~Store() = default; - virtual void compact(Index v) = 0; + virtual void compact(Index v, bool is_primary = false) = 0; virtual void rollback( const ccf::TxID& tx_id, Term term_of_next_version) = 0; virtual void initialise_term(Term t) = 0; @@ -45,12 +45,12 @@ namespace aft public: Adaptor(std::shared_ptr x) : x(x) {} - void compact(Index v) override + void compact(Index v, bool is_primary = false) override { auto p = x.lock(); if (p) { - p->compact(v); + p->compact(v, is_primary); } } diff --git a/src/consensus/aft/test/committable_suffix.cpp b/src/consensus/aft/test/committable_suffix.cpp index 3571bedea2e5..3d1aea29ca47 100644 --- a/src/consensus/aft/test/committable_suffix.cpp +++ b/src/consensus/aft/test/committable_suffix.cpp @@ -195,8 +195,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rA.is_primary()); - DOCTEST_REQUIRE(rA.get_view() == 1); + DOCTEST_REQUIRE(rA.get_light_details().is_primary()); + DOCTEST_REQUIRE(rA.get_light_details().current_view == 1); // Dispatch initial AppendEntries DOCTEST_REQUIRE(4 == dispatch_all(nodes, node_idA, channelsA->messages)); @@ -207,8 +207,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rA.is_primary()); - DOCTEST_REQUIRE(rA.get_view() == 1); + DOCTEST_REQUIRE(rA.get_light_details().is_primary()); + DOCTEST_REQUIRE(rA.get_light_details().current_view == 1); } DOCTEST_INFO("Entry at 1.1 is received by all nodes"); @@ -216,7 +216,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") auto entry = make_ledger_entry(1, 1); rA.replicate(ccf::kv::BatchVector{{1, entry, true, hooks}}, 1); DOCTEST_REQUIRE(rA.get_last_idx() == 1); - DOCTEST_REQUIRE(rA.get_committed_seqno() == 0); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 0); // Size limit was reached, so periodic is not needed // rA.periodic(request_timeout); @@ -236,7 +236,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); // Node A now knows this is committed - DOCTEST_REQUIRE(rA.get_committed_seqno() == 1); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 1); } DOCTEST_INFO( @@ -246,21 +246,21 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") auto entry = make_ledger_entry(1, 2); rA.replicate(ccf::kv::BatchVector{{2, entry, true, hooks}}, 1); DOCTEST_REQUIRE(rA.get_last_idx() == 2); - DOCTEST_REQUIRE(rA.get_committed_seqno() == 1); + DOCTEST_REQUIRE(rA.get_light_details().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); DOCTEST_REQUIRE(rA.get_last_idx() == 3); - DOCTEST_REQUIRE(rA.get_committed_seqno() == 1); + DOCTEST_REQUIRE(rA.get_light_details().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); DOCTEST_REQUIRE(rA.get_last_idx() == 4); - DOCTEST_REQUIRE(rA.get_committed_seqno() == 1); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 1); // Size limit was reached, so periodic is not needed // rA.periodic(request_timeout); @@ -329,12 +329,12 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(0 == dispatch_all(nodes, node_idE, channelsE->messages)); // Node A now knows that 1.4 is committed - DOCTEST_REQUIRE(rA.get_committed_seqno() == 4); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 4); // Nodes B and C have this commit index, and are responsible for persisting // it - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); - DOCTEST_REQUIRE(rC.get_last_idx() >= rA.get_committed_seqno()); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rC.get_last_idx() >= rA.get_light_details().committed_seqno); } DOCTEST_INFO("Node A dies"); @@ -360,8 +360,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rB.is_primary()); - DOCTEST_REQUIRE(rB.get_view() == 2); + DOCTEST_REQUIRE(rB.get_light_details().is_primary()); + DOCTEST_REQUIRE(rB.get_light_details().current_view == 2); } DOCTEST_INFO("Node B writes some entries, though they are lost"); @@ -383,8 +383,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") // The key features is that B is one of the quorum responsible for // persistence of 1.4, despites its commit index not being as high as 1.4 - DOCTEST_REQUIRE(rB.get_committed_seqno() < rA.get_committed_seqno()); - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); + DOCTEST_REQUIRE(rB.get_light_details().committed_seqno < rA.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); } DOCTEST_INFO("Node C wins an election"); @@ -407,8 +407,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rC.is_primary()); - DOCTEST_REQUIRE(rC.get_view() == 3); + DOCTEST_REQUIRE(rC.get_light_details().is_primary()); + DOCTEST_REQUIRE(rC.get_light_details().current_view == 3); DOCTEST_REQUIRE(rB.get_last_idx() == 7); DOCTEST_REQUIRE(rC.get_last_idx() == 4); @@ -419,8 +419,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(rB.get_last_idx() == 7); - DOCTEST_REQUIRE(rB.get_committed_seqno() < rA.get_committed_seqno()); - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); + DOCTEST_REQUIRE(rB.get_light_details().committed_seqno < rA.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); } DOCTEST_REQUIRE("Node C produces 3.5, 3.6, and 3.7"); @@ -479,7 +479,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(rB.get_last_idx() != tail_of_b); // B must still be holding the committed index it holds from A - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); // B's term history must match the current primary's DOCTEST_REQUIRE(rB.get_last_idx() <= rC.get_last_idx()); @@ -541,8 +541,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idB)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idC)); - DOCTEST_REQUIRE(rA.is_primary()); - DOCTEST_REQUIRE(rA.get_view() == 1); + DOCTEST_REQUIRE(rA.get_light_details().is_primary()); + DOCTEST_REQUIRE(rA.get_light_details().current_view == 1); // Election-triggered heartbeats DOCTEST_REQUIRE(2 == dispatch_all(nodes, node_idA)); @@ -561,7 +561,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // If C is in an older term, it gets a heartbeat to join this primary's // term, but nothing more - if (rC.get_view() < primary.get_view()) + if (rC.get_light_details().current_view < primary.get_light_details().current_view) { primary.periodic(request_timeout); keep_messages_for(node_idC, channels_primary->messages); @@ -569,13 +569,13 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) channelsC->messages.clear(); } - DOCTEST_REQUIRE(rC.get_view() >= primary.get_view()); + DOCTEST_REQUIRE(rC.get_light_details().current_view >= primary.get_light_details().current_view); - if (rC.get_view() > primary.get_view()) + if (rC.get_light_details().current_view > primary.get_light_details().current_view) { // Trigger a message from the intended primary, so C will respond with its // current term - if (primary.is_primary()) + if (primary.get_light_details().is_primary()) { // If we were already primary, then request_timeout will produce an // AppendEntries @@ -595,7 +595,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) keep_messages_for(primary_id, channelsC->messages); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idC)); - DOCTEST_REQUIRE(rC.get_view() == primary.get_view()); + DOCTEST_REQUIRE(rC.get_light_details().current_view == primary.get_light_details().current_view); } else { @@ -632,14 +632,14 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) })); // That's sufficient to win this election - DOCTEST_REQUIRE(primary.is_primary()); + DOCTEST_REQUIRE(primary.get_light_details().is_primary()); const auto start_idx = primary.get_last_idx(); for (auto idx = start_idx + 1; idx <= start_idx + num_entries; ++idx) { - auto entry = make_ledger_entry(primary.get_view(), idx); + auto entry = make_ledger_entry(primary.get_light_details().current_view, idx); primary.replicate( - ccf::kv::BatchVector{{idx, entry, true, hooks}}, primary.get_view()); + ccf::kv::BatchVector{{idx, entry, true, hooks}}, primary.get_light_details().current_view); } // All related AppendEntries are lost @@ -672,7 +672,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) entry = make_ledger_entry(1, 2); rA.replicate(ccf::kv::BatchVector{{2, entry, true, hooks}}, 1); DOCTEST_REQUIRE(rA.get_last_idx() == 2); - DOCTEST_REQUIRE(rA.get_committed_seqno() == 0); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 0); // Size limit was reached, so periodic is not needed // rA.periodic(request_timeout); @@ -687,7 +687,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(rC.get_last_idx() == 2); // And primary knows it is committed - DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); // After a periodic heartbeat rA.periodic(request_timeout); @@ -696,9 +696,9 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idC)); // All nodes know that this is committed - DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); - DOCTEST_REQUIRE(rB.get_committed_seqno() == 2); - DOCTEST_REQUIRE(rC.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rC.get_light_details().committed_seqno == 2); // Node A produces 2 additional entries that A and B have, and 2 additional // entries that are only present on A @@ -724,9 +724,9 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // Commit did not advance, though 4 is present on f+1 nodes and will be // persisted from here - DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); - DOCTEST_REQUIRE(rB.get_committed_seqno() == 2); - DOCTEST_REQUIRE(rC.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rC.get_light_details().committed_seqno == 2); persisted_idx = 4; persisted_entry = rB.ledger->ledger[persisted_idx - 1]; @@ -752,15 +752,15 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // Nodes A and B now have long, distinct, multi-term non-committed suffixes. // Node C has not advanced its log at all - DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); - DOCTEST_REQUIRE(rB.get_committed_seqno() == 2); - DOCTEST_REQUIRE(rC.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rC.get_light_details().committed_seqno == 2); DOCTEST_REQUIRE(rA.get_last_idx() > 4); DOCTEST_REQUIRE(rB.get_last_idx() > 3); DOCTEST_REQUIRE(rC.get_last_idx() == 2); - DOCTEST_REQUIRE(rA.get_view() != rB.get_view()); + DOCTEST_REQUIRE(rA.get_light_details().current_view != rB.get_light_details().current_view); DOCTEST_REQUIRE( rA.get_view_history(rA.get_last_idx()) != rB.get_view_history(rB.get_last_idx())); @@ -792,7 +792,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) { aim_for_a_primary = true; DOCTEST_INFO("Node A wins"); - while (rA.get_view() <= rC.get_view()) + while (rA.get_light_details().current_view <= rC.get_light_details().current_view) { channelsA->messages.clear(); rA.periodic(election_timeout); @@ -806,7 +806,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) { aim_for_a_primary = false; DOCTEST_INFO("Node B wins"); - while (rB.get_view() <= rC.get_view()) + while (rB.get_light_details().current_view <= rC.get_light_details().current_view) { channelsB->messages.clear(); rB.periodic(election_timeout); @@ -828,7 +828,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) dispatch_all(nodes, id_primary); dispatch_all(nodes, node_idC); - DOCTEST_REQUIRE(rPrimary.is_primary()); + DOCTEST_REQUIRE(rPrimary.get_light_details().is_primary()); { DOCTEST_INFO("Catch node C up"); @@ -867,7 +867,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) } DOCTEST_INFO("Bring other node in-sync"); - const auto id_other = rA.is_primary() ? node_idB : node_idA; + const auto id_other = rA.get_light_details().is_primary() ? node_idB : node_idA; { channelsA->messages.clear(); @@ -987,7 +987,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) { // One more entry in this term, and replication of it, and post-ACK // dispatch, to reach unanimous commit point - const auto view = rPrimary.get_view(); + const auto view = rPrimary.get_light_details().current_view; const auto seqno = rPrimary.get_last_idx() + 1; auto final_entry = make_ledger_entry(view, seqno); rPrimary.replicate( @@ -1012,8 +1012,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(ae.leader_commit_idx == seqno); }); - DOCTEST_REQUIRE(rPrimary.get_committed_seqno() == seqno); - DOCTEST_REQUIRE(rA.get_committed_seqno() == rB.get_committed_seqno()); + DOCTEST_REQUIRE(rPrimary.get_light_details().committed_seqno == seqno); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == rB.get_light_details().committed_seqno); } if constexpr (is_worst_case) @@ -1036,8 +1036,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(rA.get_last_idx() == rB.get_last_idx()); DOCTEST_REQUIRE(rB.get_last_idx() == rC.get_last_idx()); - DOCTEST_REQUIRE(rA.get_committed_seqno() == rB.get_committed_seqno()); - DOCTEST_REQUIRE(rB.get_committed_seqno() == rC.get_committed_seqno()); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == rB.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == rC.get_light_details().committed_seqno); const auto term_history_on_A = rA.get_view_history(rA.get_last_idx()); const auto term_history_on_B = rB.get_view_history(rB.get_last_idx()); @@ -1056,7 +1056,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // In the random case, assert that the pre-constrcted shared prefix is // still here DOCTEST_REQUIRE(rA.get_last_idx() > 3); - DOCTEST_REQUIRE(rA.get_committed_seqno() > 3); + DOCTEST_REQUIRE(rA.get_light_details().committed_seqno > 3); // And finally, that thing we said was persisted earlier (but wasn't // known to be committed), is still present on all nodes diff --git a/src/consensus/aft/test/driver.h b/src/consensus/aft/test/driver.h index aa2693bef970..c685c7b2707b 100644 --- a/src/consensus/aft/test/driver.h +++ b/src/consensus/aft/test/driver.h @@ -635,7 +635,7 @@ class RaftDriver { const auto t = r.get_view(i); auto s = fmt::format("{}.{}", t, i); - if (i == r.get_committed_seqno()) + if (i == r.get_light_details().committed_seqno) { s = fmt::format("[{}]", s); } @@ -742,17 +742,18 @@ class RaftDriver void state_one(ccf::NodeId node_id) { auto raft = _nodes.at(node_id).raft; + const auto details = raft->get_light_details(); RAFT_DRIVER_PRINT( "Note right of {}: leadership {} membership {} @{}.{} (committed " "{})", node_id, - raft->is_backup() ? + details.is_backup() ? "F" : - (raft->is_candidate() ? "C" : (raft->is_primary() ? "P" : "?")), - raft->is_retired() ? "R" : "A", - raft->get_view(), + (details.is_candidate() ? "C" : (details.is_primary() ? "P" : "?")), + details.membership_state == ccf::kv::MembershipState::Retired ? "R" : "A", + details.current_view, raft->get_last_idx(), - raft->get_committed_seqno()); + details.committed_seqno); } void state_all() @@ -934,9 +935,10 @@ class RaftDriver std::vector> primaries; for (const auto& [node_id, node_driver] : _nodes) { - if (node_driver.raft->is_primary()) + const auto details = node_driver.raft->get_light_details(); + if (details.is_primary()) { - primaries.emplace_back(node_driver.raft->get_view(), node_id); + primaries.emplace_back(details.current_view, node_id); } } return primaries; @@ -1087,20 +1089,20 @@ class RaftDriver auto [target_id, nd] = *nodes.begin(); auto& target_raft = nd.raft; - const auto target_term = target_raft->get_view(); + const auto target_term = target_raft->get_light_details().current_view; const auto target_last_idx = target_raft->get_last_idx(); - const auto target_commit_idx = target_raft->get_committed_seqno(); + const auto target_commit_idx = target_raft->get_light_details().committed_seqno; for (auto it = std::next(nodes.begin()); it != nodes.end(); ++it) { const auto& node_id = it->first; auto& raft = it->second.raft; - if (raft->get_view() != target_term) + if (raft->get_light_details().current_view != target_term) { discrepancies[node_id].push_back(fmt::format( "Term {} doesn't match term {} on {}", - raft->get_view(), + raft->get_light_details().current_view, target_term, target_id)); } @@ -1148,11 +1150,11 @@ class RaftDriver } } - if (raft->get_committed_seqno() != target_commit_idx) + if (raft->get_light_details().committed_seqno != target_commit_idx) { discrepancies[node_id].push_back(fmt::format( "Commit index {} doesn't match commit index {} on {}", - raft->get_committed_seqno(), + raft->get_light_details().committed_seqno, target_commit_idx, target_id)); } @@ -1263,7 +1265,7 @@ class RaftDriver // seqno). // Similar to the QuorumLogInv invariant from the TLA spec. const auto& raft = _nodes.at(node_id).raft; - const auto committed_seqno = raft->get_committed_seqno(); + const auto committed_seqno = raft->get_light_details().committed_seqno; auto get_ledger_prefix = [this](ccf::NodeId id, ccf::SeqNo seqno) { std::vector> prefix; @@ -1344,7 +1346,7 @@ class RaftDriver ccf::NodeId node_id, const std::string& idx_s, const size_t lineno) { auto idx = static_cast(std::stoull(idx_s)); - if (_nodes.at(node_id).raft->get_committed_seqno() != idx) + if (_nodes.at(node_id).raft->get_light_details().committed_seqno != idx) { RAFT_DRIVER_PRINT( "Note over {}: Node is not at expected commit idx {}", node_id, idx); @@ -1353,7 +1355,7 @@ class RaftDriver node_id, idx, std::to_string((int)lineno), - _nodes.at(node_id).raft->get_committed_seqno())); + _nodes.at(node_id).raft->get_light_details().committed_seqno)); } } diff --git a/src/consensus/aft/test/logging_stub.h b/src/consensus/aft/test/logging_stub.h index 509893dfc922..645cab27c15f 100644 --- a/src/consensus/aft/test/logging_stub.h +++ b/src/consensus/aft/test/logging_stub.h @@ -351,7 +351,7 @@ namespace aft set_retired_committed_hook = set_retired_committed_hook_; } - virtual void compact(Index i) {} + virtual void compact(Index i, bool is_primary = false) {} virtual void rollback(const ccf::TxID& tx_id, Term t) {} @@ -482,7 +482,7 @@ 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 = false) 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..0cae80e55175 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -29,20 +29,120 @@ DOCTEST_TEST_CASE("Single node startup" * doctest::test_suite("single")) DOCTEST_INFO("DOCTEST_REQUIRE Initial State"); - DOCTEST_REQUIRE(!r0.is_primary()); - DOCTEST_REQUIRE(!r0.primary().has_value()); - DOCTEST_REQUIRE(r0.get_view() == 0); - DOCTEST_REQUIRE(r0.get_committed_seqno() == 0); + DOCTEST_REQUIRE(!r0.get_light_details().is_primary()); + DOCTEST_REQUIRE(!r0.get_light_details().primary_id.has_value()); + DOCTEST_REQUIRE(r0.get_light_details().current_view == 0); + DOCTEST_REQUIRE(r0.get_light_details().committed_seqno == 0); DOCTEST_INFO( "In the absence of other nodes, become leader after election timeout"); r0.periodic(ms(0)); - DOCTEST_REQUIRE(!r0.is_primary()); + DOCTEST_REQUIRE(!r0.get_light_details().is_primary()); r0.periodic(election_timeout * 2); - DOCTEST_REQUIRE(r0.is_primary()); - DOCTEST_REQUIRE(r0.primary() == node_id); + DOCTEST_REQUIRE(r0.get_light_details().is_primary()); + DOCTEST_REQUIRE(r0.get_light_details().primary_id == node_id); +} + +DOCTEST_TEST_CASE( + "Consensus details and KV queries" * 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)); + + 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; + std::thread driver([&]() { + 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); + }); + + constexpr size_t reader_thread_count = 8; + std::vector readers; + readers.reserve(reader_thread_count); + for (size_t i = 0; i < reader_thread_count; ++i) + { + readers.emplace_back([&]() { + 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")) @@ -66,7 +166,7 @@ DOCTEST_TEST_CASE("Single node commit" * doctest::test_suite("single")) r0.start_ticking(); r0.periodic(election_timeout * 2); - DOCTEST_REQUIRE(r0.is_primary()); + DOCTEST_REQUIRE(r0.get_light_details().is_primary()); DOCTEST_INFO("Observe that data is committed on replicate immediately"); @@ -79,7 +179,7 @@ DOCTEST_TEST_CASE("Single node commit" * doctest::test_suite("single")) r0.replicate(ccf::kv::BatchVector{{i, entry, true, hooks}}, 1); DOCTEST_REQUIRE(r0.get_last_idx() == i); - DOCTEST_REQUIRE(r0.get_committed_seqno() == i); + DOCTEST_REQUIRE(r0.get_light_details().committed_seqno == i); } } @@ -279,7 +379,7 @@ DOCTEST_TEST_CASE( DOCTEST_INFO( "Node 0 is now leader, and sends empty append entries to other nodes"); - DOCTEST_REQUIRE(r0.is_primary()); + DOCTEST_REQUIRE(r0.get_light_details().is_primary()); DOCTEST_REQUIRE( r0c->count_messages_with_type(aft::RaftMsgType::raft_append_entries) == 3); @@ -315,7 +415,7 @@ DOCTEST_TEST_CASE( receive_message(r0, r3, *rvr_raw); - auto r3_primary = r3.primary(); + auto r3_primary = r3.get_light_details().primary_id; DOCTEST_REQUIRE(r3_primary.has_value()); DOCTEST_REQUIRE(r3_primary.value() == r0.id()); @@ -647,7 +747,7 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id1, r1c->messages)); - DOCTEST_REQUIRE(r0.is_primary()); + DOCTEST_REQUIRE(r0.get_light_details().is_primary()); DOCTEST_REQUIRE(r0c->messages.size() == 1); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(r0c->messages.size() == 0); @@ -1071,7 +1171,7 @@ DOCTEST_TEST_CASE( DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id1, r1c->messages)); - DOCTEST_REQUIRE(r0.is_primary()); + DOCTEST_REQUIRE(r0.get_light_details().is_primary()); DOCTEST_REQUIRE(r0c->messages.size() == 1); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(r0c->messages.size() == 0); diff --git a/src/endpoints/base_endpoint_registry.cpp b/src/endpoints/base_endpoint_registry.cpp index ac704a1438ef..b1ddb9e31bbe 100644 --- a/src/endpoints/base_endpoint_registry.cpp +++ b/src/endpoints/base_endpoint_registry.cpp @@ -32,7 +32,8 @@ namespace ccf reason = ccf::InvalidArgsReason::ViewSmallerThanOne; return ApiResult::InvalidArgs; } - auto latest_view = current_consensus->get_view(); + const auto latest_view = + current_consensus->get_light_details().current_view; if (since > latest_view) { // asking for something in the future @@ -96,9 +97,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) diff --git a/src/kv/kv_types.h b/src/kv/kv_types.h index 7a85981ad16d..fe1bd67ecace 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -153,16 +153,8 @@ namespace ccf::kv DECLARE_JSON_TYPE(Configuration); DECLARE_JSON_REQUIRED_FIELDS(Configuration, idx, nodes, rid); - struct ConsensusDetails + struct ConsensusLightDetails { - struct Ack - { - ccf::SeqNo seqno; - size_t last_received_ms; - }; - - std::vector configs; - std::unordered_map acks; MembershipState membership_state{}; std::optional leadership_state = std::nullopt; std::optional retirement_phase = std::nullopt; @@ -171,27 +163,60 @@ 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]] 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; }; 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); class ConfigurableConsensus { @@ -199,10 +224,8 @@ namespace ccf::kv 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; }; 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 ConsensusLightDetails get_light_details() = 0; + virtual ConsensusDetails get_details() = 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 void recv_message( const NodeId& from, const uint8_t* data, size_t size) = 0; @@ -474,10 +493,15 @@ namespace ccf::kv 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(); + const auto details = get_light_details(); + const auto committed_txid = details.committed_txid(); return ccf::evaluate_tx_status( - target_view, target_seqno, local_view, committed_view, committed_seqno); + target_view, + target_seqno, + local_view, + committed_txid.view, + committed_txid.seqno); } }; @@ -726,7 +750,7 @@ namespace ccf::kv const std::vector& data, bool public_only = false, const std::optional& expected_txid = std::nullopt) = 0; - virtual void compact(Version v) = 0; + virtual void compact(Version v, bool is_primary = false) = 0; virtual void rollback(const ccf::TxID& tx_id, Term write_term_) = 0; virtual void initialise_term(Term t) = 0; virtual CommitResult commit( diff --git a/src/kv/store.h b/src/kv/store.h index dce78e55f41d..f2c4737cc9dc 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -566,7 +566,7 @@ namespace ccf::kv return ApplyResult::PASS; } - void compact(Version v) override + void compact(Version v, bool is_primary = false) override { // This is called when the store will never be rolled back to any // state before the specified version. @@ -574,9 +574,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); } if (chunker) @@ -961,10 +959,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/stub_consensus.h b/src/kv/test/stub_consensus.h index 5d40a1558183..451e6124f8f9 100644 --- a/src/kv/test/stub_consensus.h +++ b/src/kv/test/stub_consensus.h @@ -45,16 +45,6 @@ namespace ccf::kv::test return local_id; } - virtual bool is_primary() override - { - return state == Primary; - } - - virtual bool is_candidate() override - { - return state == Candidate; - } - virtual bool can_replicate() override { return state == Primary; @@ -77,11 +67,6 @@ namespace ccf::kv::test } } - virtual bool is_backup() override - { - return state == Backup; - } - virtual void force_become_primary() override { state = Primary; @@ -162,31 +147,16 @@ 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 - { - return committed_txid.seqno; - } - - std::optional primary() override - { - return PrimaryNodeId; - } - ccf::View get_view(ccf::SeqNo seqno) override { return view_history.view_at(seqno); } - ccf::View get_view() override - { - return current_view; - } - std::vector get_view_history(ccf::SeqNo seqno) override { return view_history.get_history_until(seqno); @@ -210,14 +180,31 @@ namespace ccf::kv::test return {}; } - Configuration::Nodes get_latest_configuration() override + virtual Configuration::Nodes get_latest_configuration() { return {}; } + ConsensusLightDetails get_light_details() override + { + ConsensusLightDetails details; + details.membership_state = MembershipState::Active; + details.leadership_state = state == Primary ? LeadershipState::Leader : + state == Candidate ? LeadershipState::Candidate : + LeadershipState::Follower; + details.primary_id = state == Primary ? std::optional{PrimaryNodeId} : + std::nullopt; + 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(); + return details; } void set_last_signature_at(ccf::SeqNo seqno) @@ -231,11 +218,6 @@ namespace ccf::kv::test public: BackupStubConsensus() : StubConsensus() {} - bool is_primary() override - { - return false; - } - bool replicate(const BatchVector& entries, ccf::View view) override { return false; @@ -257,11 +239,6 @@ namespace ccf::kv::test public: PrimaryStubConsensus() : StubConsensus() {} - bool is_primary() override - { - return true; - } - bool can_replicate() override { return true; diff --git a/src/node/node_state.h b/src/node/node_state.h index eb4f64e7a911..dd2e85b8c30d 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()) { @@ -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 @@ -2767,7 +2768,7 @@ namespace ccf std::optional get_primary() override { - return consensus->primary(); + return consensus->get_light_details().primary_id; } [[nodiscard]] bool is_in_initialised_state() const override @@ -3674,8 +3675,9 @@ namespace ccf // If backup snapshot fetching is enabled and this node is a // backup, schedule a fetch task if ( - config.snapshots.backup_fetch.enabled && consensus != nullptr && - !consensus->is_primary()) + config.snapshots.backup_fetch.enabled && + consensus != nullptr && + !consensus->get_light_details().is_primary()) { ccf::tasks::Task task_to_schedule = nullptr; { diff --git a/src/node/rpc/frontend.h b/src/node/rpc/frontend.h index ec65f95ab601..dcf9321b1271 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())); @@ -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( diff --git a/src/node/rpc/node_frontend.h b/src/node/rpc/node_frontend.h index f0bca921fd55..1f3376bf5d50 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,7 +1427,10 @@ 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) { @@ -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"); From ab30c0bc8fe4356fd7f02a62b69c81e1aa85dad0 Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Thu, 20 Aug 2026 15:05:40 +0000 Subject: [PATCH 2/9] Restore test-only consensus convenience APIs Keep existing AFT and KV tests on test-owned convenience helpers while production consumers use coherent consensus details. Require Raft compaction to pass primary state explicitly and derive primary/backup stub behavior from constructor-selected state.\n\nRefs #8184 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/consensus/aft/raft_types.h | 4 +- src/consensus/aft/test/committable_suffix.cpp | 106 +++++++++--------- src/consensus/aft/test/driver.h | 42 ++++--- src/consensus/aft/test/logging_stub.h | 67 ++++++++++- src/consensus/aft/test/main.cpp | 34 +++--- src/consensus/aft/test/test_common.h | 2 +- src/kv/kv_types.h | 2 +- src/kv/store.h | 7 +- src/kv/test/stub_consensus.h | 73 +++++++----- 9 files changed, 210 insertions(+), 127 deletions(-) diff --git a/src/consensus/aft/raft_types.h b/src/consensus/aft/raft_types.h index 4c1c23d745f7..905d5eeac6cf 100644 --- a/src/consensus/aft/raft_types.h +++ b/src/consensus/aft/raft_types.h @@ -26,7 +26,7 @@ namespace aft { public: virtual ~Store() = default; - virtual void compact(Index v, bool is_primary = false) = 0; + virtual void compact(Index v, bool is_primary) = 0; virtual void rollback( const ccf::TxID& tx_id, Term term_of_next_version) = 0; virtual void initialise_term(Term t) = 0; @@ -45,7 +45,7 @@ namespace aft public: Adaptor(std::shared_ptr x) : x(x) {} - void compact(Index v, bool is_primary = false) override + void compact(Index v, bool is_primary) override { auto p = x.lock(); if (p) diff --git a/src/consensus/aft/test/committable_suffix.cpp b/src/consensus/aft/test/committable_suffix.cpp index 3d1aea29ca47..3571bedea2e5 100644 --- a/src/consensus/aft/test/committable_suffix.cpp +++ b/src/consensus/aft/test/committable_suffix.cpp @@ -195,8 +195,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rA.get_light_details().is_primary()); - DOCTEST_REQUIRE(rA.get_light_details().current_view == 1); + DOCTEST_REQUIRE(rA.is_primary()); + DOCTEST_REQUIRE(rA.get_view() == 1); // Dispatch initial AppendEntries DOCTEST_REQUIRE(4 == dispatch_all(nodes, node_idA, channelsA->messages)); @@ -207,8 +207,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rA.get_light_details().is_primary()); - DOCTEST_REQUIRE(rA.get_light_details().current_view == 1); + DOCTEST_REQUIRE(rA.is_primary()); + DOCTEST_REQUIRE(rA.get_view() == 1); } DOCTEST_INFO("Entry at 1.1 is received by all nodes"); @@ -216,7 +216,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") auto entry = make_ledger_entry(1, 1); rA.replicate(ccf::kv::BatchVector{{1, entry, true, hooks}}, 1); DOCTEST_REQUIRE(rA.get_last_idx() == 1); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 0); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 0); // Size limit was reached, so periodic is not needed // rA.periodic(request_timeout); @@ -236,7 +236,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); // Node A now knows this is committed - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 1); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 1); } DOCTEST_INFO( @@ -246,21 +246,21 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") auto entry = make_ledger_entry(1, 2); rA.replicate(ccf::kv::BatchVector{{2, entry, true, hooks}}, 1); DOCTEST_REQUIRE(rA.get_last_idx() == 2); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 1); + 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); DOCTEST_REQUIRE(rA.get_last_idx() == 3); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 1); + 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); DOCTEST_REQUIRE(rA.get_last_idx() == 4); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 1); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 1); // Size limit was reached, so periodic is not needed // rA.periodic(request_timeout); @@ -329,12 +329,12 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(0 == dispatch_all(nodes, node_idE, channelsE->messages)); // Node A now knows that 1.4 is committed - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 4); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 4); // Nodes B and C have this commit index, and are responsible for persisting // it - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); - DOCTEST_REQUIRE(rC.get_last_idx() >= rA.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); + DOCTEST_REQUIRE(rC.get_last_idx() >= rA.get_committed_seqno()); } DOCTEST_INFO("Node A dies"); @@ -360,8 +360,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rB.get_light_details().is_primary()); - DOCTEST_REQUIRE(rB.get_light_details().current_view == 2); + DOCTEST_REQUIRE(rB.is_primary()); + DOCTEST_REQUIRE(rB.get_view() == 2); } DOCTEST_INFO("Node B writes some entries, though they are lost"); @@ -383,8 +383,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") // The key features is that B is one of the quorum responsible for // persistence of 1.4, despites its commit index not being as high as 1.4 - DOCTEST_REQUIRE(rB.get_light_details().committed_seqno < rA.get_light_details().committed_seqno); - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rB.get_committed_seqno() < rA.get_committed_seqno()); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); } DOCTEST_INFO("Node C wins an election"); @@ -407,8 +407,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idD, channelsD->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idE, channelsE->messages)); - DOCTEST_REQUIRE(rC.get_light_details().is_primary()); - DOCTEST_REQUIRE(rC.get_light_details().current_view == 3); + DOCTEST_REQUIRE(rC.is_primary()); + DOCTEST_REQUIRE(rC.get_view() == 3); DOCTEST_REQUIRE(rB.get_last_idx() == 7); DOCTEST_REQUIRE(rC.get_last_idx() == 4); @@ -419,8 +419,8 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(rB.get_last_idx() == 7); - DOCTEST_REQUIRE(rB.get_light_details().committed_seqno < rA.get_light_details().committed_seqno); - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rB.get_committed_seqno() < rA.get_committed_seqno()); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); } DOCTEST_REQUIRE("Node C produces 3.5, 3.6, and 3.7"); @@ -479,7 +479,7 @@ DOCTEST_TEST_CASE("Retention of dead leader's commit") DOCTEST_REQUIRE(rB.get_last_idx() != tail_of_b); // B must still be holding the committed index it holds from A - DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rB.get_last_idx() >= rA.get_committed_seqno()); // B's term history must match the current primary's DOCTEST_REQUIRE(rB.get_last_idx() <= rC.get_last_idx()); @@ -541,8 +541,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idB)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idC)); - DOCTEST_REQUIRE(rA.get_light_details().is_primary()); - DOCTEST_REQUIRE(rA.get_light_details().current_view == 1); + DOCTEST_REQUIRE(rA.is_primary()); + DOCTEST_REQUIRE(rA.get_view() == 1); // Election-triggered heartbeats DOCTEST_REQUIRE(2 == dispatch_all(nodes, node_idA)); @@ -561,7 +561,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // If C is in an older term, it gets a heartbeat to join this primary's // term, but nothing more - if (rC.get_light_details().current_view < primary.get_light_details().current_view) + if (rC.get_view() < primary.get_view()) { primary.periodic(request_timeout); keep_messages_for(node_idC, channels_primary->messages); @@ -569,13 +569,13 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) channelsC->messages.clear(); } - DOCTEST_REQUIRE(rC.get_light_details().current_view >= primary.get_light_details().current_view); + DOCTEST_REQUIRE(rC.get_view() >= primary.get_view()); - if (rC.get_light_details().current_view > primary.get_light_details().current_view) + if (rC.get_view() > primary.get_view()) { // Trigger a message from the intended primary, so C will respond with its // current term - if (primary.get_light_details().is_primary()) + if (primary.is_primary()) { // If we were already primary, then request_timeout will produce an // AppendEntries @@ -595,7 +595,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) keep_messages_for(primary_id, channelsC->messages); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idC)); - DOCTEST_REQUIRE(rC.get_light_details().current_view == primary.get_light_details().current_view); + DOCTEST_REQUIRE(rC.get_view() == primary.get_view()); } else { @@ -632,14 +632,14 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) })); // That's sufficient to win this election - DOCTEST_REQUIRE(primary.get_light_details().is_primary()); + DOCTEST_REQUIRE(primary.is_primary()); const auto start_idx = primary.get_last_idx(); for (auto idx = start_idx + 1; idx <= start_idx + num_entries; ++idx) { - auto entry = make_ledger_entry(primary.get_light_details().current_view, idx); + auto entry = make_ledger_entry(primary.get_view(), idx); primary.replicate( - ccf::kv::BatchVector{{idx, entry, true, hooks}}, primary.get_light_details().current_view); + ccf::kv::BatchVector{{idx, entry, true, hooks}}, primary.get_view()); } // All related AppendEntries are lost @@ -672,7 +672,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) entry = make_ledger_entry(1, 2); rA.replicate(ccf::kv::BatchVector{{2, entry, true, hooks}}, 1); DOCTEST_REQUIRE(rA.get_last_idx() == 2); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 0); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 0); // Size limit was reached, so periodic is not needed // rA.periodic(request_timeout); @@ -687,7 +687,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(rC.get_last_idx() == 2); // And primary knows it is committed - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); // After a periodic heartbeat rA.periodic(request_timeout); @@ -696,9 +696,9 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_idC)); // All nodes know that this is committed - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); - DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == 2); - DOCTEST_REQUIRE(rC.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rB.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rC.get_committed_seqno() == 2); // Node A produces 2 additional entries that A and B have, and 2 additional // entries that are only present on A @@ -724,9 +724,9 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // Commit did not advance, though 4 is present on f+1 nodes and will be // persisted from here - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); - DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == 2); - DOCTEST_REQUIRE(rC.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rB.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rC.get_committed_seqno() == 2); persisted_idx = 4; persisted_entry = rB.ledger->ledger[persisted_idx - 1]; @@ -752,15 +752,15 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // Nodes A and B now have long, distinct, multi-term non-committed suffixes. // Node C has not advanced its log at all - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == 2); - DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == 2); - DOCTEST_REQUIRE(rC.get_light_details().committed_seqno == 2); + DOCTEST_REQUIRE(rA.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rB.get_committed_seqno() == 2); + DOCTEST_REQUIRE(rC.get_committed_seqno() == 2); DOCTEST_REQUIRE(rA.get_last_idx() > 4); DOCTEST_REQUIRE(rB.get_last_idx() > 3); DOCTEST_REQUIRE(rC.get_last_idx() == 2); - DOCTEST_REQUIRE(rA.get_light_details().current_view != rB.get_light_details().current_view); + DOCTEST_REQUIRE(rA.get_view() != rB.get_view()); DOCTEST_REQUIRE( rA.get_view_history(rA.get_last_idx()) != rB.get_view_history(rB.get_last_idx())); @@ -792,7 +792,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) { aim_for_a_primary = true; DOCTEST_INFO("Node A wins"); - while (rA.get_light_details().current_view <= rC.get_light_details().current_view) + while (rA.get_view() <= rC.get_view()) { channelsA->messages.clear(); rA.periodic(election_timeout); @@ -806,7 +806,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) { aim_for_a_primary = false; DOCTEST_INFO("Node B wins"); - while (rB.get_light_details().current_view <= rC.get_light_details().current_view) + while (rB.get_view() <= rC.get_view()) { channelsB->messages.clear(); rB.periodic(election_timeout); @@ -828,7 +828,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) dispatch_all(nodes, id_primary); dispatch_all(nodes, node_idC); - DOCTEST_REQUIRE(rPrimary.get_light_details().is_primary()); + DOCTEST_REQUIRE(rPrimary.is_primary()); { DOCTEST_INFO("Catch node C up"); @@ -867,7 +867,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) } DOCTEST_INFO("Bring other node in-sync"); - const auto id_other = rA.get_light_details().is_primary() ? node_idB : node_idA; + const auto id_other = rA.is_primary() ? node_idB : node_idA; { channelsA->messages.clear(); @@ -987,7 +987,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) { // One more entry in this term, and replication of it, and post-ACK // dispatch, to reach unanimous commit point - const auto view = rPrimary.get_light_details().current_view; + const auto view = rPrimary.get_view(); const auto seqno = rPrimary.get_last_idx() + 1; auto final_entry = make_ledger_entry(view, seqno); rPrimary.replicate( @@ -1012,8 +1012,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(ae.leader_commit_idx == seqno); }); - DOCTEST_REQUIRE(rPrimary.get_light_details().committed_seqno == seqno); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == rB.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rPrimary.get_committed_seqno() == seqno); + DOCTEST_REQUIRE(rA.get_committed_seqno() == rB.get_committed_seqno()); } if constexpr (is_worst_case) @@ -1036,8 +1036,8 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) DOCTEST_REQUIRE(rA.get_last_idx() == rB.get_last_idx()); DOCTEST_REQUIRE(rB.get_last_idx() == rC.get_last_idx()); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno == rB.get_light_details().committed_seqno); - DOCTEST_REQUIRE(rB.get_light_details().committed_seqno == rC.get_light_details().committed_seqno); + DOCTEST_REQUIRE(rA.get_committed_seqno() == rB.get_committed_seqno()); + DOCTEST_REQUIRE(rB.get_committed_seqno() == rC.get_committed_seqno()); const auto term_history_on_A = rA.get_view_history(rA.get_last_idx()); const auto term_history_on_B = rB.get_view_history(rB.get_last_idx()); @@ -1056,7 +1056,7 @@ DOCTEST_TEST_CASE_TEMPLATE("Multi-term divergence", T, WorstCase, RandomCase) // In the random case, assert that the pre-constrcted shared prefix is // still here DOCTEST_REQUIRE(rA.get_last_idx() > 3); - DOCTEST_REQUIRE(rA.get_light_details().committed_seqno > 3); + DOCTEST_REQUIRE(rA.get_committed_seqno() > 3); // And finally, that thing we said was persisted earlier (but wasn't // known to be committed), is still present on all nodes diff --git a/src/consensus/aft/test/driver.h b/src/consensus/aft/test/driver.h index c685c7b2707b..ea67c3034e03 100644 --- a/src/consensus/aft/test/driver.h +++ b/src/consensus/aft/test/driver.h @@ -74,10 +74,10 @@ struct LoggingStubStore_Mermaid : public aft::LoggingStubStoreConfig { using LoggingStubStoreConfig::LoggingStubStoreConfig; - void compact(aft::Index idx) override + void compact(aft::Index idx, bool is_primary) override { RAFT_DRIVER_PRINT("{}->>{}: [KV] compacting to {}", _id, _id, idx); - aft::LoggingStubStoreConfig::compact(idx); + aft::LoggingStubStoreConfig::compact(idx, is_primary); } void rollback(const ccf::TxID& tx_id, aft::Term t) override @@ -100,7 +100,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; @@ -635,7 +635,7 @@ class RaftDriver { const auto t = r.get_view(i); auto s = fmt::format("{}.{}", t, i); - if (i == r.get_light_details().committed_seqno) + if (i == r.get_committed_seqno()) { s = fmt::format("[{}]", s); } @@ -742,18 +742,17 @@ class RaftDriver void state_one(ccf::NodeId node_id) { auto raft = _nodes.at(node_id).raft; - const auto details = raft->get_light_details(); RAFT_DRIVER_PRINT( "Note right of {}: leadership {} membership {} @{}.{} (committed " "{})", node_id, - details.is_backup() ? + raft->is_backup() ? "F" : - (details.is_candidate() ? "C" : (details.is_primary() ? "P" : "?")), - details.membership_state == ccf::kv::MembershipState::Retired ? "R" : "A", - details.current_view, + (raft->is_candidate() ? "C" : (raft->is_primary() ? "P" : "?")), + raft->is_retired() ? "R" : "A", + raft->get_view(), raft->get_last_idx(), - details.committed_seqno); + raft->get_committed_seqno()); } void state_all() @@ -935,10 +934,9 @@ class RaftDriver std::vector> primaries; for (const auto& [node_id, node_driver] : _nodes) { - const auto details = node_driver.raft->get_light_details(); - if (details.is_primary()) + if (node_driver.raft->is_primary()) { - primaries.emplace_back(details.current_view, node_id); + primaries.emplace_back(node_driver.raft->get_view(), node_id); } } return primaries; @@ -1089,20 +1087,20 @@ class RaftDriver auto [target_id, nd] = *nodes.begin(); auto& target_raft = nd.raft; - const auto target_term = target_raft->get_light_details().current_view; + const auto target_term = target_raft->get_view(); const auto target_last_idx = target_raft->get_last_idx(); - const auto target_commit_idx = target_raft->get_light_details().committed_seqno; + const auto target_commit_idx = target_raft->get_committed_seqno(); for (auto it = std::next(nodes.begin()); it != nodes.end(); ++it) { const auto& node_id = it->first; auto& raft = it->second.raft; - if (raft->get_light_details().current_view != target_term) + if (raft->get_view() != target_term) { discrepancies[node_id].push_back(fmt::format( "Term {} doesn't match term {} on {}", - raft->get_light_details().current_view, + raft->get_view(), target_term, target_id)); } @@ -1150,11 +1148,11 @@ class RaftDriver } } - if (raft->get_light_details().committed_seqno != target_commit_idx) + if (raft->get_committed_seqno() != target_commit_idx) { discrepancies[node_id].push_back(fmt::format( "Commit index {} doesn't match commit index {} on {}", - raft->get_light_details().committed_seqno, + raft->get_committed_seqno(), target_commit_idx, target_id)); } @@ -1265,7 +1263,7 @@ class RaftDriver // seqno). // Similar to the QuorumLogInv invariant from the TLA spec. const auto& raft = _nodes.at(node_id).raft; - const auto committed_seqno = raft->get_light_details().committed_seqno; + const auto committed_seqno = raft->get_committed_seqno(); auto get_ledger_prefix = [this](ccf::NodeId id, ccf::SeqNo seqno) { std::vector> prefix; @@ -1346,7 +1344,7 @@ class RaftDriver ccf::NodeId node_id, const std::string& idx_s, const size_t lineno) { auto idx = static_cast(std::stoull(idx_s)); - if (_nodes.at(node_id).raft->get_light_details().committed_seqno != idx) + if (_nodes.at(node_id).raft->get_committed_seqno() != idx) { RAFT_DRIVER_PRINT( "Note over {}: Node is not at expected commit idx {}", node_id, idx); @@ -1355,7 +1353,7 @@ class RaftDriver node_id, idx, std::to_string((int)lineno), - _nodes.at(node_id).raft->get_light_details().committed_seqno)); + _nodes.at(node_id).raft->get_committed_seqno())); } } diff --git a/src/consensus/aft/test/logging_stub.h b/src/consensus/aft/test/logging_stub.h index 645cab27c15f..9669a189be87 100644 --- a/src/consensus/aft/test/logging_stub.h +++ b/src/consensus/aft/test/logging_stub.h @@ -141,6 +141,69 @@ namespace aft void commit(Index idx) {} }; + template + class TestAft : public Aft + { + public: + using Aft::Aft; + using Aft::get_view; + + std::optional primary() + { + return this->get_light_details().primary_id; + } + + 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: @@ -351,7 +414,7 @@ namespace aft set_retired_committed_hook = set_retired_committed_hook_; } - virtual void compact(Index i, bool is_primary = false) {} + virtual void compact(Index i, bool is_primary) {} virtual void rollback(const ccf::TxID& tx_id, Term t) {} @@ -482,7 +545,7 @@ 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, bool is_primary = false) override + virtual void compact(Index i, bool is_primary) 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 0cae80e55175..7f509f656233 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -5,6 +5,7 @@ #define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES #define DOCTEST_CONFIG_IMPLEMENT +#include #include using ms = std::chrono::milliseconds; @@ -29,24 +30,24 @@ DOCTEST_TEST_CASE("Single node startup" * doctest::test_suite("single")) DOCTEST_INFO("DOCTEST_REQUIRE Initial State"); - DOCTEST_REQUIRE(!r0.get_light_details().is_primary()); - DOCTEST_REQUIRE(!r0.get_light_details().primary_id.has_value()); - DOCTEST_REQUIRE(r0.get_light_details().current_view == 0); - DOCTEST_REQUIRE(r0.get_light_details().committed_seqno == 0); + DOCTEST_REQUIRE(!r0.is_primary()); + DOCTEST_REQUIRE(!r0.primary().has_value()); + DOCTEST_REQUIRE(r0.get_view() == 0); + DOCTEST_REQUIRE(r0.get_committed_seqno() == 0); DOCTEST_INFO( "In the absence of other nodes, become leader after election timeout"); r0.periodic(ms(0)); - DOCTEST_REQUIRE(!r0.get_light_details().is_primary()); + DOCTEST_REQUIRE(!r0.is_primary()); r0.periodic(election_timeout * 2); - DOCTEST_REQUIRE(r0.get_light_details().is_primary()); - DOCTEST_REQUIRE(r0.get_light_details().primary_id == node_id); + DOCTEST_REQUIRE(r0.is_primary()); + DOCTEST_REQUIRE(r0.primary() == node_id); } DOCTEST_TEST_CASE( - "Consensus details and KV queries" * doctest::test_suite("single")) + "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; @@ -108,7 +109,10 @@ DOCTEST_TEST_CASE( 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) { @@ -118,12 +122,12 @@ DOCTEST_TEST_CASE( stop.store(true, std::memory_order_release); }); - constexpr size_t reader_thread_count = 8; 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(); @@ -166,7 +170,7 @@ DOCTEST_TEST_CASE("Single node commit" * doctest::test_suite("single")) r0.start_ticking(); r0.periodic(election_timeout * 2); - DOCTEST_REQUIRE(r0.get_light_details().is_primary()); + DOCTEST_REQUIRE(r0.is_primary()); DOCTEST_INFO("Observe that data is committed on replicate immediately"); @@ -179,7 +183,7 @@ DOCTEST_TEST_CASE("Single node commit" * doctest::test_suite("single")) r0.replicate(ccf::kv::BatchVector{{i, entry, true, hooks}}, 1); DOCTEST_REQUIRE(r0.get_last_idx() == i); - DOCTEST_REQUIRE(r0.get_light_details().committed_seqno == i); + DOCTEST_REQUIRE(r0.get_committed_seqno() == i); } } @@ -379,7 +383,7 @@ DOCTEST_TEST_CASE( DOCTEST_INFO( "Node 0 is now leader, and sends empty append entries to other nodes"); - DOCTEST_REQUIRE(r0.get_light_details().is_primary()); + DOCTEST_REQUIRE(r0.is_primary()); DOCTEST_REQUIRE( r0c->count_messages_with_type(aft::RaftMsgType::raft_append_entries) == 3); @@ -415,7 +419,7 @@ DOCTEST_TEST_CASE( receive_message(r0, r3, *rvr_raw); - auto r3_primary = r3.get_light_details().primary_id; + auto r3_primary = r3.primary(); DOCTEST_REQUIRE(r3_primary.has_value()); DOCTEST_REQUIRE(r3_primary.value() == r0.id()); @@ -747,7 +751,7 @@ DOCTEST_TEST_CASE("Recv append entries logic" * doctest::test_suite("multiple")) DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id1, r1c->messages)); - DOCTEST_REQUIRE(r0.get_light_details().is_primary()); + DOCTEST_REQUIRE(r0.is_primary()); DOCTEST_REQUIRE(r0c->messages.size() == 1); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(r0c->messages.size() == 0); @@ -1171,7 +1175,7 @@ DOCTEST_TEST_CASE( DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id1, r1c->messages)); - DOCTEST_REQUIRE(r0.get_light_details().is_primary()); + DOCTEST_REQUIRE(r0.is_primary()); DOCTEST_REQUIRE(r0c->messages.size() == 1); DOCTEST_REQUIRE(1 == dispatch_all(nodes, node_id0, r0c->messages)); DOCTEST_REQUIRE(r0c->messages.size() == 0); 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/kv/kv_types.h b/src/kv/kv_types.h index fe1bd67ecace..4aa6ddf8dd0e 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -750,7 +750,7 @@ namespace ccf::kv const std::vector& data, bool public_only = false, const std::optional& expected_txid = std::nullopt) = 0; - virtual void compact(Version v, bool is_primary = false) = 0; + virtual void compact(Version v) = 0; virtual void rollback(const ccf::TxID& tx_id, Term write_term_) = 0; virtual void initialise_term(Term t) = 0; virtual CommitResult commit( diff --git a/src/kv/store.h b/src/kv/store.h index f2c4737cc9dc..f48a9aaeef00 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -566,7 +566,12 @@ namespace ccf::kv return ApplyResult::PASS; } - void compact(Version v, bool is_primary = false) override + void compact(Version v) override + { + compact(v, false); + } + + void compact(Version v, bool is_primary) { // This is called when the store will never be rolled back to any // state before the specified version. diff --git a/src/kv/test/stub_consensus.h b/src/kv/test/stub_consensus.h index 451e6124f8f9..2ba52f8dd86d 100644 --- a/src/kv/test/stub_consensus.h +++ b/src/kv/test/stub_consensus.h @@ -38,13 +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 { return local_id; } + virtual bool is_primary() + { + return state == Primary; + } + + virtual bool is_candidate() + { + return state == Candidate; + } + virtual bool can_replicate() override { return state == Primary; @@ -67,6 +81,11 @@ namespace ccf::kv::test } } + virtual bool is_backup() + { + return state == Backup; + } + virtual void force_become_primary() override { state = Primary; @@ -92,6 +111,11 @@ namespace ccf::kv::test bool replicate(const BatchVector& entries, ccf::View view) override { + if (!can_replicate()) + { + return false; + } + for (const auto& entry : entries) { replica.push_back(entry); @@ -152,11 +176,26 @@ namespace ccf::kv::test return {committed_txid.view, committed_txid.seqno}; } + virtual ccf::SeqNo get_committed_seqno() + { + return committed_txid.seqno; + } + + virtual std::optional primary() + { + return PrimaryNodeId; + } + ccf::View get_view(ccf::SeqNo seqno) override { return view_history.view_at(seqno); } + virtual ccf::View get_view() + { + return current_view; + } + std::vector get_view_history(ccf::SeqNo seqno) override { return view_history.get_history_until(seqno); @@ -192,8 +231,7 @@ namespace ccf::kv::test details.leadership_state = state == Primary ? LeadershipState::Leader : state == Candidate ? LeadershipState::Candidate : LeadershipState::Follower; - details.primary_id = state == Primary ? std::optional{PrimaryNodeId} : - std::nullopt; + details.primary_id = PrimaryNodeId; details.current_view = current_view; details.committed_view = committed_txid.view; details.committed_seqno = committed_txid.seqno; @@ -216,37 +254,12 @@ namespace ccf::kv::test class BackupStubConsensus : public StubConsensus { public: - BackupStubConsensus() : StubConsensus() {} - - 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 can_replicate() override - { - return true; - } - - Consensus::SignatureDisposition get_signature_disposition() override - { - return Consensus::SignatureDisposition::CAN_SIGN; - } + PrimaryStubConsensus() : StubConsensus(Primary) {} }; } From 9f10265b2511af49e7b476f5c152a8e8b91aa1ad Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Thu, 20 Aug 2026 15:30:43 +0000 Subject: [PATCH 3/9] Tidy and format --- src/node/node_state.h | 3 +-- src/node/rpc/node_frontend.h | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/src/node/node_state.h b/src/node/node_state.h index dd2e85b8c30d..f03e8ef8e16a 100644 --- a/src/node/node_state.h +++ b/src/node/node_state.h @@ -3675,8 +3675,7 @@ namespace ccf // If backup snapshot fetching is enabled and this node is a // backup, schedule a fetch task if ( - config.snapshots.backup_fetch.enabled && - consensus != nullptr && + config.snapshots.backup_fetch.enabled && consensus != nullptr && !consensus->get_light_details().is_primary()) { ccf::tasks::Task task_to_schedule = nullptr; diff --git a/src/node/rpc/node_frontend.h b/src/node/rpc/node_frontend.h index 1f3376bf5d50..00a910d62d10 100644 --- a/src/node/rpc/node_frontend.h +++ b/src/node/rpc/node_frontend.h @@ -1432,7 +1432,7 @@ namespace ccf 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(), From de984a868db7a78095be8af371791a452f79b350 Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Thu, 20 Aug 2026 15:47:37 +0000 Subject: [PATCH 4/9] Remove more unnecessary cruft --- src/consensus/aft/raft.h | 25 +++++++------------------ src/consensus/aft/test/logging_stub.h | 5 +++++ 2 files changed, 12 insertions(+), 18 deletions(-) diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index de48133d579d..d6cc87e0275d 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -21,7 +21,6 @@ #include "service/tables/signatures.h" #include -#include #include #include #include @@ -202,11 +201,6 @@ namespace aft // pre-deserialisation, without an additional header. static constexpr size_t max_terms_per_append_entries = 1; - void set_leadership_state(ccf::kv::LeadershipState new_state) - { - std::atomic_ref(state->leadership_state).store(new_state); - } - public: static constexpr size_t append_entries_size_limit = 20000; std::unique_ptr ledger; @@ -294,12 +288,6 @@ namespace aft return Consensus::SignatureDisposition::CANT_REPLICATE; } - bool is_primary() const - { - return std::atomic_ref(state->leadership_state).load() == - ccf::kv::LeadershipState::Leader; - } - private: bool is_retired() const { @@ -2076,7 +2064,7 @@ namespace aft return; } - set_leadership_state(ccf::kv::LeadershipState::PreVoteCandidate); + state->leadership_state = ccf::kv::LeadershipState::PreVoteCandidate; leader_id.reset(); reset_votes_for_me(); @@ -2121,7 +2109,7 @@ namespace aft return; } - set_leadership_state(ccf::kv::LeadershipState::Candidate); + state->leadership_state = ccf::kv::LeadershipState::Candidate; leader_id.reset(); voted_for = state->node_id; @@ -2178,7 +2166,7 @@ namespace aft store->initialise_term(state->current_view); } - set_leadership_state(ccf::kv::LeadershipState::Leader); + state->leadership_state = ccf::kv::LeadershipState::Leader; leader_id = state->node_id; should_sign = true; @@ -2232,7 +2220,7 @@ namespace aft restart_election_timeout(); reset_last_ack_timeouts(); - set_leadership_state(ccf::kv::LeadershipState::Follower); + state->leadership_state = ccf::kv::LeadershipState::Follower; RAFT_INFO_FMT( "Becoming follower {}: {}.{}", state->node_id, @@ -2355,7 +2343,7 @@ namespace aft nominate_successor(); leader_id.reset(); - set_leadership_state(ccf::kv::LeadershipState::None); + state->leadership_state = ccf::kv::LeadershipState::None; } state->membership_state = ccf::kv::MembershipState::Retired; @@ -2581,7 +2569,8 @@ namespace aft } RAFT_DEBUG_FMT("Compacting..."); - store->compact(idx, is_primary()); + store->compact( + idx, state->leadership_state == ccf::kv::LeadershipState::Leader); ledger->commit(idx); if (commit_callbacks != nullptr) diff --git a/src/consensus/aft/test/logging_stub.h b/src/consensus/aft/test/logging_stub.h index 9669a189be87..22d3fc187a3c 100644 --- a/src/consensus/aft/test/logging_stub.h +++ b/src/consensus/aft/test/logging_stub.h @@ -153,6 +153,11 @@ namespace aft 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(); From 6e2a7a6851fe48ea026f7f7718964b54c72a4487 Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Thu, 20 Aug 2026 15:56:37 +0000 Subject: [PATCH 5/9] Document shim --- src/consensus/aft/test/logging_stub.h | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/consensus/aft/test/logging_stub.h b/src/consensus/aft/test/logging_stub.h index 22d3fc187a3c..51c86db928a3 100644 --- a/src/consensus/aft/test/logging_stub.h +++ b/src/consensus/aft/test/logging_stub.h @@ -141,6 +141,8 @@ 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 { From fe6ed343351492c30350700cde8283966cc10faf Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Fri, 21 Aug 2026 10:04:14 +0000 Subject: [PATCH 6/9] Rationalise consensus detail queries --- doc/schemas/node_openapi.json | 21 +++++++ src/consensus/aft/raft.h | 53 ++++++++++-------- src/consensus/aft/test/driver.h | 2 +- src/consensus/aft/test/logging_stub.h | 30 ++-------- src/consensus/aft/test/main.cpp | 15 +++++ src/endpoints/base_endpoint_registry.cpp | 10 ++-- src/kv/kv_types.h | 70 ++++++++++++++++-------- src/kv/test/stub_consensus.h | 34 +++++------- src/node/historical_queries.h | 3 +- src/node/hooks.h | 28 ++-------- src/node/jwt_key_auto_refresh.h | 4 +- src/node/node_state.h | 4 +- src/node/recovery_decision_protocol.cpp | 7 ++- src/node/rpc/frontend.h | 12 ++-- src/node/test/history.cpp | 10 ---- 15 files changed, 160 insertions(+), 143 deletions(-) diff --git a/doc/schemas/node_openapi.json b/doc/schemas/node_openapi.json index 9afa960682fc..3a23391070b0 100644 --- a/doc/schemas/node_openapi.json +++ b/doc/schemas/node_openapi.json @@ -163,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", @@ -190,6 +194,17 @@ ], "type": "object" }, + "ConsensusViewHistory": { + "properties": { + "starts": { + "$ref": "#/components/schemas/uint64_array" + } + }, + "required": [ + "starts" + ], + "type": "object" + }, "ConsensusNodeConfig": { "properties": { "address": { @@ -905,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 d6cc87e0275d..83ad72b249eb 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -247,17 +247,11 @@ namespace aft ~Aft() override = default; - ccf::NodeId id() override + ccf::NodeId id() { return state->node_id; } - 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 @@ -306,7 +300,7 @@ namespace aft { 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, @@ -435,22 +429,22 @@ namespace aft return state->last_idx; } - 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. + // Called by snapshot creation from Aft::commit(), with state->lock held. return state->view_history.get_history_until(idx); } - std::vector get_view_history_since(Index idx) override + ccf::TxStatus evaluate_tx_status( + ccf::View target_view, ccf::SeqNo target_seqno) override { - // This should only be called when the spin lock is held. - return state->view_history.get_history_since(idx); + std::lock_guard guard(state->lock); + 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); + + return ccf::evaluate_tx_status( + target_view, target_seqno, local_view, committed_view, committed_seqno); } // Same as ccfraft.tla GetServerSet/IsInServerSet @@ -531,14 +525,28 @@ 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; + auto configuration = configurations.empty() ? Configuration::Nodes{} : + configurations.back().nodes; + for (const auto& [node_id, node_info] : changes) + { + if (node_info.has_value()) + { + configuration.insert_or_assign(node_id, node_info.value()); + } + else + { + configuration.erase(node_id); + } + } + add_configuration(idx, configuration); } private: @@ -575,6 +583,7 @@ namespace aft 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] = { diff --git a/src/consensus/aft/test/driver.h b/src/consensus/aft/test/driver.h index ea67c3034e03..fc2888bff60a 100644 --- a/src/consensus/aft/test/driver.h +++ b/src/consensus/aft/test/driver.h @@ -633,7 +633,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 51c86db928a3..9c2360443589 100644 --- a/src/consensus/aft/test/logging_stub.h +++ b/src/consensus/aft/test/logging_stub.h @@ -148,7 +148,11 @@ namespace aft { public: using Aft::Aft; - using Aft::get_view; + + ccf::View get_view(ccf::SeqNo seqno) + { + return this->get_details().view_history.view_at(seqno); + } std::optional primary() { @@ -375,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); } }; diff --git a/src/consensus/aft/test/main.cpp b/src/consensus/aft/test/main.cpp index 7f509f656233..825927586d21 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -10,6 +10,21 @@ 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.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; diff --git a/src/endpoints/base_endpoint_registry.cpp b/src/endpoints/base_endpoint_registry.cpp index b1ddb9e31bbe..84f3b6b3037a 100644 --- a/src/endpoints/base_endpoint_registry.cpp +++ b/src/endpoints/base_endpoint_registry.cpp @@ -32,15 +32,14 @@ namespace ccf reason = ccf::InvalidArgsReason::ViewSmallerThanOne; return ApiResult::InvalidArgs; } - const auto latest_view = - current_consensus->get_light_details().current_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; @@ -208,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 4aa6ddf8dd0e..bc363fadf178 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,6 +154,36 @@ namespace ccf::kv DECLARE_JSON_TYPE(Configuration); DECLARE_JSON_REQUIRED_FIELDS(Configuration, idx, nodes, rid); + struct ConsensusViewHistory + { + // Entry i stores the first sequence number in view i + 1. + std::vector starts; + + [[nodiscard]] ccf::View view_at(ccf::SeqNo seqno) const + { + const auto it = std::upper_bound(starts.begin(), starts.end(), seqno); + if (it == starts.begin()) + { + return ccf::VIEW_UNKNOWN; + } + + return it - starts.begin(); + } + + [[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{}; @@ -182,6 +213,13 @@ namespace ccf::kv 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}; @@ -198,6 +236,7 @@ namespace ccf::kv std::vector configs; std::unordered_map acks; + ConsensusViewHistory view_history; }; DECLARE_JSON_TYPE(ConsensusDetails::Ack); @@ -216,16 +255,16 @@ namespace ccf::kv committed_seqno); DECLARE_JSON_TYPE_WITH_BASE(ConsensusDetails, ConsensusLightDetails); - DECLARE_JSON_REQUIRED_FIELDS(ConsensusDetails, configs, acks); + 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; - [[nodiscard]] virtual Configuration::Nodes get_latest_configuration_unsafe() - const = 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 get_view_history( ccf::SeqNo seqno = std::numeric_limits::max()) = 0; - virtual std::vector get_view_history_since( - ccf::SeqNo seqno) = 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; @@ -488,21 +525,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 details = get_light_details(); - const auto committed_txid = details.committed_txid(); - - return ccf::evaluate_tx_status( - target_view, - target_seqno, - local_view, - committed_txid.view, - committed_txid.seqno); - } }; struct PendingTxInfo diff --git a/src/kv/test/stub_consensus.h b/src/kv/test/stub_consensus.h index 2ba52f8dd86d..889d85801fdb 100644 --- a/src/kv/test/stub_consensus.h +++ b/src/kv/test/stub_consensus.h @@ -44,7 +44,7 @@ namespace ccf::kv::test local_id(PrimaryNodeId) {} - virtual NodeId id() override + virtual NodeId id() { return local_id; } @@ -59,11 +59,6 @@ namespace ccf::kv::test return state == Candidate; } - virtual bool can_replicate() override - { - return state == Primary; - } - virtual bool is_at_max_capacity() override { return false; @@ -111,7 +106,7 @@ namespace ccf::kv::test bool replicate(const BatchVector& entries, ccf::View view) override { - if (!can_replicate()) + if (state != Primary) { return false; } @@ -186,11 +181,6 @@ namespace ccf::kv::test return PrimaryNodeId; } - ccf::View get_view(ccf::SeqNo seqno) override - { - return view_history.view_at(seqno); - } - virtual ccf::View get_view() { return current_view; @@ -201,9 +191,15 @@ namespace ccf::kv::test return view_history.get_history_until(seqno); } - std::vector get_view_history_since(ccf::SeqNo seqno) override + ccf::TxStatus evaluate_tx_status( + ccf::View target_view, ccf::SeqNo target_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( @@ -211,13 +207,12 @@ 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 - { - return {}; - } + void update_configuration( + ccf::SeqNo seqno, const Configuration::NodeChanges& changes) override + {} virtual Configuration::Nodes get_latest_configuration() { @@ -242,6 +237,7 @@ namespace ccf::kv::test { ConsensusDetails details; static_cast(details) = get_light_details(); + details.view_history.starts = view_history.get_history_until(); return details; } 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 f03e8ef8e16a..9e77d68af974 100644 --- a/src/node/node_state.h +++ b/src/node/node_state.h @@ -2288,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", @@ -2763,7 +2763,7 @@ 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 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 dcf9321b1271..52873fa70f55 100644 --- a/src/node/rpc/frontend.h +++ b/src/node/rpc/frontend.h @@ -310,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) { @@ -345,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) { @@ -796,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) @@ -962,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/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; From 46c164b761929d014ca076f7496c01cdf275ab7b Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Fri, 21 Aug 2026 10:19:09 +0000 Subject: [PATCH 7/9] Pass view history into snapshot creation --- src/consensus/aft/raft.h | 7 ++-- src/consensus/aft/raft_types.h | 12 ++++-- src/consensus/aft/test/driver.h | 7 +++- src/consensus/aft/test/logging_stub.h | 11 +++++- src/consensus/aft/test/main.cpp | 5 +++ src/kv/kv_types.h | 16 +++++--- src/kv/store.h | 17 +++++---- src/kv/test/kv_bench.cpp | 4 +- src/kv/test/kv_dynamic_tables.cpp | 2 +- src/kv/test/kv_snapshot.cpp | 10 ++--- src/kv/test/stub_consensus.h | 5 --- src/node/snapshotter.h | 16 +++++--- src/node/test/snapshot.cpp | 9 ++++- src/node/test/snapshotter.cpp | 54 +++++++++++++++++++-------- 14 files changed, 117 insertions(+), 58 deletions(-) diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 83ad72b249eb..7aca0aabfff8 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -429,9 +429,8 @@ namespace aft return state->last_idx; } - std::vector get_view_history(Index idx) override + std::vector get_view_history(Index idx) { - // Called by snapshot creation from Aft::commit(), with state->lock held. return state->view_history.get_history_until(idx); } @@ -2579,7 +2578,9 @@ namespace aft RAFT_DEBUG_FMT("Compacting..."); store->compact( - idx, state->leadership_state == ccf::kv::LeadershipState::Leader); + idx, + state->leadership_state == ccf::kv::LeadershipState::Leader, + {state->view_history.get_history_until(idx)}); ledger->commit(idx); if (commit_callbacks != nullptr) diff --git a/src/consensus/aft/raft_types.h b/src/consensus/aft/raft_types.h index 905d5eeac6cf..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, bool is_primary) = 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, bool is_primary) override + void compact( + Index v, + bool is_primary, + const ccf::kv::ConsensusViewHistory& view_history) override { auto p = x.lock(); if (p) { - p->compact(v, is_primary); + p->compact(v, is_primary, view_history); } } diff --git a/src/consensus/aft/test/driver.h b/src/consensus/aft/test/driver.h index fc2888bff60a..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, bool is_primary) 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, is_primary); + aft::LoggingStubStoreConfig::compact(idx, is_primary, view_history); } void rollback(const ccf::TxID& tx_id, aft::Term t) override diff --git a/src/consensus/aft/test/logging_stub.h b/src/consensus/aft/test/logging_stub.h index 9c2360443589..f2bb55aac6aa 100644 --- a/src/consensus/aft/test/logging_stub.h +++ b/src/consensus/aft/test/logging_stub.h @@ -403,7 +403,11 @@ namespace aft set_retired_committed_hook = set_retired_committed_hook_; } - virtual void compact(Index i, bool is_primary) {} + virtual void compact( + Index i, + bool is_primary, + const ccf::kv::ConsensusViewHistory& view_history) + {} virtual void rollback(const ccf::TxID& tx_id, Term t) {} @@ -534,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, bool is_primary) 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 825927586d21..d8d543b870b1 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -20,6 +20,11 @@ DOCTEST_TEST_CASE("Consensus view history snapshot") 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()); diff --git a/src/kv/kv_types.h b/src/kv/kv_types.h index bc363fadf178..624fdbe4d3df 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -170,6 +170,12 @@ namespace ccf::kv 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()) @@ -506,9 +512,6 @@ namespace ccf::kv virtual ConsensusLightDetails get_light_details() = 0; virtual ConsensusDetails get_details() = 0; - virtual std::vector get_view_history( - ccf::SeqNo seqno = std::numeric_limits::max()) = 0; - virtual ccf::TxStatus evaluate_tx_status( ccf::View target_view, ccf::SeqNo target_seqno) = 0; @@ -634,7 +637,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; @@ -782,7 +788,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 f48a9aaeef00..330f4276efa2 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)); } } @@ -568,10 +568,13 @@ namespace ccf::kv void compact(Version v) override { - compact(v, false); + compact(v, false, {}); } - void compact(Version v, bool is_primary) + 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. @@ -579,7 +582,7 @@ namespace ccf::kv if (snapshotter) { - snapshotter->commit(v, is_primary); + snapshotter->commit(v, is_primary, view_history); } if (chunker) 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 889d85801fdb..a09a7340fcab 100644 --- a/src/kv/test/stub_consensus.h +++ b/src/kv/test/stub_consensus.h @@ -186,11 +186,6 @@ namespace ccf::kv::test return current_view; } - std::vector get_view_history(ccf::SeqNo seqno) override - { - return view_history.get_history_until(seqno); - } - ccf::TxStatus evaluate_tx_status( ccf::View target_view, ccf::SeqNo target_seqno) override { 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/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..73a5b96e3db7 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,18 @@ 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 +587,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 +600,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 +616,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 +634,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 +693,10 @@ 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 +786,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 +811,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. From 258339e3680a2a7ac5ea3b63f75faf2686a8ab66 Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Fri, 21 Aug 2026 10:41:20 +0000 Subject: [PATCH 8/9] Format snapshot history plumbing --- src/consensus/aft/test/main.cpp | 3 ++- src/kv/kv_types.h | 3 ++- src/kv/store.h | 4 +--- src/node/test/snapshotter.cpp | 12 +++--------- 4 files changed, 8 insertions(+), 14 deletions(-) diff --git a/src/consensus/aft/test/main.cpp b/src/consensus/aft/test/main.cpp index d8d543b870b1..e7bde06b15f7 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -23,7 +23,8 @@ DOCTEST_TEST_CASE("Consensus view history snapshot") 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.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}); diff --git a/src/kv/kv_types.h b/src/kv/kv_types.h index 624fdbe4d3df..2260e9077e71 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -173,7 +173,8 @@ namespace ccf::kv [[nodiscard]] ConsensusViewHistory until(ccf::SeqNo seqno) const { return { - {starts.begin(), std::upper_bound(starts.begin(), starts.end(), seqno)}}; + {starts.begin(), + std::upper_bound(starts.begin(), starts.end(), seqno)}}; } [[nodiscard]] std::vector since(ccf::View view) const diff --git a/src/kv/store.h b/src/kv/store.h index 330f4276efa2..3a15786ca6a0 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -572,9 +572,7 @@ namespace ccf::kv } void compact( - Version v, - bool is_primary, - const ConsensusViewHistory& view_history) + 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. diff --git a/src/node/test/snapshotter.cpp b/src/node/test/snapshotter.cpp index 73a5b96e3db7..11dcad0b6fe6 100644 --- a/src/node/test/snapshotter.cpp +++ b/src/node/test/snapshotter.cpp @@ -567,17 +567,13 @@ TEST_CASE("Rollback before snapshot is committed") // evidence was originally committed snapshotter->commit( - snapshot_tx_interval + 1, - true, - consensus->get_details().view_history); + 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, - consensus->get_details().view_history); + snapshot_tx_interval + 2, true, consensus->get_details().view_history); REQUIRE_FALSE(latest_committed_snapshot_idx(snapshot_dir.path).has_value()); } @@ -694,9 +690,7 @@ TEST_CASE("Snapshot status updates preserve future queued snapshot") record_signature(history, snapshotter, network.tables->current_version())); snapshotter->commit( - 2 * snapshot_tx_interval, - true, - consensus->get_details().view_history); + 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 From 75dc875c56c74eba89d9e4a4bfe17c70da96794d Mon Sep 17 00:00:00 2001 From: Eddy Ashton Date: Fri, 21 Aug 2026 12:56:48 +0000 Subject: [PATCH 9/9] Update configuration handling to use try_emplace and add tests for new node integration --- src/consensus/aft/raft.h | 2 +- src/consensus/aft/test/main.cpp | 15 +++++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/consensus/aft/raft.h b/src/consensus/aft/raft.h index 7aca0aabfff8..b7c13c4a3bfe 100644 --- a/src/consensus/aft/raft.h +++ b/src/consensus/aft/raft.h @@ -538,7 +538,7 @@ namespace aft { if (node_info.has_value()) { - configuration.insert_or_assign(node_id, node_info.value()); + configuration.try_emplace(node_id, node_info.value()); } else { diff --git a/src/consensus/aft/test/main.cpp b/src/consensus/aft/test/main.cpp index e7bde06b15f7..5144b9c81ef4 100644 --- a/src/consensus/aft/test/main.cpp +++ b/src/consensus/aft/test/main.cpp @@ -96,6 +96,21 @@ DOCTEST_TEST_CASE( 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());