kv_types.h

Go to the documentation of this file.

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the Apache 2.0 License.
#pragma once
 
#include "ccf/claims_digest.h"
#include "ccf/crypto/hash_bytes.h"
#include "ccf/crypto/pem.h"
#include "ccf/ds/hex.h"
#include "ccf/ds/nonstd.h"
#include "ccf/entity_id.h"
#include "ccf/kv/get_name.h"
#include "ccf/kv/hooks.h"
#include "ccf/kv/version.h"
#include "ccf/node/cose_signatures_config.h"
#include "ccf/node/startup_config.h"
#include "ccf/service/consensus_type.h"
#include "ccf/service/reconfiguration_type.h"
#include "ccf/tx_id.h"
#include "ccf/tx_status.h"
#include "crypto/openssl/ec_key_pair.h"
#include "kv/ledger_chunker_interface.h"
#include "serialiser_declare.h"
 
#include <array>
#include <chrono>
#include <functional>
#include <limits>
#include <list>
#include <memory>
#include <set>
#include <string>
#include <unordered_set>
#include <vector>
 
namespace ccf
{
  struct PrimarySignature;
}
 
namespace ccf::kv
{
  // Term describes an epoch of Versions. It is incremented when global kv's
  // writer(s) changes. Term and Version combined give a unique identifier for
  // all accepted kv modifications. Terms are handled by Consensus via the
  // TermHistory
  using Term = uint64_t;
  using NodeId = ccf::NodeId;
 
  using ReconfigurationId = uint64_t;
 
  struct Configuration
  {
    struct NodeInfo
    {
      std::string hostname;
      std::string port;
 
      NodeInfo() = default;
 
      NodeInfo(std::string hostname_, std::string port_) :
        hostname(std::move(hostname_)),
        port(std::move(port_))
      {}
 
      bool operator==(const NodeInfo& other) const
      {
        return hostname == other.hostname && port == other.port;
      }
    };
 
    using Nodes = std::map<NodeId, NodeInfo>;
 
    ccf::SeqNo idx = 0;
    Nodes nodes;
    ReconfigurationId rid = 0;
  };
 
  inline void to_json(nlohmann::json& j, const Configuration::NodeInfo& ni)
  {
    j["address"] = fmt::format("{}:{}", ni.hostname, ni.port);
  }
 
  inline void from_json(const nlohmann::json& j, Configuration::NodeInfo& ni)
  {
    const std::string addr(j["address"]);
    const auto& [h, p] = ccf::nonstd::split_1(addr, ":");
    ni.hostname = h;
    ni.port = p;
  }
 
  inline std::string schema_name(const Configuration::NodeInfo* /*unused*/)
  {
    return "Configuration__NodeInfo";
  }
 
  inline void fill_json_schema(
    nlohmann::json& schema, const Configuration::NodeInfo* /*unused*/)
  {
    schema["type"] = "object";
    schema["required"] = nlohmann::json::array();
    schema["required"].push_back("address");
    schema["properties"] = nlohmann::json::object();
    schema["properties"]["address"] = nlohmann::json::object();
    schema["properties"]["address"]["$ref"] = "#/components/schemas/string";
  }
 
  enum class LeadershipState : uint8_t
  {
    None,
    Leader,
    Follower,
    PreVoteCandidate,
    Candidate,
  };
 
  DECLARE_JSON_ENUM(
    LeadershipState,
    {{LeadershipState::None, "None"},
     {LeadershipState::Leader, "Leader"},
     {LeadershipState::Follower, "Follower"},
     {LeadershipState::PreVoteCandidate, "PreVoteCandidate"},
     {LeadershipState::Candidate, "Candidate"}});
 
  enum class MembershipState : uint8_t
  {
    Active,
    Retired
  };
 
  DECLARE_JSON_ENUM(
    MembershipState,
    {{MembershipState::Active, "Active"},
     {MembershipState::Retired, "Retired"}});
 
