RCLConsensus.h

#pragma once
 
#include <xrpld/app/consensus/RCLCxLedger.h>
#include <xrpld/app/consensus/RCLCxPeerPos.h>
#include <xrpld/app/consensus/RCLCxTx.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/app/misc/FeeVote.h>
#include <xrpld/app/misc/NegativeUNLVote.h>
 
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/consensus/CensorshipDetector.h>
#include <xrpl/consensus/Consensus.h>
#include <xrpl/consensus/ConsensusParms.h>
#include <xrpl/consensus/ConsensusTypes.h>
#include <xrpl/json/json_value.h>
#include <xrpl/ledger/CanonicalTXSet.h>
#include <xrpl/protocol/Protocol.h>
#include <xrpl/protocol/PublicKey.h>
#include <xrpl/protocol/RippleLedgerHash.h>
#include <xrpl/protocol/UintTypes.h>
 
#include <xrpl.pb.h>
 
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <mutex>
#include <optional>
#include <set>
#include <sstream>
#include <string>
#include <utility>
 
namespace xrpl {
 
class InboundTransactions;
class LocalTxs;
class LedgerMaster;
class ValidatorKeys;

class RCLConsensus
{
    static constexpr unsigned int kCensorshipWarnInternal = 15;
 
    // Implements the Adaptor template interface required by Consensus.
    class Adaptor
    {
        Application& app_;
        std::unique_ptr<FeeVote> feeVote_;
        LedgerMaster& ledgerMaster_;
        LocalTxs& localTxs_;
        InboundTransactions& inboundTransactions_;
        beast::Journal const j_;
 
        // If the server is validating, the necessary keying information:
        ValidatorKeys const& validatorKeys_;
 
        // A randomly selected non-zero value used to tag our validations
        std::uint64_t const valCookie_;
 
        // Ledger we most recently needed to acquire
        LedgerHash acquiringLedger_;
        ConsensusParms parms_;
 
        // The timestamp of the last validation we used
        NetClock::time_point lastValidationTime_;
 
        // These members are queried via public accessors and are atomic for
        // thread safety.
        std::atomic<bool> validating_{false};
        std::atomic<std::size_t> prevProposers_{0};
        std::atomic<std::chrono::milliseconds> prevRoundTime_{std::chrono::milliseconds{0}};
        std::atomic<ConsensusMode> mode_{ConsensusMode::Observing};
 
        CensorshipDetector<TxID, LedgerIndex> censorshipDetector_;
        NegativeUNLVote nUnlVote_;
 
    public:
        using Ledger_t = RCLCxLedger;
        using NodeID_t = NodeID;
        using NodeKey_t = PublicKey;
        using TxSet_t = RCLTxSet;
        using PeerPosition_t = RCLCxPeerPos;
 
        using Result = ConsensusResult<Adaptor>;
 
        Adaptor(
            Application& app,
            std::unique_ptr<FeeVote>&& feeVote,
            LedgerMaster& ledgerMaster,
            LocalTxs& localTxs,
            InboundTransactions& inboundTransactions,
            ValidatorKeys const& validatorKeys,
            beast::Journal journal);
 
        bool
        validating() const
        {
            return validating_;
        }
 
        std::size_t
        prevProposers() const
        {
            return prevProposers_;
        }
 
        std::chrono::milliseconds
        prevRoundTime() const
        {
            return prevRoundTime_;
        }
 
        ConsensusMode
        mode() const
        {
            return mode_;
        }

        bool
        preStartRound(RCLCxLedger const& prevLedger, hash_set<NodeID> const& nowTrusted);
 
        bool
        haveValidated() const;
 
        LedgerIndex
        getValidLedgerIndex() const;
 
        std::pair<std::size_t, hash_set<NodeKey_t>>
        getQuorumKeys() const;
 
        std::size_t
        laggards(Ledger_t::Seq const seq, hash_set<NodeKey_t>& trustedKeys) const;

        bool
        validator() const;

        void
        updateOperatingMode(std::size_t const positions) const;

        ConsensusParms const&
        parms() const
        {
            return parms_;
        }
 
    private:
        //---------------------------------------------------------------------
        // The following members implement the generic Consensus requirements
        // and are marked private to indicate ONLY Consensus<Adaptor> will call
        // them (via friendship). Since they are called only from
        // Consensus<Adaptor> methods and since RCLConsensus::consensus_ should
        // only be accessed under lock, these will only be called under lock.
        //
        // In general, the idea is that there is only ONE thread that is running
        // consensus code at anytime. The only special case is the dispatched
        // onAccept call, which does not take a lock and relies on Consensus not
        // changing state until a future call to startRound.
        friend class Consensus<Adaptor>;

        std::optional<RCLCxLedger>
        acquireLedger(LedgerHash const& hash);

        void
        share(RCLCxPeerPos const& peerPos);

        void
        share(RCLCxTx const& tx);

        std::optional<RCLTxSet>
        acquireTxSet(RCLTxSet::ID const& setId);

