Fix RpcFrontend consensus and history publication

include/ccf/endpoint_registry.h

@@ -9,6 +9,7 @@
 #include "ccf/rpc_context.h"
 #include "ccf/tx.h"
 
+#include <atomic>
 #include <charconv>
 #include <functional>
 #include <llhttp/llhttp.h>
@@ -172,8 +173,18 @@ namespace ccf::endpoints
       std::map<RESTVerb, std::shared_ptr<PathTemplatedEndpoint>>>
       templated_endpoints;
 
-    ccf::kv::Consensus* consensus = nullptr;
-    ccf::kv::TxHistory* history = nullptr;
+    std::atomic<ccf::kv::Consensus*> consensus{nullptr};
+    std::atomic<ccf::kv::TxHistory*> 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_) :

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",

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<RpcContextImpl> ctx) = 0;

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;

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 =

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);
   }
 }

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(

src/node/node_state.h

@@ -3531,6 +3531,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(