Consensus.h

#pragma once
 
#include <xrpld/consensus/ConsensusParms.h>
#include <xrpld/consensus/ConsensusProposal.h>
#include <xrpld/consensus/ConsensusTypes.h>
#include <xrpld/consensus/DisputedTx.h>
#include <xrpld/consensus/LedgerTiming.h>
 
#include <xrpl/basics/Log.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/json/json_writer.h>
 
#include <algorithm>
#include <chrono>
#include <deque>
#include <optional>
#include <sstream>
 
namespace xrpl {
 
bool
shouldCloseLedger(
    bool anyTransactions,
    std::size_t prevProposers,
    std::size_t proposersClosed,
    std::size_t proposersValidated,
    std::chrono::milliseconds prevRoundTime,
    std::chrono::milliseconds timeSincePrevClose,
    std::chrono::milliseconds openTime,
    std::chrono::milliseconds idleInterval,
    ConsensusParms const& parms,
    beast::Journal j,
    std::unique_ptr<std::stringstream> const& clog = {});
 
ConsensusState
checkConsensus(
    std::size_t prevProposers,
    std::size_t currentProposers,
    std::size_t currentAgree,
    std::size_t currentFinished,
    std::chrono::milliseconds previousAgreeTime,
    std::chrono::milliseconds currentAgreeTime,
    bool stalled,
    ConsensusParms const& parms,
    bool proposing,
    beast::Journal j,
    std::unique_ptr<std::stringstream> const& clog = {});
 
template <class Adaptor>
class Consensus
{
    using Ledger_t = typename Adaptor::Ledger_t;
    using TxSet_t = typename Adaptor::TxSet_t;
    using NodeID_t = typename Adaptor::NodeID_t;
    using Tx_t = typename TxSet_t::Tx;
    using PeerPosition_t = typename Adaptor::PeerPosition_t;
    using Proposal_t = ConsensusProposal<NodeID_t, typename Ledger_t::ID, typename TxSet_t::ID>;
 
    using Result = ConsensusResult<Adaptor>;
 
    // Helper class to ensure adaptor is notified whenever the ConsensusMode
    // changes
    class MonitoredMode
    {
        ConsensusMode mode_;
 
    public:
        MonitoredMode(ConsensusMode m) : mode_{m}
        {
        }
        ConsensusMode
        get() const
        {
            return mode_;
        }
 
        void
        set(ConsensusMode mode, Adaptor& a)
        {
            a.onModeChange(mode_, mode);
            mode_ = mode;
        }
    };
 
public:
    using clock_type = beast::abstract_clock<std::chrono::steady_clock>;
 
    Consensus(Consensus&&) noexcept = default;
 
    Consensus(clock_type const& clock, Adaptor& adaptor, beast::Journal j);
 
    void
    startRound(
        NetClock::time_point const& now,
        typename Ledger_t::ID const& prevLedgerID,
        Ledger_t prevLedger,
        hash_set<NodeID_t> const& nowUntrusted,
        bool proposing,
        std::unique_ptr<std::stringstream> const& clog = {});
 
    bool
    peerProposal(NetClock::time_point const& now, PeerPosition_t const& newProposal);
 
    void
    timerEntry(NetClock::time_point const& now, std::unique_ptr<std::stringstream> const& clog = {});
 
    void
    gotTxSet(NetClock::time_point const& now, TxSet_t const& txSet);
 
    void
    simulate(NetClock::time_point const& now, std::optional<std::chrono::milliseconds> consensusDelay);
 
    typename Ledger_t::ID
    prevLedgerID() const
    {
        return prevLedgerID_;
    }
 
    ConsensusPhase
    phase() const
    {
        return phase_;
    }
 
    Json::Value
    getJson(bool full) const;
 
private:
    void
    startRoundInternal(
        NetClock::time_point const& now,
        typename Ledger_t::ID const& prevLedgerID,
        Ledger_t const& prevLedger,
        ConsensusMode mode,
        std::unique_ptr<std::stringstream> const& clog);
 
    // Change our view of the previous ledger
    void
    handleWrongLedger(typename Ledger_t::ID const& lgrId, std::unique_ptr<std::stringstream> const& clog);
 
    void
    checkLedger(std::unique_ptr<std::stringstream> const& clog);
 
    void
    playbackProposals();
 
    bool
    peerProposalInternal(NetClock::time_point const& now, PeerPosition_t const& newProposal);
 
    void
    phaseOpen(std::unique_ptr<std::stringstream> const& clog);
 
    void
    phaseEstablish(std::unique_ptr<std::stringstream> const& clog);
 
    bool
    shouldPause(std::unique_ptr<std::stringstream> const& clog) const;
 
    // Close the open ledger and establish initial position.
    void
    closeLedger(std::unique_ptr<std::stringstream> const& clog);
 
    // Adjust our positions to try to agree with other validators.
    void
    updateOurPositions(std::unique_ptr<std::stringstream> const& clog);
 
    bool
    haveConsensus(std::unique_ptr<std::stringstream> const& clog);
 
    // Create disputes between our position and the provided one.
    void
    createDisputes(TxSet_t const& o, std::unique_ptr<std::stringstream> const& clog = {});
 
    // Update our disputes given that this node has adopted a new position.
    // Will call createDisputes as needed.
    void
    updateDisputes(NodeID_t const& node, TxSet_t const& other);
 
    // Revoke our outstanding proposal, if any, and cease proposing
    // until this round ends.
    void
    leaveConsensus(std::unique_ptr<std::stringstream> const& clog);
 
    // The rounded or effective close time estimate from a proposer
    NetClock::time_point
    asCloseTime(NetClock::time_point raw) const;
 
private:
    Adaptor& adaptor_;
 
    ConsensusPhase phase_{ConsensusPhase::accepted};
    MonitoredMode mode_{ConsensusMode::observing};
    bool firstRound_ = true;
    bool haveCloseTimeConsensus_ = false;
 
    clock_type const& clock_;
 
    // How long the consensus convergence has taken, expressed as
    // a percentage of the time that we expected it to take.
    int convergePercent_{0};
 