        bool
        hasOpenTransactions() const;

        std::size_t
        proposersValidated(LedgerHash const& h) const;

        std::size_t
        proposersFinished(RCLCxLedger const& ledger, LedgerHash const& h) const;

        void
        propose(RCLCxPeerPos::Proposal const& proposal);

        void
        share(RCLTxSet const& txns);

        uint256
        getPrevLedger(uint256 ledgerID, RCLCxLedger const& ledger, ConsensusMode mode);

        void
        onModeChange(ConsensusMode before, ConsensusMode after);

        Result
        onClose(
            RCLCxLedger const& ledger,
            NetClock::time_point const& closeTime,
            ConsensusMode mode);

        void
        onAccept(
            Result const& result,
            RCLCxLedger const& prevLedger,
            NetClock::duration const& closeResolution,
            ConsensusCloseTimes const& rawCloseTimes,
            ConsensusMode const& mode,
            json::Value&& consensusJson,
            bool const validating);

        void
        onForceAccept(
            Result const& result,
            RCLCxLedger const& prevLedger,
            NetClock::duration const& closeResolution,
            ConsensusCloseTimes const& rawCloseTimes,
            ConsensusMode const& mode,
            json::Value&& consensusJson);

        void
        notify(protocol::NodeEvent ne, RCLCxLedger const& ledger, bool haveCorrectLCL);

        void
        doAccept(
            Result const& result,
            RCLCxLedger const& prevLedger,
            NetClock::duration closeResolution,
            ConsensusCloseTimes const& rawCloseTimes,
            ConsensusMode const& mode,
            json::Value&& consensusJson);

        RCLCxLedger
        buildLCL(
            RCLCxLedger const& previousLedger,
            CanonicalTXSet& retriableTxs,
            NetClock::time_point closeTime,
            bool closeTimeCorrect,
            NetClock::duration closeResolution,
            std::chrono::milliseconds roundTime,
            std::set<TxID>& failedTxs);

        void
        validate(RCLCxLedger const& ledger, RCLTxSet const& txns, bool proposing);
    };
 
public:
    RCLConsensus(
        Application& app,
        std::unique_ptr<FeeVote>&& feeVote,
        LedgerMaster& ledgerMaster,
        LocalTxs& localTxs,
        InboundTransactions& inboundTransactions,
        Consensus<Adaptor>::clock_type const& clock,
        ValidatorKeys const& validatorKeys,
        beast::Journal journal);
 
    RCLConsensus(RCLConsensus const&) = delete;
 
    RCLConsensus&
    operator=(RCLConsensus const&) = delete;

    bool
    validating() const
    {
        return adaptor_.validating();
    }

    std::size_t
    prevProposers() const
    {
        return adaptor_.prevProposers();
    }

    std::chrono::milliseconds
    prevRoundTime() const
    {
        return adaptor_.prevRoundTime();
    }

    ConsensusMode
    mode() const
    {
        return adaptor_.mode();
    }
 
    ConsensusPhase
    phase() const
    {
        return consensus_.phase();
    }

    json::Value
    getJson(bool full) const;

    void
    startRound(
        NetClock::time_point const& now,
        RCLCxLedger::ID const& prevLgrId,
        RCLCxLedger const& prevLgr,
        hash_set<NodeID> const& nowUntrusted,
        hash_set<NodeID> const& nowTrusted,
        std::unique_ptr<std::stringstream> const& clog);

    void
    timerEntry(
        NetClock::time_point const& now,
        std::unique_ptr<std::stringstream> const& clog = {});

    void
    gotTxSet(NetClock::time_point const& now, RCLTxSet const& txSet);

    RCLCxLedger::ID
    prevLedgerID() const
    {
        std::scoped_lock const _{mutex_};
        return consensus_.prevLedgerID();
    }

    void
    simulate(
        NetClock::time_point const& now,
        std::optional<std::chrono::milliseconds> consensusDelay);

    bool
    peerProposal(NetClock::time_point const& now, RCLCxPeerPos const& newProposal);
 
    ConsensusParms const&
    parms() const
    {
        return adaptor_.parms();
    }
 
private:
    // Since Consensus does not provide intrinsic thread-safety, this mutex
    // guards all calls to consensus_. adaptor_ uses atomics internally
    // to allow concurrent access of its data members that have getters.
    mutable std::recursive_mutex mutex_;
 
    Adaptor adaptor_;
    Consensus<Adaptor> consensus_;
    beast::Journal const j_;
};

class RclConsensusLogger
{
    std::string header_;
    beast::Journal j_;
    std::unique_ptr<std::stringstream> ss_;
    std::chrono::steady_clock::time_point start_;
 
public:
    explicit RclConsensusLogger(char const* label, bool validating, beast::Journal j);
    ~RclConsensusLogger();
 
    std::unique_ptr<std::stringstream> const&
    ss()
    {
        return ss_;
    }
};
}  // namespace xrpl