  enum class RetirementPhase : uint8_t
  {
    Ordered = 1,
    Signed = 2,
    Completed = 3,
    RetiredCommitted = 4,
  };
 
  DECLARE_JSON_ENUM(
    RetirementPhase,
    {{RetirementPhase::Ordered, "Ordered"},
     {RetirementPhase::Signed, "Signed"},
     {RetirementPhase::Completed, "Completed"},
     {RetirementPhase::RetiredCommitted, "RetiredCommitted"}});
 
  DECLARE_JSON_TYPE(Configuration);
  DECLARE_JSON_REQUIRED_FIELDS(Configuration, idx, nodes, rid);
 
  struct ConsensusDetails
  {
    struct Ack
    {
      ccf::SeqNo seqno;
      size_t last_received_ms;
    };
 
    std::vector<Configuration> configs;
    std::unordered_map<ccf::NodeId, Ack> acks;
    MembershipState membership_state{};
    std::optional<LeadershipState> leadership_state = std::nullopt;
    std::optional<RetirementPhase> retirement_phase = std::nullopt;
    std::optional<std::unordered_map<ccf::NodeId, ccf::SeqNo>> learners =
      std::nullopt;
    std::optional<ccf::ReconfigurationType> reconfiguration_type = std::nullopt;
    std::optional<ccf::NodeId> primary_id = std::nullopt;
    ccf::View current_view = 0;
    bool ticking = false;
  };
 
  DECLARE_JSON_TYPE(ConsensusDetails::Ack);
  DECLARE_JSON_REQUIRED_FIELDS(ConsensusDetails::Ack, seqno, last_received_ms);
 
  DECLARE_JSON_TYPE_WITH_OPTIONAL_FIELDS(ConsensusDetails);
  DECLARE_JSON_REQUIRED_FIELDS(
    ConsensusDetails,
    configs,
    acks,
    membership_state,
    primary_id,
    current_view,
    ticking);
  DECLARE_JSON_OPTIONAL_FIELDS(
    ConsensusDetails,
    reconfiguration_type,
    learners,
    leadership_state,
    retirement_phase);
 
  class ConfigurableConsensus
  {
  public:
    virtual ~ConfigurableConsensus() = default;
    virtual void add_configuration(
      ccf::SeqNo seqno, const Configuration::Nodes& conf) = 0;
    virtual Configuration::Nodes get_latest_configuration() = 0;
    [[nodiscard]] virtual Configuration::Nodes get_latest_configuration_unsafe()
      const = 0;
    virtual ConsensusDetails get_details() = 0;
  };
 
  using BatchVector = std::vector<std::tuple<
    Version,
    std::shared_ptr<std::vector<uint8_t>>,
    bool,
    std::shared_ptr<ConsensusHookPtrs>>>;
 
  enum CommitResult : uint8_t
  {
    SUCCESS = 1,
    FAIL_CONFLICT = 2,
    FAIL_NO_REPLICATE = 3
  };
 
  enum SecurityDomain : uint8_t
  {
    PUBLIC, // Public domain indicates the version and always appears first
    PRIVATE,
    SECURITY_DOMAIN_MAX
  };
 
  enum AccessCategory : uint8_t
  {
    INTERNAL,
    GOVERNANCE,
    APPLICATION
  };
 
  enum class EntryType : uint8_t
  {
    WriteSet = 0,
    Snapshot = 1,
    WriteSetWithClaims = 2,
    WriteSetWithCommitEvidence = 3,
    WriteSetWithCommitEvidenceAndClaims = 4,
    MAX = WriteSetWithCommitEvidenceAndClaims
  };
 
  static bool has_claims(const EntryType& et)
  {
    return et == EntryType::WriteSetWithClaims ||
      et == EntryType::WriteSetWithCommitEvidenceAndClaims;
  }
 
