[WIP] Refactor Consensus API and KV Interaction

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]
 

doc/schemas/node_openapi.json

@@ -134,6 +134,12 @@
           "configs": {
             "$ref": "#/components/schemas/Configuration_array"
           },
+          "committed_seqno": {
+            "$ref": "#/components/schemas/uint64"
+          },
+          "committed_view": {
+            "$ref": "#/components/schemas/uint64"
+          },
           "current_view": {
             "$ref": "#/components/schemas/uint64"
           },
@@ -157,11 +163,15 @@
           },
           "ticking": {
             "$ref": "#/components/schemas/boolean"
+          },
+          "view_history": {
+            "$ref": "#/components/schemas/ConsensusViewHistory"
           }
         },
         "required": [
           "configs",
           "acks",
+          "view_history",
           "membership_state",
           "primary_id",
           "current_view",
@@ -184,6 +194,17 @@
         ],
         "type": "object"
       },
+      "ConsensusViewHistory": {
+        "properties": {
+          "starts": {
+            "$ref": "#/components/schemas/uint64_array"
+          }
+        },
+        "required": [
+          "starts"
+        ],
+        "type": "object"
+      },
       "ConsensusNodeConfig": {
         "properties": {
           "address": {
@@ -899,6 +920,12 @@
         "maximum": 18446744073709551615,
         "minimum": 0,
         "type": "integer"
+      },
+      "uint64_array": {
+        "items": {
+          "$ref": "#/components/schemas/uint64"
+        },
+        "type": "array"
       }
     },
     "x-ccf-forwarding": {

src/consensus/aft/raft.h

@@ -247,32 +247,11 @@ namespace aft
 
     ~Aft() override = default;
 
-    std::optional<ccf::NodeId> primary() override
-    {
-      return leader_id;
-    }
-
-    ccf::NodeId id() override
+    ccf::NodeId id()
     {
       return state->node_id;
     }
 
-    bool is_primary() override
-    {
-      return state->leadership_state == ccf::kv::LeadershipState::Leader;
-    }
-
-    bool is_candidate() override
-    {
-      return state->leadership_state == ccf::kv::LeadershipState::Candidate;
-    }
-
-    bool can_replicate() override
-    {
-      std::unique_lock<ccf::pal::Mutex> guard(state->lock);
-      return can_replicate_unsafe();
-    }
-
     /**
      * Returns true if the node is primary, max_uncommitted_tx_count is non-zero
      * and the number of transactions replicated but not yet committed exceeds
@@ -303,16 +282,7 @@ namespace aft
       return Consensus::SignatureDisposition::CANT_REPLICATE;
     }
 
-    bool is_backup() override
-    {
-      return state->leadership_state == ccf::kv::LeadershipState::Follower;
-    }
-
-    bool is_active() const
-    {
-      return state->membership_state == ccf::kv::MembershipState::Active;
-    }
-
+  private:
     bool is_retired() const
     {
       return state->membership_state == ccf::kv::MembershipState::Retired;
@@ -324,18 +294,13 @@ namespace aft
         state->retirement_phase == ccf::kv::RetirementPhase::RetiredCommitted;
     }
 
-    bool is_retired_completed() const
-    {
-      return state->membership_state == ccf::kv::MembershipState::Retired &&
-        state->retirement_phase == ccf::kv::RetirementPhase::Completed;
-    }
-
+  public:
     void set_retired_committed(
       ccf::SeqNo seqno, const std::vector<ccf::kv::NodeId>& node_ids) override
     {
       for (const auto& node_id : node_ids)
       {
-        if (id() == node_id)
+        if (state->node_id == node_id)
         {
           CCF_ASSERT(
             state->membership_state == ccf::kv::MembershipState::Retired,
@@ -464,41 +429,21 @@ namespace aft
       return state->last_idx;
     }
 
-    Index get_committed_seqno() override
+    std::vector<Index> get_view_history(Index idx)
     {
-      std::lock_guard<ccf::pal::Mutex> guard(state->lock);
-      return get_commit_idx_unsafe();
-    }
-
-    Term get_view() override
-    {
-      std::lock_guard<ccf::pal::Mutex> guard(state->lock);
-      return state->current_view;
+      return state->view_history.get_history_until(idx);
     }
 
-    std::pair<Term, Index> get_committed_txid() override
+    ccf::TxStatus evaluate_tx_status(
+      ccf::View target_view, ccf::SeqNo target_seqno) override
     {
       std::lock_guard<ccf::pal::Mutex> guard(state->lock);
-      ccf::SeqNo commit_idx = get_commit_idx_unsafe();
-      return {get_term_internal(commit_idx), commit_idx};
-    }
+      const auto local_view = get_term_internal(target_seqno);
+      const auto committed_seqno = get_commit_idx_unsafe();
+      const auto committed_view = get_term_internal(committed_seqno);
 
-    Term get_view(Index idx) override
-    {
-      std::lock_guard<ccf::pal::Mutex> guard(state->lock);
-      return get_term_internal(idx);
-    }
-
-    std::vector<Index> get_view_history(Index idx) override
-    {
-      // This should only be called when the spin lock is held.
-      return state->view_history.get_history_until(idx);
-    }
-
-    std::vector<Index> get_view_history_since(Index idx) override
-    {
-      // This should only be called when the spin lock is held.
-      return state->view_history.get_history_since(idx);
+      return ccf::evaluate_tx_status(
+        target_view, target_seqno, local_view, committed_view, committed_seqno);
     }
 
     // Same as ccfraft.tla GetServerSet/IsInServerSet
@@ -579,45 +524,70 @@ namespace aft
       }
     }
 
-    Configuration::Nodes get_latest_configuration_unsafe() const override
+    void update_configuration(
+      Index idx, const Configuration::NodeChanges& changes) override
     {
-      if (configurations.empty())
+      if (changes.empty())
       {
-        return {};
+        return;
       }
 
-      return configurations.back().nodes;
-    }
-
-    Configuration::Nodes get_latest_configuration() override
-    {
-      std::lock_guard<ccf::pal::Mutex> guard(state->lock);
-      return get_latest_configuration_unsafe();
+      auto configuration = configurations.empty() ? Configuration::Nodes{} :
+                                                    configurations.back().nodes;
+      for (const auto& [node_id, node_info] : changes)
+      {
+        if (node_info.has_value())
+        {
+          configuration.try_emplace(node_id, node_info.value());
+        }
+        else
+        {
+          configuration.erase(node_id);
+        }
+      }
+      add_configuration(idx, configuration);
     }
 
-    ccf::kv::ConsensusDetails get_details() override
+  private:
+    ccf::kv::ConsensusLightDetails get_light_details_unsafe()
     {
-      ccf::kv::ConsensusDetails details;
-      std::lock_guard<ccf::pal::Mutex> 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;
       if (is_retired())
       {
         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<ccf::pal::Mutex> guard(state->lock);
+      return get_light_details_unsafe();
+    }
+
+    ccf::kv::ConsensusDetails get_details() override
+    {
+      ccf::kv::ConsensusDetails details;
+      std::lock_guard<ccf::pal::Mutex> guard(state->lock);
+      static_cast<ccf::kv::ConsensusLightDetails&>(details) =
+        get_light_details_unsafe();
+      details.configs.assign(configurations.begin(), configurations.end());
+      details.view_history.starts = state->view_history.get_history_until();
       for (auto& [k, v] : all_other_nodes)
       {
         details.acks[k] = {
           v.match_idx, static_cast<size_t>(v.last_ack_timeout.count())};
       }
-      details.reconfiguration_type = ccf::ReconfigurationType::ONE_TRANSACTION;
       return details;
     }
 
@@ -2607,7 +2577,10 @@ namespace aft
       }
 
       RAFT_DEBUG_FMT("Compacting...");
-      store->compact(idx);
+      store->compact(
+        idx,
+        state->leadership_state == ccf::kv::LeadershipState::Leader,
+        {state->view_history.get_history_until(idx)});
       ledger->commit(idx);
 
       if (commit_callbacks != nullptr)
@@ -2650,7 +2623,9 @@ namespace aft
       if (changed)
       {
         create_and_remove_node_state();
-        if (retired_node_cleanup && is_primary())
+        if (
+          retired_node_cleanup &&
+          state->leadership_state == ccf::kv::LeadershipState::Leader)
         {
           retired_node_cleanup->cleanup();
         }

src/consensus/aft/raft_types.h

@@ -26,7 +26,10 @@ namespace aft
   {
   public:
     virtual ~Store() = default;
-    virtual void compact(Index v) = 0;
+    virtual void compact(
+      Index v,
+      bool is_primary,
+      const ccf::kv::ConsensusViewHistory& view_history) = 0;
     virtual void rollback(
       const ccf::TxID& tx_id, Term term_of_next_version) = 0;
     virtual void initialise_term(Term t) = 0;
@@ -45,12 +48,15 @@ namespace aft
   public:
     Adaptor(std::shared_ptr<T> x) : x(x) {}
 
-    void compact(Index v) override
+    void compact(
+      Index v,
+      bool is_primary,
+      const ccf::kv::ConsensusViewHistory& view_history) override
     {
       auto p = x.lock();
       if (p)
       {
-        p->compact(v);
+        p->compact(v, is_primary, view_history);
       }
     }
 

src/consensus/aft/test/driver.h

@@ -74,10 +74,13 @@ struct LoggingStubStore_Mermaid : public aft::LoggingStubStoreConfig
 {
   using LoggingStubStoreConfig::LoggingStubStoreConfig;
 
-  void compact(aft::Index idx) override
+  void compact(
+    aft::Index idx,
+    bool is_primary,
+    const ccf::kv::ConsensusViewHistory& view_history) override
   {
     RAFT_DRIVER_PRINT("{}->>{}: [KV] compacting to {}", _id, _id, idx);
-    aft::LoggingStubStoreConfig::compact(idx);
+    aft::LoggingStubStoreConfig::compact(idx, is_primary, view_history);
   }
 
   void rollback(const ccf::TxID& tx_id, aft::Term t) override
@@ -100,7 +103,7 @@ struct LoggingStubStore_Mermaid : public aft::LoggingStubStoreConfig
 };
 
 using ms = std::chrono::milliseconds;
-using TRaft = aft::Aft<LedgerStubProxy_Mermaid>;
+using TRaft = aft::TestAft<LedgerStubProxy_Mermaid>;
 using Store = LoggingStubStore_Mermaid;
 using Adaptor = aft::Adaptor<Store>;
 
@@ -633,7 +636,7 @@ class RaftDriver
     std::vector<std::string> 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())
       {