    // How long has this round been open
    ConsensusTimer openTime_;
 
    NetClock::duration closeResolution_ = ledgerDefaultTimeResolution;
 
    ConsensusParms::AvalancheState closeTimeAvalancheState_ = ConsensusParms::init;
 
    // Time it took for the last consensus round to converge
    std::chrono::milliseconds prevRoundTime_;
 
    //-------------------------------------------------------------------------
    // Network time measurements of consensus progress
 
    // The current network adjusted time.  This is the network time the
    // ledger would close if it closed now
    NetClock::time_point now_;
    NetClock::time_point prevCloseTime_;
 
    //-------------------------------------------------------------------------
    // Non-peer (self) consensus data
 
    // Last validated ledger ID provided to consensus
    typename Ledger_t::ID prevLedgerID_;
    // Last validated ledger seen by consensus
    Ledger_t previousLedger_;
 
    // Transaction Sets, indexed by hash of transaction tree
    hash_map<typename TxSet_t::ID, TxSet_t const> acquired_;
 
    std::optional<Result> result_;
    ConsensusCloseTimes rawCloseTimes_;
 
    // The number of calls to phaseEstablish where none of our peers
    // have changed any votes on disputed transactions.
    std::size_t peerUnchangedCounter_ = 0;
 
    // The total number of times we have called phaseEstablish
    std::size_t establishCounter_ = 0;
 
    //-------------------------------------------------------------------------
    // Peer related consensus data
 
    // Peer proposed positions for the current round
    hash_map<NodeID_t, PeerPosition_t> currPeerPositions_;
 
    // Recently received peer positions, available when transitioning between
    // ledgers or rounds
    hash_map<NodeID_t, std::deque<PeerPosition_t>> recentPeerPositions_;
 
    // The number of proposers who participated in the last consensus round
    std::size_t prevProposers_ = 0;
 
    // nodes that have bowed out of this consensus process
    hash_set<NodeID_t> deadNodes_;
 