  static bool has_commit_evidence(const EntryType& et)
  {
    return et == EntryType::WriteSetWithCommitEvidence ||
      et == EntryType::WriteSetWithCommitEvidenceAndClaims;
  }
 
  // EntryType must be backwards compatible with the older
  // bool is_snapshot field
  static_assert(sizeof(EntryType) == sizeof(bool));
 
  constexpr auto public_domain_prefix = "public:";
 
  static inline SecurityDomain get_security_domain(const std::string& name)
  {
    if (name.starts_with(public_domain_prefix))
    {
      return SecurityDomain::PUBLIC;
    }
 
    return SecurityDomain::PRIVATE;
  }
 
  static inline std::pair<SecurityDomain, AccessCategory> parse_map_name(
    std::string_view name)
  {
    constexpr auto internal_category_prefix = "ccf.internal.";
    constexpr auto governance_category_prefix = "ccf.gov.";
    constexpr auto reserved_category_prefix = "ccf.";
 
    auto security_domain = SecurityDomain::PRIVATE;
    if (name.starts_with(public_domain_prefix))
    {
      name.remove_prefix(strlen(public_domain_prefix));
      security_domain = SecurityDomain::PUBLIC;
    }
 
    auto access_category = AccessCategory::APPLICATION;
    if (name.starts_with(internal_category_prefix))
    {
      access_category = AccessCategory::INTERNAL;
    }
    else if (name.starts_with(governance_category_prefix))
    {
      access_category = AccessCategory::GOVERNANCE;
    }
    else if (name.starts_with(reserved_category_prefix))
    {
      throw std::logic_error(fmt::format(
        "Map name '{}' includes disallowed reserved prefix '{}'",
        name,
        reserved_category_prefix));
    }
 
    return {security_domain, access_category};
  }
 
  enum ApplyResult : uint8_t
  {
    PASS = 1,
    PASS_SIGNATURE = 2,
    PASS_BACKUP_SIGNATURE = 3,
    PASS_BACKUP_SIGNATURE_SEND_ACK = 4,
    PASS_NONCES = 5,
    PASS_NEW_VIEW = 6,
    PASS_ENCRYPTED_PAST_LEDGER_SECRET = 8,
    PASS_APPLY = 9,
    FAIL = 10
  };
 
  class KvSerialiserException : public std::exception
  {
  private:
    std::string msg;
 
  public:
    KvSerialiserException(std::string msg_) : msg(std::move(msg_)) {}
 
    [[nodiscard]] const char* what() const noexcept override
    {
      return msg.c_str();
    }
  };
 
  class TxHistory
  {
  public:
    virtual ~TxHistory() = default;
    virtual bool verify_root_signatures(ccf::kv::Version version) = 0;
    virtual void try_emit_signature() = 0;
    virtual void emit_signature() = 0;
    virtual ccf::crypto::Sha256Hash get_replicated_state_root() = 0;
    virtual std::tuple<
      ccf::TxID /* TxID of last transaction seen by history */,
      ccf::crypto::Sha256Hash /* root as of TxID */,
      ccf::kv::Term /* term_of_next_version */>
    get_replicated_state_txid_and_root() = 0;
    virtual std::vector<uint8_t> get_proof(Version v) = 0;
    virtual bool verify_proof(const std::vector<uint8_t>& proof) = 0;
    virtual bool init_from_snapshot(
      const std::vector<uint8_t>& hash_at_snapshot) = 0;
    virtual std::vector<uint8_t> get_raw_leaf(uint64_t index) = 0;
    virtual void append(const std::vector<uint8_t>& data) = 0;
    virtual void append_entry(
      const ccf::crypto::Sha256Hash& digest,
      std::optional<ccf::kv::Term> expected_term = std::nullopt) = 0;
    virtual void rollback(
      const ccf::TxID& tx_id, ccf::kv::Term term_of_next_version_) = 0;
    virtual void compact(Version v) = 0;
    virtual void set_term(ccf::kv::Term) = 0;
    virtual std::vector<uint8_t> serialise_tree(size_t to) = 0;
    virtual void set_endorsed_certificate(const ccf::crypto::Pem& cert) = 0;
    virtual void start_signature_emit_timer() = 0;
    virtual void set_service_signing_identity(
      std::shared_ptr<ccf::crypto::ECKeyPair_OpenSSL> keypair,
      const COSESignaturesConfig& cose_signatures) = 0;
    virtual const ccf::COSESignaturesConfig& get_cose_signatures_config() = 0;
  };
 
