From ca3d8d5e8accc98bf1d03b84eef1203fb1966ca7 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Thu, 6 Aug 2026 16:08:55 +0000 Subject: [PATCH 1/2] Fix RpcFrontend pointer publication --- src/enclave/rpc_handler.h | 4 + src/node/node_state.h | 5 ++ src/node/rpc/frontend.h | 134 ++++++++++++++-------------- src/node/rpc/test/frontend_test.cpp | 106 ++++++++++++++++++++++ tsan_env_suppressions | 3 - 5 files changed, 183 insertions(+), 69 deletions(-) diff --git a/src/enclave/rpc_handler.h b/src/enclave/rpc_handler.h index 2005d3bd1b44..92cd49e60efb 100644 --- a/src/enclave/rpc_handler.h +++ b/src/enclave/rpc_handler.h @@ -14,6 +14,8 @@ namespace ccf::kv { class CommittableTx; + class Consensus; + class TxHistory; } namespace ccf @@ -33,6 +35,8 @@ namespace ccf virtual void tick(std::chrono::milliseconds /*elapsed*/) {} virtual void open() = 0; virtual bool is_open() = 0; + virtual void set_consensus_and_history( + ccf::kv::Consensus* consensus, ccf::kv::TxHistory* history) = 0; // Used by rpcendpoint to process incoming client RPCs virtual void process(std::shared_ptr ctx) = 0; diff --git a/src/node/node_state.h b/src/node/node_state.h index 070f4ff61a01..212c6d1cdb2c 100644 --- a/src/node/node_state.h +++ b/src/node/node_state.h @@ -3532,6 +3532,11 @@ namespace ccf network.tables->set_consensus(consensus); network.tables->set_snapshotter(snapshotter); + for (auto& [actor, frontend] : rpc_map->frontends()) + { + frontend->set_consensus_and_history(consensus.get(), history.get()); + } + // When a node is added, even locally, inform consensus so that it // can add a new active configuration. network.tables->set_map_hook( diff --git a/src/node/rpc/frontend.h b/src/node/rpc/frontend.h index 9532eddcb02e..6eed17354e42 100644 --- a/src/node/rpc/frontend.h +++ b/src/node/rpc/frontend.h @@ -4,6 +4,7 @@ #include "ccf/endpoint_registry.h" #include "ccf/http_status.h" +#include "ccf/node/node_configuration_interface.h" #include "ccf/node_context.h" #include "ccf/pal/locking.h" #include "ccf/rpc_exception.h" @@ -25,6 +26,7 @@ #define FMT_HEADER_ONLY +#include #include #include #include @@ -42,34 +44,17 @@ namespace ccf ccf::pal::Mutex open_lock; bool is_open_ = false; - ccf::kv::Consensus* consensus{nullptr}; + std::atomic consensus{nullptr}; std::shared_ptr cmd_forwarder; - ccf::kv::TxHistory* history{nullptr}; + std::atomic history{nullptr}; size_t sig_tx_interval = 5000; std::chrono::milliseconds sig_ms_interval = std::chrono::milliseconds(1000); std::chrono::milliseconds ms_to_sig = std::chrono::milliseconds(1000); - std::shared_ptr node_configuration_subsystem = + std::shared_ptr node_configuration_subsystem = nullptr; - void update_consensus() - { - auto* c = tables.get_consensus().get(); - - if (consensus != c) - { - consensus = c; - endpoints.set_consensus(consensus); - } - } - - void update_history() - { - history = tables.get_history().get(); - endpoints.set_history(history); - } - endpoints::EndpointDefinitionPtr find_endpoint( std::shared_ptr ctx, ccf::kv::CommittableTx& tx) { @@ -130,7 +115,7 @@ namespace ccf if (!node_configuration_subsystem) { node_configuration_subsystem = - node_context.get_subsystem(); + node_context.get_subsystem(); if (!node_configuration_subsystem) { ctx->set_response_status(HTTP_STATUS_INTERNAL_SERVER_ERROR); @@ -223,7 +208,8 @@ namespace ccf std::optional resolve_redirect_location( const RedirectionResolverConfig& resolver, ccf::kv::ReadOnlyTx& tx, - const ccf::ListenInterfaceID& incoming_interface) + const ccf::ListenInterfaceID& incoming_interface, + ccf::kv::Consensus* current_consensus) { switch (resolver.kind) { @@ -248,8 +234,14 @@ namespace ccf std::vector::const_iterator> target_node_its; const auto nodes = InternalTablesAccess::get_trusted_nodes(tx); + + if (current_consensus == nullptr) { - const auto primary_id = consensus->primary(); + return std::nullopt; + } + + { + const auto primary_id = current_consensus->primary(); if (seeking_primary && primary_id.has_value()) { target_node_its.push_back(nodes.find(primary_id.value())); @@ -303,7 +295,8 @@ namespace ccf ccf::kv::ReadOnlyTx& tx, std::shared_ptr ctx, const endpoints::EndpointDefinitionPtr& endpoint, - const ccf::NodeInfoNetwork_v2::NetInterface::Redirections& redirections) + const ccf::NodeInfoNetwork_v2::NetInterface::Redirections& redirections, + ccf::kv::Consensus* current_consensus) { auto rs = endpoint->properties.redirection_strategy; @@ -317,7 +310,7 @@ namespace ccf case (ccf::endpoints::RedirectionStrategy::ToPrimary): { const bool is_primary = - (consensus != nullptr) && consensus->can_replicate(); + current_consensus != nullptr && current_consensus->can_replicate(); if (!is_primary) { @@ -326,8 +319,8 @@ namespace ccf const auto listen_interface = ctx->get_session_context()->interface_id.value_or( PRIMARY_RPC_INTERFACE); - const auto location = - resolve_redirect_location(resolver, tx, listen_interface); + const auto location = resolve_redirect_location( + resolver, tx, listen_interface, current_consensus); if (location.has_value()) { ctx->set_response_header( @@ -352,7 +345,7 @@ namespace ccf case (ccf::endpoints::RedirectionStrategy::ToBackup): { const bool is_backup = - (consensus != nullptr) && !consensus->can_replicate(); + current_consensus != nullptr && !current_consensus->can_replicate(); if (!is_backup) { @@ -361,8 +354,8 @@ namespace ccf const auto listen_interface = ctx->get_session_context()->interface_id.value_or( PRIMARY_RPC_INTERFACE); - const auto location = - resolve_redirect_location(resolver, tx, listen_interface); + const auto location = resolve_redirect_location( + resolver, tx, listen_interface, current_consensus); if (location.has_value()) { ctx->set_response_header( @@ -398,7 +391,7 @@ namespace ccf if (!node_configuration_subsystem) { node_configuration_subsystem = - node_context.get_subsystem(); + node_context.get_subsystem(); if (!node_configuration_subsystem) { LOG_FAIL_FMT("Unable to access NodeConfigurationSubsystem"); @@ -420,11 +413,13 @@ namespace ccf return interface_it->second.redirections; } - bool check_session_consistency(std::shared_ptr ctx) + bool check_session_consistency( + std::shared_ptr ctx, + ccf::kv::Consensus* current_consensus) { - if (consensus != nullptr) + if (current_consensus != nullptr) { - auto current_view = consensus->get_view(); + auto current_view = current_consensus->get_view(); auto session_ctx = ctx->get_session_context(); if (!session_ctx->active_view.has_value()) { @@ -532,7 +527,8 @@ namespace ccf void forward( std::shared_ptr ctx, ccf::kv::ReadOnlyTx& /*tx*/, - const endpoints::EndpointDefinitionPtr& /*endpoint*/) + const endpoints::EndpointDefinitionPtr& /*endpoint*/, + ccf::kv::Consensus* current_consensus) { // HTTP/2 does not support forwarding if (ctx->get_http_version() == HttpVersion::HTTP2) @@ -545,7 +541,7 @@ namespace ccf return; } - if (!cmd_forwarder || (consensus == nullptr)) + if (!cmd_forwarder || current_consensus == nullptr) { ctx->set_error( HTTP_STATUS_INTERNAL_SERVER_ERROR, @@ -569,12 +565,12 @@ namespace ccf // Before attempting to forward, make sure we're in the same View as we // previously thought we were. - if (!check_session_consistency(ctx)) + if (!check_session_consistency(ctx, current_consensus)) { return; } - auto primary_id = consensus->primary(); + auto primary_id = current_consensus->primary(); if (!primary_id.has_value()) { ctx->set_error( @@ -729,16 +725,16 @@ namespace ccf return; } - // Readiness is published after consensus setup. Refresh only after the - // acquire above so this request cannot proceed with pre-publication - // cached state. - update_consensus(); + // Readiness is published after these pointers. Load them only after + // the acquire above so this request observes their publication. + auto* current_consensus = consensus.load(std::memory_order_acquire); + auto* current_history = history.load(std::memory_order_acquire); - if (consensus != nullptr) + if (current_consensus != nullptr) { if ( endpoints.apply_uncommitted_tx_backpressure() && - consensus->is_at_max_capacity()) + current_consensus->is_at_max_capacity()) { ctx->set_error( HTTP_STATUS_SERVICE_UNAVAILABLE, @@ -749,7 +745,7 @@ namespace ccf } std::unique_ptr tx_p = tables.create_tx_ptr(); - set_root_on_proposals(*ctx, *tx_p); + set_root_on_proposals(*ctx, *tx_p, current_history); if (attempts > 0) { @@ -759,7 +755,6 @@ namespace ccf } ++attempts; - update_history(); endpoint = find_endpoint(ctx, *tx_p); if (endpoint == nullptr) @@ -790,16 +785,17 @@ namespace ccf // and no forwarding is done if (redirections.has_value()) { - if (check_redirect(*tx_p, ctx, endpoint, *redirections)) + if (check_redirect( + *tx_p, ctx, endpoint, *redirections, current_consensus)) { return; } } else { - bool is_primary = - (consensus == nullptr) || consensus->can_replicate(); - const bool forwardable = (consensus != nullptr); + bool is_primary = current_consensus == nullptr || + current_consensus->can_replicate(); + const bool forwardable = current_consensus != nullptr; if (!is_primary && forwardable) { @@ -814,7 +810,7 @@ namespace ccf { if (ctx->get_session_context()->is_forwarding) { - forward(ctx, *tx_p, endpoint); + forward(ctx, *tx_p, endpoint, current_consensus); return; } break; @@ -822,7 +818,7 @@ namespace ccf case endpoints::ForwardingRequired::Always: { - forward(ctx, *tx_p, endpoint); + forward(ctx, *tx_p, endpoint, current_consensus); return; } } @@ -851,7 +847,7 @@ namespace ccf // If we've seen a View change, abandon this transaction as // inconsistent - if (!check_session_consistency(ctx)) + if (!check_session_consistency(ctx, current_consensus)) { return; } @@ -912,7 +908,7 @@ namespace ccf case ccf::kv::CommitResult::SUCCESS: { auto tx_id_opt = tx.get_txid(); - if (tx_id_opt.has_value() && consensus != nullptr) + if (tx_id_opt.has_value() && current_consensus != nullptr) { ccf::TxID tx_id = tx_id_opt.value(); @@ -963,10 +959,11 @@ namespace ccf } if ( - consensus != nullptr && consensus->can_replicate() && - history != nullptr) + current_consensus != nullptr && + current_consensus->can_replicate() && + current_history != nullptr) { - history->try_emit_signature(); + current_history->try_emit_signature(); } return; @@ -1089,6 +1086,15 @@ namespace ccf } } + void set_consensus_and_history( + ccf::kv::Consensus* consensus_, ccf::kv::TxHistory* history_) override + { + endpoints.set_consensus(consensus_); + endpoints.set_history(history_); + history.store(history_, std::memory_order_release); + consensus.store(consensus_, std::memory_order_release); + } + bool is_open() override { std::lock_guard mguard(open_lock); @@ -1096,19 +1102,20 @@ namespace ccf } void set_root_on_proposals( - const ccf::RpcContextImpl& ctx, ccf::kv::CommittableTx& tx) + const ccf::RpcContextImpl& ctx, + ccf::kv::CommittableTx& tx, + ccf::kv::TxHistory* current_history) { if (endpoints.request_needs_root(ctx)) { - update_history(); - if (history != nullptr) + if (current_history != nullptr) { // Warning: Retrieving the current TxID and root from the history // should only ever be used for the proposal creation endpoint and // nothing else. Many bad things could happen otherwise (e.g. breaking // session consistency). const auto& [txid, root, term_of_next_version] = - history->get_replicated_state_txid_and_root(); + current_history->get_replicated_state_txid_and_root(); tx.set_read_txid(txid, term_of_next_version); tx.set_root_at_read_version(root); } @@ -1125,8 +1132,6 @@ namespace ccf */ void process(std::shared_ptr ctx) override { - update_consensus(); - // NB: If we want to re-execute on backups, the original command could // be propagated from here process_command(ctx); @@ -1144,7 +1149,6 @@ namespace ccf "Processing forwarded command with unitialised forwarded context"); } - update_consensus(); process_command(ctx); if (ctx->response_is_pending) { @@ -1156,8 +1160,6 @@ namespace ccf void tick(std::chrono::milliseconds elapsed) override { - update_consensus(); - endpoints.tick(elapsed); } }; diff --git a/src/node/rpc/test/frontend_test.cpp b/src/node/rpc/test/frontend_test.cpp index 047a22b3258e..f1e8bfbc5f37 100644 --- a/src/node/rpc/test/frontend_test.cpp +++ b/src/node/rpc/test/frontend_test.cpp @@ -23,7 +23,9 @@ #include #include +#include #include +#include using namespace ccf; using namespace std; @@ -465,6 +467,33 @@ UserId user_id; MemberId member_id; MemberId invalid_member_id; +class TestNodeConfiguration : public NodeConfigurationInterface +{ +private: + StartupConfig config; + NodeConfigurationState state; + +public: + TestNodeConfiguration() : state{config, {}, true} + { + NodeInfoNetwork_v2::NetInterface interface; + interface.redirections = NodeInfoNetwork_v2::NetInterface::Redirections{}; + config.network.rpc_interfaces.emplace("test_interface", interface); + } + + const NodeConfigurationState& get() override + { + return state; + } +}; + +void publish_frontend_state(RpcHandler& frontend, NetworkState& network) +{ + const auto consensus = network.tables->get_consensus(); + const auto history = network.tables->get_history(); + frontend.set_consensus_and_history(consensus.get(), history.get()); +} + void prepare_callers(NetworkState& network) { // It is necessary to set a consensus before committing the first transaction, @@ -487,6 +516,72 @@ void prepare_callers(NetworkState& network) CHECK(tx.commit() == ccf::kv::CommitResult::SUCCESS); } +TEST_CASE("Frontend state publication is thread-safe") +{ + NetworkState network; + prepare_callers(network); + TestUserFrontend frontend(*network.tables); + + auto first_consensus = + std::make_shared(); + auto second_consensus = + std::make_shared(); + const auto history = network.tables->get_history(); + + constexpr size_t iterations = 1'000; + std::latch start(2); + std::thread publisher([&]() { + start.arrive_and_wait(); + for (size_t i = 0; i < iterations; ++i) + { + ccf::kv::Consensus* current_consensus = first_consensus.get(); + if (i % 2 != 0) + { + current_consensus = second_consensus.get(); + } + frontend.set_consensus_and_history(current_consensus, history.get()); + std::this_thread::yield(); + } + }); + + const auto request = create_simple_request("/empty_function_no_auth"); + const auto serialised_request = request.build_request(); + auto session = std::make_shared( + ccf::InvalidSessionId, anonymous_caller_der); + + bool all_requests_succeeded = true; + start.arrive_and_wait(); + for (size_t i = 0; i < iterations; ++i) + { + auto rpc_ctx = ccf::make_rpc_context(session, serialised_request); + frontend.process(rpc_ctx); + all_requests_succeeded &= rpc_ctx->get_response_status() == HTTP_STATUS_OK; + std::this_thread::yield(); + } + + publisher.join(); + REQUIRE(all_requests_succeeded); +} + +TEST_CASE("Redirect resolution handles unpublished consensus") +{ + NetworkState network; + prepare_callers(network); + TestUserFrontend frontend(*network.tables); + frontend.context.install_subsystem(std::make_shared()); + + const auto request = create_simple_request("/empty_function_no_auth"); + const auto serialised_request = request.build_request(); + auto session = std::make_shared( + ccf::InvalidSessionId, anonymous_caller_der, "test_interface"); + auto rpc_ctx = ccf::make_rpc_context(session, serialised_request); + + frontend.process(rpc_ctx); + + REQUIRE(!rpc_ctx->response_is_pending); + REQUIRE(rpc_ctx->get_response_status() == HTTP_STATUS_SERVICE_UNAVAILABLE); +} + TEST_CASE("SignedReq to and from json") { SignedReq sr; @@ -1102,6 +1197,8 @@ TEST_CASE("Forwarding" * doctest::test_suite("forwarding")) auto backup_consensus = std::make_shared(); network_backup.tables->set_consensus(backup_consensus); + publish_frontend_state(user_frontend_primary, network_primary); + publish_frontend_state(user_frontend_backup, network_backup); auto simple_call = create_simple_request(); auto serialized_call = simple_call.build_request(); @@ -1127,6 +1224,7 @@ TEST_CASE("Forwarding" * doctest::test_suite("forwarding")) { INFO("Read command is not forwarded to primary"); TestUserFrontend user_frontend_backup_read(*network_backup.tables); + publish_frontend_state(user_frontend_backup_read, network_backup); REQUIRE(channel_stub->is_empty()); user_frontend_backup_read.process(backup_ctx); @@ -1160,6 +1258,7 @@ TEST_CASE("Forwarding" * doctest::test_suite("forwarding")) }; prepare_callers(network_primary); + publish_frontend_state(user_frontend_primary, network_primary); { INFO("Valid caller"); @@ -1200,6 +1299,7 @@ TEST_CASE("Forwarding" * doctest::test_suite("forwarding")) TestUserFrontend user_frontend_backup_read(*network_backup.tables); user_frontend_backup_read.set_cmd_forwarder(backup_forwarder); + publish_frontend_state(user_frontend_backup_read, network_backup); REQUIRE(channel_stub->is_empty()); user_frontend_backup_read.process(backup_ctx); @@ -1250,6 +1350,8 @@ TEST_CASE("Nodefrontend forwarding" * doctest::test_suite("forwarding")) auto backup_consensus = std::make_shared(); network_backup.tables->set_consensus(backup_consensus); + publish_frontend_state(node_frontend_primary, network_primary); + publish_frontend_state(node_frontend_backup, network_backup); auto write_req = create_simple_request(); auto serialized_call = write_req.build_request(); @@ -1301,6 +1403,8 @@ TEST_CASE("Userfrontend forwarding" * doctest::test_suite("forwarding")) auto backup_consensus = std::make_shared(); network_backup.tables->set_consensus(backup_consensus); + publish_frontend_state(user_frontend_primary, network_primary); + publish_frontend_state(user_frontend_backup, network_backup); auto write_req = create_simple_request(); auto serialized_call = write_req.build_request(); @@ -1352,6 +1456,8 @@ TEST_CASE("Memberfrontend forwarding" * doctest::test_suite("forwarding")) auto backup_consensus = std::make_shared(); network_backup.tables->set_consensus(backup_consensus); + publish_frontend_state(member_frontend_primary, network_primary); + publish_frontend_state(member_frontend_backup, network_backup); auto write_req = create_simple_request(); auto serialized_call = write_req.build_request(); diff --git a/tsan_env_suppressions b/tsan_env_suppressions index 3ee6d5ecdee7..ee8f15a37677 100644 --- a/tsan_env_suppressions +++ b/tsan_env_suppressions @@ -8,9 +8,6 @@ deadlock:*/store.h deadlock:*/untyped_map.h -# For governance_test -race:*/node/*rpc/*frontend.h - # Race between closedir and epoll_ctl. race:closedir race:epoll_ctl From 2d8b5d3c41ed5651a0414c8956d20db5e3facdb0 Mon Sep 17 00:00:00 2001 From: GitHub Copilot Date: Fri, 7 Aug 2026 20:08:47 +0000 Subject: [PATCH 2/2] Fix endpoint registry pointer publication --- include/ccf/endpoint_registry.h | 15 +++++- samples/apps/logging/logging.cpp | 3 +- src/endpoints/base_endpoint_registry.cpp | 23 +++++---- src/endpoints/common_endpoint_registry.cpp | 3 +- src/endpoints/endpoint_registry.cpp | 4 +- src/js/registry.cpp | 3 +- src/node/rpc/frontend.h | 2 +- src/node/rpc/node_frontend.h | 54 +++++++++++++--------- src/node/rpc/test/frontend_test.cpp | 4 +- 9 files changed, 72 insertions(+), 39 deletions(-) diff --git a/include/ccf/endpoint_registry.h b/include/ccf/endpoint_registry.h index b6746632c268..be68f5f46b9c 100644 --- a/include/ccf/endpoint_registry.h +++ b/include/ccf/endpoint_registry.h @@ -9,6 +9,7 @@ #include "ccf/rpc_context.h" #include "ccf/tx.h" +#include #include #include #include @@ -172,8 +173,18 @@ namespace ccf::endpoints std::map>> templated_endpoints; - ccf::kv::Consensus* consensus = nullptr; - ccf::kv::TxHistory* history = nullptr; + std::atomic consensus{nullptr}; + std::atomic history{nullptr}; + + [[nodiscard]] ccf::kv::Consensus* get_consensus() const + { + return consensus.load(std::memory_order_acquire); + } + + [[nodiscard]] ccf::kv::TxHistory* get_history() const + { + return history.load(std::memory_order_acquire); + } public: EndpointRegistry(std::string method_prefix_) : diff --git a/samples/apps/logging/logging.cpp b/samples/apps/logging/logging.cpp index 3033d0bc7376..0b3630535838 100644 --- a/samples/apps/logging/logging.cpp +++ b/samples/apps/logging/logging.cpp @@ -1548,8 +1548,9 @@ namespace loggingapp auto is_tx_committed = [this](ccf::View view, ccf::SeqNo seqno, std::string& error_reason) { + auto* current_consensus = get_consensus(); return ccf::historical::is_tx_committed_v2( - consensus, view, seqno, error_reason); + current_consensus, view, seqno, error_reason); }; make_read_only_endpoint( "/log/private/historical", diff --git a/src/endpoints/base_endpoint_registry.cpp b/src/endpoints/base_endpoint_registry.cpp index aa488342b36b..ac704a1438ef 100644 --- a/src/endpoints/base_endpoint_registry.cpp +++ b/src/endpoints/base_endpoint_registry.cpp @@ -24,20 +24,22 @@ namespace ccf { try { - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { if (since < 1) { reason = ccf::InvalidArgsReason::ViewSmallerThanOne; return ApiResult::InvalidArgs; } - auto latest_view = consensus->get_view(); + auto latest_view = current_consensus->get_view(); if (since > latest_view) { // asking for something in the future return ApiResult::NotFound; } - const auto view_history = consensus->get_view_history_since(since); + const auto view_history = + current_consensus->get_view_history_since(since); for (ccf::View i = 0; i < view_history.size(); i++) { const auto view = i + since; @@ -67,9 +69,10 @@ namespace ccf { try { - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - tx_status = consensus->evaluate_tx_status(view, seqno); + tx_status = current_consensus->evaluate_tx_status(view, seqno); } else { @@ -88,11 +91,12 @@ namespace ccf ApiResult BaseEndpointRegistry::get_last_committed_txid_v1( ccf::View& view, ccf::SeqNo& seqno) { - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { try { - const auto [v, s] = consensus->get_committed_txid(); + const auto [v, s] = current_consensus->get_committed_txid(); view = v; seqno = s; return ApiResult::OK; @@ -199,9 +203,10 @@ namespace ccf { try { - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - const auto v = consensus->get_view(seqno); + const auto v = current_consensus->get_view(seqno); if (v != ccf::VIEW_UNKNOWN) { view = v; diff --git a/src/endpoints/common_endpoint_registry.cpp b/src/endpoints/common_endpoint_registry.cpp index 6d79f7323444..ad45e916522a 100644 --- a/src/endpoints/common_endpoint_registry.cpp +++ b/src/endpoints/common_endpoint_registry.cpp @@ -256,8 +256,9 @@ namespace ccf auto is_tx_committed = [this](ccf::View view, ccf::SeqNo seqno, std::string& error_reason) { + auto* current_consensus = get_consensus(); return ccf::historical::is_tx_committed_v2( - consensus, view, seqno, error_reason); + current_consensus, view, seqno, error_reason); }; auto get_receipt = diff --git a/src/endpoints/endpoint_registry.cpp b/src/endpoints/endpoint_registry.cpp index 498d97abb462..1d9a3f4f4fd2 100644 --- a/src/endpoints/endpoint_registry.cpp +++ b/src/endpoints/endpoint_registry.cpp @@ -663,11 +663,11 @@ namespace ccf::endpoints void EndpointRegistry::set_consensus(ccf::kv::Consensus* c) { - consensus = c; + consensus.store(c, std::memory_order_release); } void EndpointRegistry::set_history(ccf::kv::TxHistory* h) { - history = h; + history.store(h, std::memory_order_release); } } diff --git a/src/js/registry.cpp b/src/js/registry.cpp index cdbd702ffa50..c6dc7333a4da 100644 --- a/src/js/registry.cpp +++ b/src/js/registry.cpp @@ -440,8 +440,9 @@ namespace ccf::js { auto is_tx_committed = [this](ccf::View view, ccf::SeqNo seqno, std::string& error_reason) { + auto* current_consensus = get_consensus(); return ccf::historical::is_tx_committed_v2( - consensus, view, seqno, error_reason); + current_consensus, view, seqno, error_reason); }; ccf::historical::read_write_adapter_v4( diff --git a/src/node/rpc/frontend.h b/src/node/rpc/frontend.h index 6eed17354e42..345c7328235f 100644 --- a/src/node/rpc/frontend.h +++ b/src/node/rpc/frontend.h @@ -1089,8 +1089,8 @@ namespace ccf void set_consensus_and_history( ccf::kv::Consensus* consensus_, ccf::kv::TxHistory* history_) override { - endpoints.set_consensus(consensus_); endpoints.set_history(history_); + endpoints.set_consensus(consensus_); history.store(history_, std::memory_order_release); consensus.store(consensus_, std::memory_order_release); } diff --git a/src/node/rpc/node_frontend.h b/src/node/rpc/node_frontend.h index 332d4825b520..4e872da18ccf 100644 --- a/src/node/rpc/node_frontend.h +++ b/src/node/rpc/node_frontend.h @@ -534,9 +534,11 @@ namespace ccf } // Not the primary => Redirect if possible to primary - if (consensus != nullptr && !this->node_operation.can_replicate()) + auto* current_consensus = get_consensus(); + if ( + current_consensus != nullptr && !this->node_operation.can_replicate()) { - auto primary_id = consensus->primary(); + auto primary_id = current_consensus->primary(); if (primary_id.has_value()) { const auto address = node::get_redirect_address_for_node( @@ -892,10 +894,11 @@ namespace ccf out.service_data = service_value.service_data; out.current_service_create_txid = service_value.current_service_create_txid; - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - out.current_view = consensus->get_view(); - auto primary_id = consensus->primary(); + out.current_view = current_consensus->get_view(); + auto primary_id = current_consensus->primary(); if (primary_id.has_value()) { out.primary_id = primary_id.value(); @@ -975,7 +978,8 @@ namespace ccf GetNodes::Out out; auto nodes = args.tx.ro(this->network.nodes); - nodes->foreach([this, host, port, status, &out, nodes]( + auto* current_consensus = get_consensus(); + nodes->foreach([host, port, status, &out, nodes, current_consensus]( const NodeId& nid, const NodeInfo& ni) { if (status.has_value() && status.value() != ni.status) { @@ -1004,9 +1008,9 @@ namespace ccf } bool is_primary = false; - if (consensus != nullptr) + if (current_consensus != nullptr) { - is_primary = consensus->primary() == nid; + is_primary = current_consensus->primary() == nid; } out.nodes.push_back( @@ -1197,9 +1201,10 @@ namespace ccf } bool is_primary = false; - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - auto primary = consensus->primary(); + auto primary = current_consensus->primary(); if (primary.has_value() && primary.value() == node_id) { is_primary = true; @@ -1228,9 +1233,10 @@ namespace ccf auto info = nodes->get(node_id); bool is_primary = false; - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - auto primary = consensus->primary(); + auto primary = current_consensus->primary(); if (primary.has_value() && primary.value() == node_id) { is_primary = true; @@ -1281,9 +1287,10 @@ namespace ccf .install(); auto get_primary_node = [this](auto& args, nlohmann::json&&) { - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - auto primary_id = consensus->primary(); + auto primary_id = current_consensus->primary(); if (!primary_id.has_value()) { return make_error( @@ -1333,7 +1340,8 @@ namespace ccf } else { - if (consensus == nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus == nullptr) { args.rpc_ctx->set_error( HTTP_STATUS_INTERNAL_SERVER_ERROR, @@ -1342,7 +1350,7 @@ namespace ccf return; } - auto primary_id = consensus->primary(); + auto primary_id = current_consensus->primary(); if (!primary_id.has_value()) { args.rpc_ctx->set_error( @@ -1405,9 +1413,10 @@ namespace ccf auto consensus_config = [this](auto& /*args*/, nlohmann::json&&) { // Query node for configurations, separate current from pending - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - auto cfg = consensus->get_latest_configuration(); + auto cfg = current_consensus->get_latest_configuration(); ConsensusConfig cc; for (auto& [nid, ninfo] : cfg) { @@ -1435,9 +1444,11 @@ namespace ccf .install(); auto consensus_state = [this](auto& /*args*/, nlohmann::json&&) { - if (consensus != nullptr) + auto* current_consensus = get_consensus(); + if (current_consensus != nullptr) { - return make_success(ConsensusConfigDetails{consensus->get_details()}); + return make_success( + ConsensusConfigDetails{current_consensus->get_details()}); } return make_error( @@ -1699,7 +1710,8 @@ namespace ccf auto refresh_jwt_keys = [this](auto& ctx, nlohmann::json&& body) { // All errors are server errors since the client is the server. - auto primary_id = consensus->primary(); + auto* current_consensus = get_consensus(); + auto primary_id = current_consensus->primary(); if (!primary_id.has_value()) { LOG_FAIL_FMT("JWT key auto-refresh: primary unknown"); diff --git a/src/node/rpc/test/frontend_test.cpp b/src/node/rpc/test/frontend_test.cpp index f1e8bfbc5f37..51a29457e287 100644 --- a/src/node/rpc/test/frontend_test.cpp +++ b/src/node/rpc/test/frontend_test.cpp @@ -544,7 +544,9 @@ TEST_CASE("Frontend state publication is thread-safe") } }); - const auto request = create_simple_request("/empty_function_no_auth"); + auto request = create_simple_request("/tx"); + request.set_method(HTTP_GET); + request.set_query_param("transaction_id", "1.1"); const auto serialised_request = request.build_request(); auto session = std::make_shared( ccf::InvalidSessionId, anonymous_caller_der);