    // Journal for debugging
    beast::Journal const j_;
};
 
template <class Adaptor>
Consensus<Adaptor>::Consensus(clock_type const& clock, Adaptor& adaptor, beast::Journal journal)
    : adaptor_(adaptor), clock_(clock), j_{journal}
{
    JLOG(j_.debug()) << "Creating consensus object";
}
 
template <class Adaptor>
void
Consensus<Adaptor>::startRound(
    NetClock::time_point const& now,
    typename Ledger_t::ID const& prevLedgerID,
    Ledger_t prevLedger,
    hash_set<NodeID_t> const& nowUntrusted,
    bool proposing,
    std::unique_ptr<std::stringstream> const& clog)
{
    if (firstRound_)
    {
        // take our initial view of closeTime_ from the seed ledger
        prevRoundTime_ = adaptor_.parms().ledgerIDLE_INTERVAL;
        prevCloseTime_ = prevLedger.closeTime();
        firstRound_ = false;
    }
    else
    {
        prevCloseTime_ = rawCloseTimes_.self;
    }
 
    for (NodeID_t const& n : nowUntrusted)
        recentPeerPositions_.erase(n);
 
    ConsensusMode startMode = proposing ? ConsensusMode::proposing : ConsensusMode::observing;
 
    // We were handed the wrong ledger
    if (prevLedger.id() != prevLedgerID)
    {
        // try to acquire the correct one
        if (auto newLedger = adaptor_.acquireLedger(prevLedgerID))
        {
            prevLedger = *newLedger;
        }
        else  // Unable to acquire the correct ledger
        {
            startMode = ConsensusMode::wrongLedger;
            JLOG(j_.info()) << "Entering consensus with: " << previousLedger_.id();
            JLOG(j_.info()) << "Correct LCL is: " << prevLedgerID;
        }
    }
 
    startRoundInternal(now, prevLedgerID, prevLedger, startMode, clog);
}
template <class Adaptor>
void
Consensus<Adaptor>::startRoundInternal(
    NetClock::time_point const& now,
    typename Ledger_t::ID const& prevLedgerID,
    Ledger_t const& prevLedger,
    ConsensusMode mode,
    std::unique_ptr<std::stringstream> const& clog)
{
    phase_ = ConsensusPhase::open;
    JLOG(j_.debug()) << "transitioned to ConsensusPhase::open ";
    CLOG(clog) << "startRoundInternal transitioned to ConsensusPhase::open, "
                  "previous ledgerID: "
               << prevLedgerID << ", seq: " << prevLedger.seq() << ". ";
    mode_.set(mode, adaptor_);
    now_ = now;
    prevLedgerID_ = prevLedgerID;
    previousLedger_ = prevLedger;
    result_.reset();
    convergePercent_ = 0;
    closeTimeAvalancheState_ = ConsensusParms::init;
    haveCloseTimeConsensus_ = false;
    openTime_.reset(clock_.now());
    currPeerPositions_.clear();
    acquired_.clear();
    rawCloseTimes_.peers.clear();
    rawCloseTimes_.self = {};
    deadNodes_.clear();
 
    closeResolution_ = getNextLedgerTimeResolution(
        previousLedger_.closeTimeResolution(),
        previousLedger_.closeAgree(),
        previousLedger_.seq() + typename Ledger_t::Seq{1});
 
    playbackProposals();
    CLOG(clog) << "number of peer proposals,previous proposers: " << currPeerPositions_.size() << ',' << prevProposers_
               << ". ";
    if (currPeerPositions_.size() > (prevProposers_ / 2))
    {
        // We may be falling behind, don't wait for the timer
        // consider closing the ledger immediately
        CLOG(clog) << "consider closing the ledger immediately. ";
        timerEntry(now_, clog);
    }
}
 
template <class Adaptor>
bool
Consensus<Adaptor>::peerProposal(NetClock::time_point const& now, PeerPosition_t const& newPeerPos)
{
    JLOG(j_.debug()) << "PROPOSAL " << newPeerPos.render();
    auto const& peerID = newPeerPos.proposal().nodeID();
 
    // Always need to store recent positions
    {
        auto& props = recentPeerPositions_[peerID];
 
        if (props.size() >= 10)
            props.pop_front();
 
        props.push_back(newPeerPos);
    }
    return peerProposalInternal(now, newPeerPos);
}
 
template <class Adaptor>
bool
Consensus<Adaptor>::peerProposalInternal(NetClock::time_point const& now, PeerPosition_t const& newPeerPos)
{
    // Nothing to do for now if we are currently working on a ledger
    if (phase_ == ConsensusPhase::accepted)
        return false;
 
    now_ = now;
 
    auto const& newPeerProp = newPeerPos.proposal();
 
    if (newPeerProp.prevLedger() != prevLedgerID_)
    {
        JLOG(j_.debug()) << "Got proposal for " << newPeerProp.prevLedger() << " but we are on " << prevLedgerID_;
        return false;
    }
 
    auto const& peerID = newPeerProp.nodeID();
 
    if (deadNodes_.find(peerID) != deadNodes_.end())
    {
        JLOG(j_.info()) << "Position from dead node: " << peerID;
        return false;
    }
 
    {
        // update current position
        auto peerPosIt = currPeerPositions_.find(peerID);
 
        if (peerPosIt != currPeerPositions_.end())
        {
            if (newPeerProp.proposeSeq() <= peerPosIt->second.proposal().proposeSeq())
            {
                return false;
            }
        }
 
        if (newPeerProp.isBowOut())
        {
            JLOG(j_.info()) << "Peer " << peerID << " bows out";
            if (result_)
            {
                for (auto& it : result_->disputes)
                    it.second.unVote(peerID);
            }
            if (peerPosIt != currPeerPositions_.end())
                currPeerPositions_.erase(peerID);
            deadNodes_.insert(peerID);
 
            return true;
        }
 
        if (peerPosIt != currPeerPositions_.end())
            peerPosIt->second = newPeerPos;
        else
            currPeerPositions_.emplace(peerID, newPeerPos);
    }
 
    if (newPeerProp.isInitial())
    {
        // Record the close time estimate
        JLOG(j_.trace()) << "Peer reports close time as " << newPeerProp.closeTime().time_since_epoch().count();
        ++rawCloseTimes_.peers[newPeerProp.closeTime()];
    }
 
    JLOG(j_.trace()) << "Processing peer proposal " << newPeerProp.proposeSeq() << "/" << newPeerProp.position();
 
    {
        auto const ait = acquired_.find(newPeerProp.position());
        if (ait == acquired_.end())
        {
            // acquireTxSet will return the set if it is available, or
            // spawn a request for it and return nullopt/nullptr.  It will call
            // gotTxSet once it arrives
            if (auto set = adaptor_.acquireTxSet(newPeerProp.position()))
                gotTxSet(now_, *set);
            else
                JLOG(j_.debug()) << "Don't have tx set for peer";
        }
        else if (result_)
        {
            updateDisputes(newPeerProp.nodeID(), ait->second);
        }
    }
 
    return true;
}
 
template <class Adaptor>
void
Consensus<Adaptor>::timerEntry(NetClock::time_point const& now, std::unique_ptr<std::stringstream> const& clog)
{
    CLOG(clog) << "Consensus<Adaptor>::timerEntry. ";
    // Nothing to do if we are currently working on a ledger
    if (phase_ == ConsensusPhase::accepted)
    {
        CLOG(clog) << "Nothing to do during accepted phase. ";
        return;
    }
 
    now_ = now;
    CLOG(clog) << "Set network adjusted time to " << to_string(now) << ". ";
 
    // Check we are on the proper ledger (this may change phase_)
    auto const phaseOrig = phase_;
    CLOG(clog) << "Phase " << to_string(phaseOrig) << ". ";
    checkLedger(clog);
    if (phaseOrig != phase_)
    {
        CLOG(clog) << "Changed phase to << " << to_string(phase_) << ". ";
    }
 
    if (phase_ == ConsensusPhase::open)
        phaseOpen(clog);
    else if (phase_ == ConsensusPhase::establish)
        phaseEstablish(clog);
    CLOG(clog) << "timerEntry finishing in phase " << to_string(phase_) << ". ";
}
 
template <class Adaptor>
void
Consensus<Adaptor>::gotTxSet(NetClock::time_point const& now, TxSet_t const& txSet)
{
    // Nothing to do if we've finished work on a ledger
    if (phase_ == ConsensusPhase::accepted)
        return;
 
    now_ = now;
 
    auto id = txSet.id();
 
    // If we've already processed this transaction set since requesting
    // it from the network, there is nothing to do now
    if (!acquired_.emplace(id, txSet).second)
        return;
 
    if (!result_)
    {
        JLOG(j_.debug()) << "Not creating disputes: no position yet.";
    }
    else
    {
        // Our position is added to acquired_ as soon as we create it,
        // so this txSet must differ
        XRPL_ASSERT(id != result_->position.position(), "xrpl::Consensus::gotTxSet : updated transaction set");
        bool any = false;
        for (auto const& [nodeId, peerPos] : currPeerPositions_)
        {
            if (peerPos.proposal().position() == id)
            {
                updateDisputes(nodeId, txSet);
                any = true;
            }
        }
 
        if (!any)
        {
            JLOG(j_.warn()) << "By the time we got " << id << " no peers were proposing it";
        }
    }
}
 
template <class Adaptor>
void
Consensus<Adaptor>::simulate(NetClock::time_point const& now, std::optional<std::chrono::milliseconds> consensusDelay)
{
    using namespace std::chrono_literals;
    JLOG(j_.info()) << "Simulating consensus";
    now_ = now;
    closeLedger({});
    result_->roundTime.tick(consensusDelay.value_or(100ms));
    result_->proposers = prevProposers_ = currPeerPositions_.size();
    prevRoundTime_ = result_->roundTime.read();
    phase_ = ConsensusPhase::accepted;
    adaptor_.onForceAccept(*result_, previousLedger_, closeResolution_, rawCloseTimes_, mode_.get(), getJson(true));
    JLOG(j_.info()) << "Simulation complete";
}
 
template <class Adaptor>
Json::Value
Consensus<Adaptor>::getJson(bool full) const
{
    using std::to_string;
    using Int = Json::Value::Int;
 
    Json::Value ret(Json::objectValue);
 
    ret["proposing"] = (mode_.get() == ConsensusMode::proposing);
    ret["proposers"] = static_cast<int>(currPeerPositions_.size());
 
    if (mode_.get() != ConsensusMode::wrongLedger)
    {
        ret["synched"] = true;
        ret["ledger_seq"] = static_cast<std::uint32_t>(previousLedger_.seq()) + 1;
        ret["close_granularity"] = static_cast<Int>(closeResolution_.count());
    }
    else
        ret["synched"] = false;
 
    ret["phase"] = to_string(phase_);
 
    if (result_ && !result_->disputes.empty() && !full)
        ret["disputes"] = static_cast<Int>(result_->disputes.size());
 
    if (result_)
        ret["our_position"] = result_->position.getJson();
 
    if (full)
    {
        if (result_)
            ret["current_ms"] = static_cast<Int>(result_->roundTime.read().count());
        ret["converge_percent"] = convergePercent_;
        ret["close_resolution"] = static_cast<Int>(closeResolution_.count());
        ret["have_time_consensus"] = haveCloseTimeConsensus_;
        ret["previous_proposers"] = static_cast<Int>(prevProposers_);
        ret["previous_mseconds"] = static_cast<Int>(prevRoundTime_.count());
 
        if (!currPeerPositions_.empty())
        {
            Json::Value ppj(Json::objectValue);
 
            for (auto const& [nodeId, peerPos] : currPeerPositions_)
            {
                ppj[to_string(nodeId)] = peerPos.getJson();
            }
            ret["peer_positions"] = std::move(ppj);
        }
 
        if (!acquired_.empty())
        {
            Json::Value acq(Json::arrayValue);
            for (auto const& at : acquired_)
            {
                acq.append(to_string(at.first));
            }
            ret["acquired"] = std::move(acq);
        }
 
        if (result_ && !result_->disputes.empty())
        {
            Json::Value dsj(Json::objectValue);
            for (auto const& [txId, dispute] : result_->disputes)
            {
                dsj[to_string(txId)] = dispute.getJson();
            }
            ret["disputes"] = std::move(dsj);
        }
 
        if (!rawCloseTimes_.peers.empty())
        {
            Json::Value ctj(Json::objectValue);
            for (auto const& ct : rawCloseTimes_.peers)
            {
                ctj[std::to_string(ct.first.time_since_epoch().count())] = ct.second;
            }
            ret["close_times"] = std::move(ctj);
        }
 
        if (!deadNodes_.empty())
        {
            Json::Value dnj(Json::arrayValue);
            for (auto const& dn : deadNodes_)
            {
                dnj.append(to_string(dn));
            }
            ret["dead_nodes"] = std::move(dnj);
        }
    }
 
    return ret;
}
 
// Handle a change in the prior ledger during a consensus round
template <class Adaptor>
void
Consensus<Adaptor>::handleWrongLedger(
    typename Ledger_t::ID const& lgrId,
    std::unique_ptr<std::stringstream> const& clog)
{
    CLOG(clog) << "handleWrongLedger. ";
    XRPL_ASSERT(
        lgrId != prevLedgerID_ || previousLedger_.id() != lgrId,
        "xrpl::Consensus::handleWrongLedger : have wrong ledger");
 
    // Stop proposing because we are out of sync
    leaveConsensus(clog);
 
    // First time switching to this ledger
    if (prevLedgerID_ != lgrId)
    {
        prevLedgerID_ = lgrId;
 
        // Clear out state
        if (result_)
        {
            result_->disputes.clear();
            result_->compares.clear();
        }
 
        currPeerPositions_.clear();
        rawCloseTimes_.peers.clear();
        deadNodes_.clear();
 
        // Get back in sync, this will also recreate disputes
        playbackProposals();
    }
 
    if (previousLedger_.id() == prevLedgerID_)
    {
        CLOG(clog) << "previousLedger_.id() == prevLeverID_ " << prevLedgerID_ << ". ";
        return;
    }
 
    // we need to switch the ledger we're working from
    if (auto newLedger = adaptor_.acquireLedger(prevLedgerID_))
    {
        JLOG(j_.info()) << "Have the consensus ledger " << prevLedgerID_;
        CLOG(clog) << "Have the consensus ledger " << prevLedgerID_ << ". ";
        startRoundInternal(now_, lgrId, *newLedger, ConsensusMode::switchedLedger, clog);
    }
    else
    {
        CLOG(clog) << "Still on wrong ledger. ";
        mode_.set(ConsensusMode::wrongLedger, adaptor_);
    }
}
 
template <class Adaptor>
void
Consensus<Adaptor>::checkLedger(std::unique_ptr<std::stringstream> const& clog)
{
    CLOG(clog) << "checkLedger. ";
 
    auto netLgr = adaptor_.getPrevLedger(prevLedgerID_, previousLedger_, mode_.get());
    CLOG(clog) << "network ledgerid " << netLgr << ",  "
               << "previous ledger " << prevLedgerID_ << ". ";
 
    if (netLgr != prevLedgerID_)
    {
        std::stringstream ss;
        ss << "View of consensus changed during " << to_string(phase_) << " mode=" << to_string(mode_.get()) << ", "
           << prevLedgerID_ << " to " << netLgr << ", " << Json::Compact{previousLedger_.getJson()} << ". ";
        JLOG(j_.warn()) << ss.str();
        CLOG(clog) << ss.str();
        CLOG(clog) << "State on consensus change " << Json::Compact{getJson(true)} << ". ";
        handleWrongLedger(netLgr, clog);
    }
    else if (previousLedger_.id() != prevLedgerID_)
    {
        CLOG(clog) << "previousLedger_.id() != prevLedgerID_: " << previousLedger_.id() << ','
                   << to_string(prevLedgerID_) << ". ";
        handleWrongLedger(netLgr, clog);
    }
}
 
template <class Adaptor>
void
Consensus<Adaptor>::playbackProposals()
{
    for (auto const& it : recentPeerPositions_)
    {
        for (auto const& pos : it.second)
        {
            if (pos.proposal().prevLedger() == prevLedgerID_)
            {
                if (peerProposalInternal(now_, pos))
                    adaptor_.share(pos);
            }
        }
    }
}
 
template <class Adaptor>
void
Consensus<Adaptor>::phaseOpen(std::unique_ptr<std::stringstream> const& clog)
{
    CLOG(clog) << "phaseOpen. ";
    using namespace std::chrono;
 
    // it is shortly before ledger close time
    bool anyTransactions = adaptor_.hasOpenTransactions();
    auto proposersClosed = currPeerPositions_.size();
    auto proposersValidated = adaptor_.proposersValidated(prevLedgerID_);
 
    openTime_.tick(clock_.now());
 
    // This computes how long since last ledger's close time
    milliseconds sinceClose;
    {
        auto const mode = mode_.get();
        bool const closeAgree = previousLedger_.closeAgree();
        auto const prevCloseTime = previousLedger_.closeTime();
        auto const prevParentCloseTimePlus1 = previousLedger_.parentCloseTime() + 1s;
        bool const previousCloseCorrect =
            (mode != ConsensusMode::wrongLedger) && closeAgree && (prevCloseTime != prevParentCloseTimePlus1);
 
        auto const lastCloseTime = previousCloseCorrect ? prevCloseTime    // use consensus timing
                                                        : prevCloseTime_;  // use the time we saw internally
 
        if (now_ >= lastCloseTime)
            sinceClose = duration_cast<milliseconds>(now_ - lastCloseTime);
        else
            sinceClose = -duration_cast<milliseconds>(lastCloseTime - now_);
        CLOG(clog) << "calculating how long since last ledger's close time "
                      "based on mode : "
                   << to_string(mode) << ", previous closeAgree: " << closeAgree
                   << ", previous close time: " << to_string(prevCloseTime)
                   << ", previous parent close time + 1s: " << to_string(prevParentCloseTimePlus1)
                   << ", previous close time seen internally: " << to_string(prevCloseTime_)
                   << ", last close time: " << to_string(lastCloseTime) << ", since close: " << sinceClose.count()
                   << ". ";
    }
 
    auto const idleInterval =
        std::max<milliseconds>(adaptor_.parms().ledgerIDLE_INTERVAL, 2 * previousLedger_.closeTimeResolution());
    CLOG(clog) << "idle interval set to " << idleInterval.count() << "ms based on "
               << "ledgerIDLE_INTERVAL: " << adaptor_.parms().ledgerIDLE_INTERVAL.count()
               << ", previous ledger close time resolution: " << previousLedger_.closeTimeResolution().count()
               << "ms. ";
 
    // Decide if we should close the ledger
    if (shouldCloseLedger(
            anyTransactions,
            prevProposers_,
            proposersClosed,
            proposersValidated,
            prevRoundTime_,
            sinceClose,
            openTime_.read(),
            idleInterval,
            adaptor_.parms(),
            j_,
            clog))
    {
        CLOG(clog) << "closing ledger. ";
        closeLedger(clog);
    }
}
 
template <class Adaptor>
bool
Consensus<Adaptor>::shouldPause(std::unique_ptr<std::stringstream> const& clog) const
{
    CLOG(clog) << "shouldPause? ";
    auto const& parms = adaptor_.parms();
    std::uint32_t const ahead(previousLedger_.seq() - std::min(adaptor_.getValidLedgerIndex(), previousLedger_.seq()));
    auto [quorum, trustedKeys] = adaptor_.getQuorumKeys();
    std::size_t const totalValidators = trustedKeys.size();
    std::size_t laggards = adaptor_.laggards(previousLedger_.seq(), trustedKeys);
    std::size_t const offline = trustedKeys.size();
 
    std::stringstream vars;
    vars << " consensuslog (working seq: " << previousLedger_.seq() << ", "
         << "validated seq: " << adaptor_.getValidLedgerIndex() << ", "
         << "am validator: " << adaptor_.validator() << ", "
         << "have validated: " << adaptor_.haveValidated() << ", "
         << "roundTime: " << result_->roundTime.read().count() << ", "
         << "max consensus time: " << parms.ledgerMAX_CONSENSUS.count() << ", "
         << "validators: " << totalValidators << ", "
         << "laggards: " << laggards << ", "
         << "offline: " << offline << ", "
         << "quorum: " << quorum << ")";
 
    if (!ahead || !laggards || !totalValidators || !adaptor_.validator() || !adaptor_.haveValidated() ||
        result_->roundTime.read() > parms.ledgerMAX_CONSENSUS)
    {
        j_.debug() << "not pausing (early)" << vars.str();
        CLOG(clog) << "Not pausing (early). ";
        return false;
    }
 
    bool willPause = false;
 
    constexpr static std::size_t maxPausePhase = 4;
 
    std::size_t const phase = (ahead - 1) % (maxPausePhase + 1);
 
    // validators that remain after the laggards() function are considered
    // offline, and should be considered as laggards for purposes of
    // evaluating whether the threshold for non-laggards has been reached.
    switch (phase)
    {
        case 0:
            // Laggards and offline shouldn't preclude consensus.
            if (laggards + offline > totalValidators - quorum)
                willPause = true;
            break;
        case maxPausePhase:
            // No tolerance.
            willPause = true;
            break;
        default:
            // Ensure that sufficient validators are known to be not lagging.
            // Their sufficiently most recent validation sequence was equal to
            // or greater than our own.
            //
            // The threshold is the amount required for quorum plus
            // the proportion of the remainder based on number of intermediate
            // phases between 0 and max.
            float const nonLaggards = totalValidators - (laggards + offline);
            float const quorumRatio = static_cast<float>(quorum) / totalValidators;
            float const allowedDissent = 1.0f - quorumRatio;
            float const phaseFactor = static_cast<float>(phase) / maxPausePhase;
 
            if (nonLaggards / totalValidators < quorumRatio + (allowedDissent * phaseFactor))
            {
                willPause = true;
            }
    }
 
    if (willPause)
    {
        j_.warn() << "pausing" << vars.str();
        CLOG(clog) << "pausing " << vars.str() << ". ";
    }
    else
    {
        j_.debug() << "not pausing" << vars.str();
        CLOG(clog) << "not pausing. ";
    }
    return willPause;
}
 
template <class Adaptor>
void
Consensus<Adaptor>::phaseEstablish(std::unique_ptr<std::stringstream> const& clog)
{
    CLOG(clog) << "phaseEstablish. ";
    // can only establish consensus if we already took a stance
    XRPL_ASSERT(result_, "xrpl::Consensus::phaseEstablish : result is set");
 
    ++peerUnchangedCounter_;
    ++establishCounter_;
 
    using namespace std::chrono;
    ConsensusParms const& parms = adaptor_.parms();
 
    result_->roundTime.tick(clock_.now());
    result_->proposers = currPeerPositions_.size();
 
    convergePercent_ =
        result_->roundTime.read() * 100 / std::max<milliseconds>(prevRoundTime_, parms.avMIN_CONSENSUS_TIME);
    CLOG(clog) << "convergePercent_ " << convergePercent_
               << " is based on round duration so far: " << result_->roundTime.read().count() << "ms, "
               << "previous round duration: " << prevRoundTime_.count() << "ms, "
               << "avMIN_CONSENSUS_TIME: " << parms.avMIN_CONSENSUS_TIME.count() << "ms. ";
 
    // Give everyone a chance to take an initial position
    if (result_->roundTime.read() < parms.ledgerMIN_CONSENSUS)
    {
        CLOG(clog) << "ledgerMIN_CONSENSUS not reached: " << parms.ledgerMIN_CONSENSUS.count() << "ms. ";
        return;
    }
 
    updateOurPositions(clog);
 
    // Nothing to do if too many laggards or we don't have consensus.
    if (shouldPause(clog) || !haveConsensus(clog))
        return;
 
    if (!haveCloseTimeConsensus_)
    {
        JLOG(j_.info()) << "We have TX consensus but not CT consensus";
        CLOG(clog) << "We have TX consensus but not CT consensus. ";
        return;
    }
 
    JLOG(j_.info()) << "Converge cutoff (" << currPeerPositions_.size() << " participants)";
    CLOG(clog) << "Converge cutoff (" << currPeerPositions_.size()
               << " participants). Transitioned to ConsensusPhase::accepted. ";
    adaptor_.updateOperatingMode(currPeerPositions_.size());
    prevProposers_ = currPeerPositions_.size();
    prevRoundTime_ = result_->roundTime.read();
    phase_ = ConsensusPhase::accepted;
    JLOG(j_.debug()) << "transitioned to ConsensusPhase::accepted";
    adaptor_.onAccept(
        *result_, previousLedger_, closeResolution_, rawCloseTimes_, mode_.get(), getJson(true), adaptor_.validating());
}
 
template <class Adaptor>
void
Consensus<Adaptor>::closeLedger(std::unique_ptr<std::stringstream> const& clog)
{
    // We should not be closing if we already have a position
    XRPL_ASSERT(!result_, "xrpl::Consensus::closeLedger : result is not set");
 
    phase_ = ConsensusPhase::establish;
    JLOG(j_.debug()) << "transitioned to ConsensusPhase::establish";
    rawCloseTimes_.self = now_;
    peerUnchangedCounter_ = 0;
    establishCounter_ = 0;
 
    result_.emplace(adaptor_.onClose(previousLedger_, now_, mode_.get()));
    result_->roundTime.reset(clock_.now());
    // Share the newly created transaction set if we haven't already
    // received it from a peer
    if (acquired_.emplace(result_->txns.id(), result_->txns).second)
        adaptor_.share(result_->txns);
 
    auto const mode = mode_.get();
    CLOG(clog) << "closeLedger transitioned to ConsensusPhase::establish, mode: " << to_string(mode)
               << ", number of peer positions: " << currPeerPositions_.size() << ". ";
    if (mode == ConsensusMode::proposing)
        adaptor_.propose(result_->position);
 
    // Create disputes with any peer positions we have transactions for
    for (auto const& pit : currPeerPositions_)
    {
        auto const& pos = pit.second.proposal().position();
        auto const it = acquired_.find(pos);
        if (it != acquired_.end())
            createDisputes(it->second, clog);
    }
}
 
inline int
participantsNeeded(int participants, int percent)
{
    int result = ((participants * percent) + (percent / 2)) / 100;
 
    return (result == 0) ? 1 : result;
}
 
template <class Adaptor>
void
Consensus<Adaptor>::updateOurPositions(std::unique_ptr<std::stringstream> const& clog)
{
    // We must have a position if we are updating it
    XRPL_ASSERT(result_, "xrpl::Consensus::updateOurPositions : result is set");
    ConsensusParms const& parms = adaptor_.parms();
 
    // Compute a cutoff time
    auto const peerCutoff = now_ - parms.proposeFRESHNESS;
    auto const ourCutoff = now_ - parms.proposeINTERVAL;
    CLOG(clog) << "updateOurPositions. peerCutoff " << to_string(peerCutoff) << ", ourCutoff " << to_string(ourCutoff)
               << ". ";
 
    // Verify freshness of peer positions and compute close times
    std::map<NetClock::time_point, int> closeTimeVotes;
    {
        auto it = currPeerPositions_.begin();
        while (it != currPeerPositions_.end())
        {
            Proposal_t const& peerProp = it->second.proposal();
            if (peerProp.isStale(peerCutoff))
            {
                // peer's proposal is stale, so remove it
                NodeID_t const& peerID = peerProp.nodeID();
                JLOG(j_.warn()) << "Removing stale proposal from " << peerID;
                for (auto& dt : result_->disputes)
                    dt.second.unVote(peerID);
                it = currPeerPositions_.erase(it);
            }
            else
            {
                // proposal is still fresh
                ++closeTimeVotes[asCloseTime(peerProp.closeTime())];
                ++it;
            }
        }
    }
 
    // This will stay unseated unless there are any changes
    std::optional<TxSet_t> ourNewSet;
 
    // Update votes on disputed transactions
    {
        std::optional<typename TxSet_t::MutableTxSet> mutableSet;
        for (auto& [txId, dispute] : result_->disputes)
        {
            // Because the threshold for inclusion increases,
            //  time can change our position on a dispute
            if (dispute.updateVote(convergePercent_, mode_.get() == ConsensusMode::proposing, parms))
            {
                if (!mutableSet)
                    mutableSet.emplace(result_->txns);
 
                if (dispute.getOurVote())
                {
                    // now a yes
                    mutableSet->insert(dispute.tx());
                }
                else
                {
                    // now a no
                    mutableSet->erase(txId);
                }
            }
        }
 
        if (mutableSet)
            ourNewSet.emplace(std::move(*mutableSet));
    }
 
    NetClock::time_point consensusCloseTime = {};
    haveCloseTimeConsensus_ = false;
 
    if (currPeerPositions_.empty())
    {
        // no other times
        haveCloseTimeConsensus_ = true;
        consensusCloseTime = asCloseTime(result_->position.closeTime());
    }
    else
    {
        // We don't track rounds for close time, so just pass 0s
        auto const [neededWeight, newState] = getNeededWeight(parms, closeTimeAvalancheState_, convergePercent_, 0, 0);
        if (newState)
            closeTimeAvalancheState_ = *newState;
        CLOG(clog) << "neededWeight " << neededWeight << ". ";
 
        int participants = currPeerPositions_.size();
        if (mode_.get() == ConsensusMode::proposing)
        {
            ++closeTimeVotes[asCloseTime(result_->position.closeTime())];
            ++participants;
        }
 
        // Threshold for non-zero vote
        int threshVote = participantsNeeded(participants, neededWeight);
 
        // Threshold to declare consensus
        int const threshConsensus = participantsNeeded(participants, parms.avCT_CONSENSUS_PCT);
 
        std::stringstream ss;
        ss << "Proposers:" << currPeerPositions_.size() << " nw:" << neededWeight << " thrV:" << threshVote
           << " thrC:" << threshConsensus;
        JLOG(j_.info()) << ss.str();
        CLOG(clog) << ss.str();
 
        for (auto const& [t, v] : closeTimeVotes)
        {
            JLOG(j_.debug()) << "CCTime: seq " << static_cast<std::uint32_t>(previousLedger_.seq()) + 1 << ": "
                             << t.time_since_epoch().count() << " has " << v << ", " << threshVote << " required";
 
            if (v >= threshVote)
            {
                // A close time has enough votes for us to try to agree
                consensusCloseTime = t;
                threshVote = v;
 
                if (threshVote >= threshConsensus)
                    haveCloseTimeConsensus_ = true;
            }
        }
 
        if (!haveCloseTimeConsensus_)
        {
            JLOG(j_.debug()) << "No CT consensus:"
                             << " Proposers:" << currPeerPositions_.size() << " Mode:" << to_string(mode_.get())
                             << " Thresh:" << threshConsensus
                             << " Pos:" << consensusCloseTime.time_since_epoch().count();
            CLOG(clog) << "No close time consensus. ";
        }
    }
 
    if (!ourNewSet &&
        ((consensusCloseTime != asCloseTime(result_->position.closeTime())) || result_->position.isStale(ourCutoff)))
    {
        // close time changed or our position is stale
        ourNewSet.emplace(result_->txns);
    }
 
    if (ourNewSet)
    {
        auto newID = ourNewSet->id();
 
        result_->txns = std::move(*ourNewSet);
 
        std::stringstream ss;
        ss << "Position change: CTime " << consensusCloseTime.time_since_epoch().count() << ", tx " << newID;
        JLOG(j_.info()) << ss.str();
        CLOG(clog) << ss.str();
 
        result_->position.changePosition(newID, consensusCloseTime, now_);
 
        // Share our new transaction set and update disputes
        // if we haven't already received it
        if (acquired_.emplace(newID, result_->txns).second)
        {
            if (!result_->position.isBowOut())
                adaptor_.share(result_->txns);
 
            for (auto const& [nodeId, peerPos] : currPeerPositions_)
            {
                Proposal_t const& p = peerPos.proposal();
                if (p.position() == newID)
                    updateDisputes(nodeId, result_->txns);
            }
        }
 
        // Share our new position if we are still participating this round
        if (!result_->position.isBowOut() && (mode_.get() == ConsensusMode::proposing))
            adaptor_.propose(result_->position);
    }
}
 
template <class Adaptor>
bool
Consensus<Adaptor>::haveConsensus(std::unique_ptr<std::stringstream> const& clog)
{
    // Must have a stance if we are checking for consensus
    XRPL_ASSERT(result_, "xrpl::Consensus::haveConsensus : has result");
 
    // CHECKME: should possibly count unacquired TX sets as disagreeing
    int agree = 0, disagree = 0;
 
    auto ourPosition = result_->position.position();
 
    // Count number of agreements/disagreements with our position
    for (auto const& [nodeId, peerPos] : currPeerPositions_)
    {
        Proposal_t const& peerProp = peerPos.proposal();
        if (peerProp.position() == ourPosition)
        {
            ++agree;
        }
        else
        {
            JLOG(j_.debug()) << "Proposal disagreement: Peer " << nodeId << " has " << peerProp.position();
            ++disagree;
        }
    }
    auto currentFinished = adaptor_.proposersFinished(previousLedger_, prevLedgerID_);
 
    JLOG(j_.debug()) << "Checking for TX consensus: agree=" << agree << ", disagree=" << disagree;
 
    ConsensusParms const& parms = adaptor_.parms();
    // Stalling is BAD. It means that we have a consensus on the close time, so
    // peers are talking, but we have disputed transactions that peers are
    // unable or unwilling to come to agreement on one way or the other.
    bool const stalled = haveCloseTimeConsensus_ && !result_->disputes.empty() &&
        std::ranges::all_of(result_->disputes, [this, &parms, &clog](auto const& dispute) {
                             return dispute.second.stalled(
                                 parms, mode_.get() == ConsensusMode::proposing, peerUnchangedCounter_, j_, clog);
                         });
    if (stalled)
    {
        std::stringstream ss;
        ss << "Consensus detects as stalled with " << (agree + disagree) << "/" << prevProposers_ << " proposers, and "
           << result_->disputes.size() << " stalled disputed transactions.";
        JLOG(j_.error()) << ss.str();
        CLOG(clog) << ss.str();
    }
 
    // Determine if we actually have consensus or not
    result_->state = checkConsensus(
        prevProposers_,
        agree + disagree,
        agree,
        currentFinished,
        prevRoundTime_,
        result_->roundTime.read(),
        stalled,
        parms,
        mode_.get() == ConsensusMode::proposing,
        j_,
        clog);
 
    if (result_->state == ConsensusState::No)
    {
        CLOG(clog) << "No consensus. ";
        return false;
    }
 
    // Consensus has taken far too long. Drop out of the round.
    if (result_->state == ConsensusState::Expired)
    {
        static auto const minimumCounter = parms.avalancheCutoffs.size() * parms.avMIN_ROUNDS;
        std::stringstream ss;
        if (establishCounter_ < minimumCounter)
        {
            // If each round of phaseEstablish takes a very long time, we may
            // "expire" before we've given consensus enough time at each
            // avalanche level to actually come to a consensus. In that case,
            // keep trying. This should only happen if there are an extremely
            // large number of disputes such that each round takes an inordinate
            // amount of time.
 
            ss << "Consensus time has expired in round " << establishCounter_ << "; continue until round "
               << minimumCounter << ". " << Json::Compact{getJson(false)};
            JLOG(j_.error()) << ss.str();
            CLOG(clog) << ss.str() << ". ";
            return false;
        }
        ss << "Consensus expired. " << Json::Compact{getJson(true)};
        JLOG(j_.error()) << ss.str();
        CLOG(clog) << ss.str() << ". ";
        leaveConsensus(clog);
    }
    // There is consensus, but we need to track if the network moved on
    // without us.
    if (result_->state == ConsensusState::MovedOn)
    {
        JLOG(j_.error()) << "Unable to reach consensus";
        JLOG(j_.error()) << Json::Compact{getJson(true)};
        CLOG(clog) << "Unable to reach consensus " << Json::Compact{getJson(true)} << ". ";
    }
 
    CLOG(clog) << "Consensus has been reached. ";
    return true;
}
 
template <class Adaptor>
void
Consensus<Adaptor>::leaveConsensus(std::unique_ptr<std::stringstream> const& clog)
{
    if (mode_.get() == ConsensusMode::proposing)
    {
        if (result_ && !result_->position.isBowOut())
        {
            result_->position.bowOut(now_);
            adaptor_.propose(result_->position);
        }
 
        mode_.set(ConsensusMode::observing, adaptor_);
        JLOG(j_.info()) << "Bowing out of consensus";
        CLOG(clog) << "Bowing out of consensus. ";
    }
}
 
template <class Adaptor>
void
Consensus<Adaptor>::createDisputes(TxSet_t const& o, std::unique_ptr<std::stringstream> const& clog)
{
    // Cannot create disputes without our stance
    XRPL_ASSERT(result_, "xrpl::Consensus::createDisputes : result is set");
 
    // Only create disputes if this is a new set
    auto const emplaced = result_->compares.emplace(o.id()).second;
    CLOG(clog) << "createDisputes: new set? " << !emplaced << ". ";
    if (!emplaced)
        return;
 
    // Nothing to dispute if we agree
    if (result_->txns.id() == o.id())
    {
        CLOG(clog) << "both sets are identical. ";
        return;
    }
 
    CLOG(clog) << "comparing existing with new set: " << result_->txns.id() << ',' << o.id() << ". ";
    JLOG(j_.debug()) << "createDisputes " << result_->txns.id() << " to " << o.id();
 
    auto differences = result_->txns.compare(o);
 
    int dc = 0;
 
    for (auto const& [txId, inThisSet] : differences)
    {
        ++dc;
        // create disputed transactions (from the ledger that has them)
        XRPL_ASSERT(
            (inThisSet && result_->txns.find(txId) && !o.find(txId)) ||
                (!inThisSet && !result_->txns.find(txId) && o.find(txId)),
            "xrpl::Consensus::createDisputes : has disputed transactions");
 
        Tx_t tx = inThisSet ? result_->txns.find(txId) : o.find(txId);
        auto txID = tx.id();
 
        if (result_->disputes.find(txID) != result_->disputes.end())
            continue;
 
        JLOG(j_.debug()) << "Transaction " << txID << " is disputed";
 
        typename Result::Dispute_t dtx{
            tx, result_->txns.exists(txID), std::max(prevProposers_, currPeerPositions_.size()), j_};
 
        // Update all of the available peer's votes on the disputed transaction
        for (auto const& [nodeId, peerPos] : currPeerPositions_)
        {
            Proposal_t const& peerProp = peerPos.proposal();
            auto const cit = acquired_.find(peerProp.position());
            if (cit != acquired_.end() && dtx.setVote(nodeId, cit->second.exists(txID)))
                peerUnchangedCounter_ = 0;
        }
        adaptor_.share(dtx.tx());
 
        result_->disputes.emplace(txID, std::move(dtx));
    }
    JLOG(j_.debug()) << dc << " differences found";
    CLOG(clog) << "disputes: " << dc << ". ";
}
 
template <class Adaptor>
void
Consensus<Adaptor>::updateDisputes(NodeID_t const& node, TxSet_t const& other)
{
    // Cannot updateDisputes without our stance
    XRPL_ASSERT(result_, "xrpl::Consensus::updateDisputes : result is set");
 
    // Ensure we have created disputes against this set if we haven't seen
    // it before
    if (result_->compares.find(other.id()) == result_->compares.end())
        createDisputes(other);
 
    for (auto& it : result_->disputes)
    {
        auto& d = it.second;
        if (d.setVote(node, other.exists(d.tx().id())))
            peerUnchangedCounter_ = 0;
    }
}
 
template <class Adaptor>
NetClock::time_point
Consensus<Adaptor>::asCloseTime(NetClock::time_point raw) const
{
    return roundCloseTime(raw, closeResolution_);
}
 
}  // namespace xrpl