  class Consensus : public ConfigurableConsensus
  {
  public:
    ~Consensus() override = default;
 
    virtual NodeId id() = 0;
    virtual bool is_primary() = 0;
    virtual bool is_backup() = 0;
    virtual bool is_candidate() = 0;
    virtual bool can_replicate() = 0;
    virtual bool is_at_max_capacity() = 0;
 
    enum class SignatureDisposition : uint8_t
    {
      CANT_REPLICATE,
      CAN_SIGN,
      SHOULD_SIGN,
    };
    virtual SignatureDisposition get_signature_disposition() = 0;
 
    virtual void force_become_primary() = 0;
    virtual void force_become_primary(
      ccf::SeqNo, ccf::View, const std::vector<ccf::SeqNo>&, ccf::SeqNo) = 0;
    virtual void init_as_backup(
      ccf::SeqNo, ccf::View, const std::vector<ccf::SeqNo>&, ccf::SeqNo) = 0;
 
    virtual bool replicate(const BatchVector& entries, ccf::View view) = 0;
    virtual std::pair<ccf::View, ccf::SeqNo> get_committed_txid() = 0;
 
    virtual ccf::View get_view(ccf::SeqNo seqno) = 0;
    virtual ccf::View get_view() = 0;
    virtual std::vector<ccf::SeqNo> get_view_history(
      ccf::SeqNo seqno = std::numeric_limits<ccf::SeqNo>::max()) = 0;
    virtual std::vector<ccf::SeqNo> get_view_history_since(
      ccf::SeqNo seqno) = 0;
    virtual ccf::SeqNo get_committed_seqno() = 0;
    virtual std::optional<NodeId> primary() = 0;
 
    virtual void recv_message(
      const NodeId& from, const uint8_t* data, size_t size) = 0;
 
    virtual void periodic(std::chrono::milliseconds /*elapsed*/) {}
    virtual void periodic_end() {}
 
    virtual void enable_all_domains() {}
 
    virtual void set_retired_committed(
      ccf::SeqNo /*seqno*/, const std::vector<NodeId>& node_ids)
    {}
 
    virtual void nominate_successor() {}
 
    ccf::TxStatus evaluate_tx_status(
      ccf::View target_view, ccf::SeqNo target_seqno)
    {
      const auto local_view = get_view(target_seqno);
      const auto [committed_view, committed_seqno] = get_committed_txid();
 
      return ccf::evaluate_tx_status(
        target_view, target_seqno, local_view, committed_view, committed_seqno);
    }
  };
 
  struct PendingTxInfo
  {
    CommitResult success;
    std::vector<uint8_t> data;
    ccf::ClaimsDigest claims_digest;
    ccf::crypto::Sha256Hash commit_evidence_digest;
    std::vector<ConsensusHookPtr> hooks;
 
    PendingTxInfo(
      CommitResult success_,
      std::vector<uint8_t>&& data_,
      ccf::ClaimsDigest&& claims_digest_,
      ccf::crypto::Sha256Hash&& commit_evidence_digest_,
      std::vector<ConsensusHookPtr>&& hooks_) :
      success(success_),
      data(std::move(data_)),
      claims_digest(claims_digest_),
      commit_evidence_digest(commit_evidence_digest_),
      hooks(std::move(hooks_))
    {}
  };
 
  class PendingTx
  {
  public:
    virtual PendingTxInfo call() = 0;
    virtual ~PendingTx() = default;
  };
 
  class MovePendingTx : public PendingTx
  {
  private:
    std::vector<uint8_t> data;
    ccf::ClaimsDigest claims_digest;
    ccf::crypto::Sha256Hash commit_evidence_digest;
    ConsensusHookPtrs hooks;
 
  public:
    MovePendingTx(
      std::vector<uint8_t>&& data_,
      ccf::ClaimsDigest&& claims_digest_,
      ccf::crypto::Sha256Hash&& commit_evidence_digest_,
      ConsensusHookPtrs&& hooks_) :
      data(std::move(data_)),
      claims_digest(std::move(claims_digest_)),
      commit_evidence_digest(std::move(commit_evidence_digest_)),
      hooks(std::move(hooks_))
    {}
 
    PendingTxInfo call() override
    {
      return {
        CommitResult::SUCCESS,
        std::move(data),
        std::move(claims_digest),
        std::move(commit_evidence_digest),
        std::move(hooks)};
    }
  };
 
  class AbstractTxEncryptor
  {
  public:
    virtual ~AbstractTxEncryptor() = default;
 
    virtual bool encrypt(
      const std::vector<uint8_t>& plain,
      const std::vector<uint8_t>& additional_data,
      std::vector<uint8_t>& serialised_header,
      std::vector<uint8_t>& cipher,
      const ccf::TxID& tx_id,
      EntryType entry_type = EntryType::WriteSet,
      bool historical_hint = false) = 0;
    virtual bool decrypt(
      const std::vector<uint8_t>& cipher,
      const std::vector<uint8_t>& additional_data,
      const std::vector<uint8_t>& serialised_header,
      std::vector<uint8_t>& plain,
      Version version,
      Term& term,
      bool historical_hint = false) = 0;
 
    virtual void rollback(Version version) = 0;
 
    virtual size_t get_header_length() = 0;
    virtual uint64_t get_term(const uint8_t* data, size_t size) = 0;
 
    virtual ccf::crypto::HashBytes get_commit_nonce(
      const ccf::TxID& tx_id, bool historical_hint = false) = 0;
 
    std::string get_commit_evidence(
      const ccf::TxID& tx_id, bool historical_hint = false)
    {
      auto nonce = get_commit_nonce(tx_id, historical_hint);
      return fmt::format(
        "ce:{}.{}:{}", tx_id.view, tx_id.seqno, ccf::ds::to_hex(nonce));
    }
  };
  using EncryptorPtr = std::shared_ptr<AbstractTxEncryptor>;
 
  class AbstractSnapshotter
  {
  public:
    virtual ~AbstractSnapshotter() = default;
 
    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 rollback(ccf::kv::Version v) = 0;
  };
  using SnapshotterPtr = std::shared_ptr<AbstractSnapshotter>;
 
  class AbstractChangeSet
  {
  public:
    virtual ~AbstractChangeSet() = default;
 
    [[nodiscard]] virtual bool has_writes() const = 0;
  };
 
  class AbstractCommitter
  {
  public:
    virtual ~AbstractCommitter() = default;
 
    virtual bool has_writes() = 0;
    virtual bool prepare() = 0;
    virtual void commit(Version v, bool track_deletes_on_missing_keys) = 0;
    virtual ConsensusHookPtr post_commit() = 0;
  };
 
  class AbstractStore;
  class AbstractMap : public std::enable_shared_from_this<AbstractMap>,
                      public GetName
  {
  public:
    class Snapshot
    {
    public:
      virtual ~Snapshot() = default;
      virtual void serialise(KvStoreSerialiser& s) = 0;
      [[nodiscard]] virtual SecurityDomain get_security_domain() const = 0;
    };
 
    using GetName::GetName;
    ~AbstractMap() override = default;
 
    virtual std::unique_ptr<AbstractCommitter> create_committer(
      AbstractChangeSet* changes) = 0;
 
    virtual AbstractStore* get_store() = 0;
    virtual void serialise_changes(
      const AbstractChangeSet* changes,
      KvStoreSerialiser& s,
      bool include_reads) = 0;
    virtual void compact(Version v) = 0;
    virtual std::unique_ptr<Snapshot> snapshot(Version v) = 0;
    virtual void post_compact() = 0;
    virtual void rollback(Version v) = 0;
    virtual void lock() = 0;
    virtual void unlock() = 0;
    virtual SecurityDomain get_security_domain() = 0;
    virtual void clear() = 0;
 
    virtual AbstractMap* clone(AbstractStore* store) = 0;
    virtual void swap(AbstractMap* map) = 0;
  };
 
  class Tx;
 
  class AbstractExecutionWrapper
  {
  public:
    virtual ~AbstractExecutionWrapper() = default;
    virtual ccf::kv::ApplyResult apply(
      bool track_deletes_on_missing_keys = false) = 0;
    virtual ccf::kv::ConsensusHookPtrs& get_hooks() = 0;
    virtual const std::vector<uint8_t>& get_entry() = 0;
    virtual ccf::kv::Term get_term() = 0;
    virtual ccf::kv::Version get_index() = 0;
    virtual bool support_async_execution() = 0;
    virtual bool is_public_only() = 0;
    virtual ccf::ClaimsDigest&& consume_claims_digest() = 0;
    virtual std::optional<ccf::crypto::Sha256Hash>&&
    consume_commit_evidence_digest() = 0;
 
    // Setting a short rollback is a work around that should be fixed
    // shortly. In BFT mode when we deserialize and realize we need to
    // create a new map we remember this. If we need to create the same
    // map multiple times (for tx in the same group of append entries) the
    // first create successes but the second fails because the map is
    // already there. This works around the problem by stopping just
    // before the 2nd create (which failed at this point) and when the
    // primary resends the append entries we will succeed as the map is
    // already there. This will only occur on BFT startup so not a perf
    // problem but still need to be resolved.
    //
    // Thus, a large rollback is one which did not result from the map creating
    // issue. https://github.com/microsoft/CCF/issues/2799
    virtual bool should_rollback_to_last_committed() = 0;
  };
 
  class AbstractStore
  {
  public:
    class AbstractSnapshot
    {
    public:
      virtual ~AbstractSnapshot() = default;
      [[nodiscard]] virtual Version get_version() const = 0;
      virtual std::vector<uint8_t> serialise(
        const std::shared_ptr<AbstractTxEncryptor>& encryptor) = 0;
    };
 
    virtual ~AbstractStore() = default;
 
    virtual void lock_map_set() = 0;
    virtual void unlock_map_set() = 0;
 
    virtual Version next_version() = 0;
    virtual std::tuple<Version, Version> next_version(bool commit_new_map) = 0;
    virtual ccf::TxID next_txid() = 0;
 
    virtual Version current_version() = 0;
    virtual ccf::TxID current_txid() = 0;
    virtual std::pair<ccf::TxID, Term> current_txid_and_commit_term() = 0;
 
    virtual Version compacted_version() = 0;
    virtual Term commit_view() = 0;
 
    virtual std::shared_ptr<AbstractMap> get_map(
      Version v, const std::string& map_name) = 0;
    virtual std::shared_ptr<AbstractMap> get_map_unsafe(
      Version v, const std::string& map_name) = 0;
    virtual void add_dynamic_map(
      Version v, const std::shared_ptr<AbstractMap>& map) = 0;
 
    virtual std::shared_ptr<Consensus> get_consensus() = 0;
    virtual std::shared_ptr<TxHistory> get_history() = 0;
    virtual std::shared_ptr<ILedgerChunker> get_chunker() = 0;
    virtual EncryptorPtr get_encryptor() = 0;
    virtual std::unique_ptr<AbstractExecutionWrapper> deserialize(
      const std::vector<uint8_t>& data,
      bool public_only = false,
      const std::optional<ccf::TxID>& expected_txid = std::nullopt) = 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(
      const ccf::TxID& txid,
      std::unique_ptr<PendingTx> pending_tx,
      bool globally_committable) = 0;
    virtual bool check_rollback_count(Version count) = 0;
 
    virtual std::unique_ptr<AbstractSnapshot> snapshot_unsafe_maps(
      Version v) = 0;
    virtual void lock_maps() = 0;
    virtual void unlock_maps() = 0;
    virtual std::vector<uint8_t> serialise_snapshot(
      std::unique_ptr<AbstractSnapshot> snapshot) = 0;
    virtual ApplyResult deserialise_snapshot(
      const uint8_t* data,
      size_t size,
      ConsensusHookPtrs& hooks,
      std::vector<Version>* view_history = nullptr,
      bool public_only = false) = 0;
    virtual bool should_create_ledger_chunk(Version version) = 0;
    virtual bool should_create_ledger_chunk_unsafe(Version version) = 0;
 
    virtual size_t committable_gap() = 0;
 
    enum class StoreFlag : uint8_t
    {
      SNAPSHOT_AT_NEXT_SIGNATURE = 0x02
    };
 
    virtual void set_flag(StoreFlag f) = 0;
    virtual void unset_flag(StoreFlag f) = 0;
    virtual bool flag_enabled(StoreFlag f) = 0;
    virtual void set_flag_unsafe(StoreFlag f) = 0;
    virtual void unset_flag_unsafe(StoreFlag f) = 0;
    [[nodiscard]] virtual bool flag_enabled_unsafe(StoreFlag f) const = 0;
  };
 
  template <class StorePointer>
  class ScopedStoreMapsLock
  {
  public:
    ScopedStoreMapsLock() = delete;
    ScopedStoreMapsLock(StorePointer _store) : store(std::move(_store))
    {
      store->lock_maps();
    }
    ~ScopedStoreMapsLock()
    {
      store->unlock_maps();
    }
 
  private:
    StorePointer store;
  };
 
}
 
FMT_BEGIN_NAMESPACE
 
template <>
struct formatter<ccf::kv::Configuration::Nodes>
{
  template <typename ParseContext>
  constexpr auto parse(ParseContext& ctx)
  {
    return ctx.begin();
  }
 
  template <typename FormatContext>
  auto format(const ccf::kv::Configuration::Nodes& nodes, FormatContext& ctx)
    const -> decltype(ctx.out())
  {
    std::set<ccf::NodeId> node_ids;
    for (const auto& [nid, _] : nodes)
    {
      node_ids.insert(nid);
    }
    return fmt::format_to(ctx.out(), "{{{}}}", fmt::join(node_ids, " "));
  }
};
 
template <>
struct formatter<ccf::kv::MembershipState>
{
  template <typename ParseContext>
  constexpr auto parse(ParseContext& ctx)
  {
    return ctx.begin();
  }
 
  template <typename FormatContext>
  auto format(const ccf::kv::MembershipState& state, FormatContext& ctx) const
    -> decltype(ctx.out())
  {
    const auto s = nlohmann::json(state).get<std::string>();
    return format_to(ctx.out(), "{}", s);
  }
};
 
template <>
struct formatter<ccf::kv::LeadershipState>
{
  template <typename ParseContext>
  constexpr auto parse(ParseContext& ctx)
  {
    return ctx.begin();
  }
 
  template <typename FormatContext>
  auto format(const ccf::kv::LeadershipState& state, FormatContext& ctx) const
    -> decltype(ctx.out())
  {
    const auto s = nlohmann::json(state).get<std::string>();
    return format_to(ctx.out(), "{}", s);
  }
};
 
FMT_END_NAMESPACE