Consensus.h

//------------------------------------------------------------------------------
/*
    This file is part of rippled: https://github.com/ripple/rippled
    Copyright (c) 2012-2017 Ripple Labs Inc.
 
    Permission to use, copy, modify, and/or distribute this software for any
    purpose  with  or without fee is hereby granted, provided that the above
    copyright notice and this permission notice appear in all copies.
 
    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
 
#ifndef RIPPLE_CONSENSUS_CONSENSUS_H_INCLUDED
#define RIPPLE_CONSENSUS_CONSENSUS_H_INCLUDED
 
#include <ripple/basics/Log.h>
#include <ripple/basics/chrono.h>
#include <ripple/beast/container/aged_unordered_map.h>
#include <ripple/beast/utility/Journal.h>
#include <ripple/consensus/ConsensusParms.h>
#include <ripple/consensus/ConsensusProposal.h>
#include <ripple/consensus/ConsensusTypes.h>
#include <ripple/consensus/DisputedTx.h>
#include <ripple/consensus/LedgerTiming.h>
#include <ripple/json/json_writer.h>
#include <ripple/shamap/SHAMap.h>
#include <boost/logic/tribool.hpp>
 
#include <chrono>
#include <deque>
#include <iterator>
#include <optional>
#include <sstream>
 
namespace ripple {
 
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::optional<std::chrono::milliseconds> validationDelay,
    std::chrono::milliseconds idleInterval,
    ConsensusParms const& parms,
    beast::Journal j);
 
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,
    ConsensusParms const& parms,
    bool proposing,
    beast::Journal j);
 
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,
        typename Ledger_t::Seq>;
 
    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& 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);
 
    bool
    peerProposal(
        NetClock::time_point const& now,
        PeerPosition_t const& newProposal);
 
    void
    timerEntry(NetClock::time_point const& now);
 
    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);
 
    // Change our view of the previous ledger
    void
    handleWrongLedger(typename Ledger_t::ID const& lgrId);
 
    void
    checkLedger();
 
    void
    playbackProposals();
 
    bool
    peerProposalInternal(
        NetClock::time_point const& now,
        PeerPosition_t const& newProposal);
 
    void
    phaseOpen();
 
    void
    phaseEstablish();
 
    bool
    shouldPause() const;
 
    // Close the open ledger and establish initial position.
    void
    closeLedger();
 
    // Adjust our positions to try to agree with other validators.
    void
    updateOurPositions(bool const share);
 
    bool
    haveConsensus();
 
    // Create disputes between our position and the provided one.
    void
    createDisputes(TxSet_t const& o);
 
    // 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();
 
    // The rounded or effective close time estimate from a proposer
    NetClock::time_point
    asCloseTime(NetClock::time_point raw) const;
 
    Adaptor& adaptor_;
 
    ConsensusPhase phase_{ConsensusPhase::accepted};
    MonitoredMode mode_{ConsensusMode::observing};
    bool firstRound_ = true;
    bool haveCloseTimeConsensus_ = false;
 
    clock_type& 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;
 
    // 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.
    using AcquiredType = beast::aged_unordered_map<
        typename TxSet_t::ID,
        const TxSet_t,
        clock_type::clock_type,
        beast::uhash<>>;
    AcquiredType acquired_;
 
    // Tx sets that can be purged only once there is a new consensus round.
    std::stack<typename TxSet_t::ID> acquiredPurge_;
 
    std::optional<Result> result_;
    ConsensusCloseTimes rawCloseTimes_;
 
    //-------------------------------------------------------------------------
    // 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. Collected by ledger sequence. This allows us to
    // know which positions are likely relevant to the ledger on which we are
    // currently working. Also allows us to catch up faster if we fall behind
    // the rest of the network since we won't need to re-aquire proposals
    // and related transaction sets.
    std::map<typename Ledger_t::Seq, hash_map<NodeID_t, PeerPosition_t>>
        recentPeerPositions_;
 
    // These are for peers not using code that adds a ledger sequence
    // to the proposal message. TODO This should be removed eventually when
    // the network fully upgrades.
    hash_map<NodeID_t, std::deque<PeerPosition_t>> recentPeerPositionsLegacy_;
 
    // 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& clock,
    Adaptor& adaptor,
    beast::Journal journal)
    : adaptor_(adaptor), clock_(clock), acquired_(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)
{
    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;
    }
 
    // Clear positions that we know will not ever be necessary again.
    auto it = recentPeerPositions_.begin();
    while (it != recentPeerPositions_.end() && it->first <= prevLedger.seq())
        it = recentPeerPositions_.erase(it);
    // Get rid of untrusted positions for the current working ledger.
    auto currentPositions =
        recentPeerPositions_.find(prevLedger.seq() + typename Ledger_t::Seq{1});
    if (currentPositions != recentPeerPositions_.end())
    {
        for (NodeID_t const& n : nowUntrusted)
            currentPositions->second.erase(n);
    }
 
    for (NodeID_t const& n : nowUntrusted)
        recentPeerPositionsLegacy_.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);
}
template <class Adaptor>
void
Consensus<Adaptor>::startRoundInternal(
    NetClock::time_point const& now,
    typename Ledger_t::ID const& prevLedgerID,
    Ledger_t const& prevLedger,
    ConsensusMode mode)
{
    phase_ = ConsensusPhase::open;
    JLOG(j_.debug()) << "transitioned to ConsensusPhase::open";
    mode_.set(mode, adaptor_);
    now_ = now;
    prevLedgerID_ = prevLedgerID;
    previousLedger_ = prevLedger;
    result_.reset();
    convergePercent_ = 0;
    haveCloseTimeConsensus_ = false;
    openTime_.reset(clock_.now());
 
    // beast::aged_unordered_map::erase by key is broken and
    // is not used anywhere in the existing codebase.
    while (!acquiredPurge_.empty())
    {
        auto found = acquired_.find(acquiredPurge_.top());
        if (found != acquired_.end())
            acquired_.erase(found);
        acquiredPurge_.pop();
    }
    for (auto it = currPeerPositions_.begin(); it != currPeerPositions_.end();)
    {
        if (auto found = acquired_.find(it->second.proposal().position());
            found != acquired_.end())
        {
            acquired_.erase(found);
        }
        it = currPeerPositions_.erase(it);
    }
 
    // Hold up to 30 minutes worth of acquired tx sets. This to help
    // catch up quickly from extended de-sync periods.
    beast::expire(acquired_, std::chrono::minutes(30));
    rawCloseTimes_.peers.clear();
    rawCloseTimes_.self = {};
    deadNodes_.clear();
 
    closeResolution_ = getNextLedgerTimeResolution(
        previousLedger_.closeTimeResolution(),
        previousLedger_.closeAgree(),
        previousLedger_.seq() + typename Ledger_t::Seq{1});
 
    playbackProposals();
    if (currPeerPositions_.size() > (prevProposers_ / 2))
    {
        // We may be falling behind, don't wait for the timer
        // consider closing the ledger immediately
        timerEntry(now_);
    }
}
 
template <class Adaptor>
bool
Consensus<Adaptor>::peerProposal(
    NetClock::time_point const& now,
    PeerPosition_t const& newPeerPos)
{
    auto const& peerID = newPeerPos.proposal().nodeID();
 
    // Always need to store recent positions
    if (newPeerPos.proposal().ledgerSeq().has_value())
    {
        // Ignore proposals from prior ledgers.
        typename Ledger_t::Seq const& propLedgerSeq =
            *newPeerPos.proposal().ledgerSeq();
        if (propLedgerSeq <= previousLedger_.seq())
            return false;
 
        auto& bySeq = recentPeerPositions_[propLedgerSeq];
        {
            auto peerProp = bySeq.find(peerID);
            if (peerProp == bySeq.end())
            {
                bySeq.emplace(peerID, newPeerPos);
            }
            else
            {
                // Only store if it's the latest proposal from this peer for the
                // consensus round in the proposal.
                if (newPeerPos.proposal().proposeSeq() <=
                    peerProp->second.proposal().proposeSeq())
                {
                    return false;
                }
                peerProp->second = newPeerPos;
            }
        }
    }
    else
    {
        // legacy proposal with no ledger sequence
        auto& props = recentPeerPositionsLegacy_[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)
{
    now_ = now;
 
    auto const& newPeerProp = newPeerPos.proposal();
 
    if (newPeerProp.prevLedger() != prevLedgerID_)
    {
        JLOG(j_.debug()) << "Got proposal for " << newPeerProp.prevLedger()
                         << " but we are on " << prevLedgerID_;
 
        if (!acquired_.count(newPeerProp.position()))
        {
            // 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 we're behind, this should save
            // time when we catch up.
            if (auto set = adaptor_.acquireTxSet(newPeerProp.position()))
                gotTxSet(now_, *set);
            else
                JLOG(j_.debug()) << "Do not have tx set for peer";
        }
 
        // There's nothing else to do with this proposal currently.
        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())
            {
                // Remove from acquired_ or else it will consume space for
                // awhile. beast::aged_unordered_map::erase by key is broken and
                // is not used anywhere in the existing codebase.
                if (auto found =
                        acquired_.find(peerPosIt->second.proposal().position());
                    found != acquired_.end())
                {
                    acquiredPurge_.push(
                        peerPosIt->second.proposal().position());
                }
                currPeerPositions_.erase(peerID);
            }
            deadNodes_.insert(peerID);
 
            return true;
        }
 
        if (peerPosIt != currPeerPositions_.end())
        {
            // Remove from acquired_ or else it will consume space for awhile.
            // beast::aged_unordered_container::erase by key is broken and
            // is not used anywhere in the existing codebase.
            if (auto found = acquired_.find(newPeerPos.proposal().position());
                found != acquired_.end())
            {
                acquiredPurge_.push(newPeerPos.proposal().position());
            }
            // The proposal's arrival time determines how long the network
            // has been proposing, so new proposals from the same peer
            // should reflect the original's arrival time.
            newPeerPos.proposal().arrivalTime() =
                peerPosIt->second.proposal().arrivalTime();
            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)
{
    // Nothing to do if we are currently working on a ledger
    if (phase_ == ConsensusPhase::accepted)
        return;
 
    now_ = now;
 
    // Check we are on the proper ledger (this may change phase_)
    checkLedger();
 
    if (phase_ == ConsensusPhase::open)
        phaseOpen();
    else if (phase_ == ConsensusPhase::establish)
        phaseEstablish();
}
 
template <class Adaptor>
void
Consensus<Adaptor>::gotTxSet(
    NetClock::time_point const& now,
    TxSet_t const& txSet)
{
    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
        assert(id != result_->position.position());
        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)
{
    assert(lgrId != prevLedgerID_ || previousLedger_.id() != lgrId);
 
    // Stop proposing because we are out of sync
    leaveConsensus();
 
    // First time switching to this ledger
    if (prevLedgerID_ != lgrId)
    {
        prevLedgerID_ = lgrId;
 
        // Clear out state
        if (result_)
        {
            result_->disputes.clear();
            result_->compares.clear();
        }
 
        for (auto it = currPeerPositions_.begin();
             it != currPeerPositions_.end();)
        {
            // beast::aged_unordered_map::erase by key is broken and
            // is not used anywhere in the existing codebase.
            if (auto found = acquired_.find(it->second.proposal().position());
                found != acquired_.end())
            {
                acquiredPurge_.push(it->second.proposal().position());
            }
            it = currPeerPositions_.erase(it);
        }
        rawCloseTimes_.peers.clear();
        deadNodes_.clear();
 
        // Get back in sync, this will also recreate disputes
        playbackProposals();
    }
 
    if (previousLedger_.id() == 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_;
        startRoundInternal(
            now_, lgrId, *newLedger, ConsensusMode::switchedLedger);
    }
    else
    {
        mode_.set(ConsensusMode::wrongLedger, adaptor_);
    }
}
 
template <class Adaptor>
void
Consensus<Adaptor>::checkLedger()
{
    auto netLgr =
        adaptor_.getPrevLedger(prevLedgerID_, previousLedger_, mode_.get());
 
    if (netLgr != prevLedgerID_)
    {
        JLOG(j_.warn()) << "View of consensus changed during "
                        << to_string(phase_) << " status=" << to_string(phase_)
                        << ", "
                        << " mode=" << to_string(mode_.get());
        JLOG(j_.warn()) << prevLedgerID_ << " to " << netLgr;
        JLOG(j_.warn()) << Json::Compact{previousLedger_.getJson()};
        JLOG(j_.debug()) << "State on consensus change "
                         << Json::Compact{getJson(true)};
        handleWrongLedger(netLgr);
    }
    else if (previousLedger_.id() != prevLedgerID_)
        handleWrongLedger(netLgr);
}
 
template <class Adaptor>
void
Consensus<Adaptor>::playbackProposals()
{
    // Only use proposals for the ledger sequence we're currently working on.
    auto const currentPositions = recentPeerPositions_.find(
        previousLedger_.seq() + typename Ledger_t::Seq{1});
    if (currentPositions != recentPeerPositions_.end())
    {
        for (auto const& [peerID, pos] : currentPositions->second)
        {
            if (pos.proposal().prevLedger() == prevLedgerID_ &&
                peerProposalInternal(now_, pos))
            {
                adaptor_.share(pos);
            }
        }
    }
 
    // It's safe to do this--if a proposal is based on the wrong ledger,
    // then peerProposalInternal() will not replace it in currPeerPositions_.
    // TODO Eventually, remove code to check for non-existent ledger sequence
    // in peer proposal messages and make that parameter required in
    // the protobuf definition. Do this only after the network is running on
    // rippled versions with that parameter set in peer proposals. This
    // can be done once an amendment for another feature forces that kind
    // of upgrade, but this particular feature does not require an amendment.
    for (auto const& it : recentPeerPositionsLegacy_)
    {
        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()
{
    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;
    {
        bool previousCloseCorrect =
            (mode_.get() != ConsensusMode::wrongLedger) &&
            previousLedger_.closeAgree() &&
            (previousLedger_.closeTime() !=
             (previousLedger_.parentCloseTime() + 1s));
 
        auto lastCloseTime = previousCloseCorrect
            ? previousLedger_.closeTime()  // 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_);
    }
 
    auto const idleInterval = std::max<milliseconds>(
        adaptor_.parms().ledgerIDLE_INTERVAL,
        2 * previousLedger_.closeTimeResolution());
 
    // Decide if we should close the ledger
    if (shouldCloseLedger(
            anyTransactions,
            prevProposers_,
            proposersClosed,
            proposersValidated,
            prevRoundTime_,
            sinceClose,
            openTime_.read(),
            adaptor_.getValidationDelay(),
            idleInterval,
            adaptor_.parms(),
            j_))
    {
        closeLedger();
        adaptor_.setValidationDelay();
    }
}
 
template <class Adaptor>
bool
Consensus<Adaptor>::shouldPause() const
{
    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 << " (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();
        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();
    else
        j_.debug() << "not pausing" << vars.str();
    return willPause;
}
 
template <class Adaptor>
void
Consensus<Adaptor>::phaseEstablish()
{
    // can only establish consensus if we already took a stance
    assert(result_);
 
    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);
 
    {
        // Give everyone a chance to take an initial position unless enough
        // have already submitted theirs a long enough time ago
        // --because that means we're already
        // behind. Optimize pause duration if pausing. Pause until exactly
        // the number of ms after roundTime.read(), or the time since
        // receiving the earliest qualifying peer proposal. To protect
        // from faulty peers on the UNL, discard the earliest proposals
        // beyond the quorum threshold. For example, with a UNL of 20,
        // 80% quorum is 16. Assume the remaining 4 are Byzantine actors.
        // We therefore ignore the first 4 proposals received
        // for this calculation. We then take the earliest of either the
        // 5th proposal or our own proposal to determine whether enough
        // time has passed to possibly close. If not, then use that to
        // precisely determine how long to pause until checking again.
        std::size_t const q = adaptor_.quorum();
        std::size_t const discard =
            static_cast<std::size_t>(q / parms.minCONSENSUS_FACTOR) - q;
 
        std::chrono::milliseconds beginning;
        if (currPeerPositions_.size() > discard)
        {
            std::multiset<std::chrono::milliseconds> arrivals;
            for (auto& pos : currPeerPositions_)
            {
                pos.second.proposal().arrivalTime().tick(clock_.now());
                arrivals.insert(pos.second.proposal().arrivalTime().read());
            }
            auto it = arrivals.rbegin();
            std::advance(it, discard);
            beginning = *it;
        }
        else
        {
            beginning = result_->roundTime.read();
        }
 
        // Give everyone a chance to take an initial position
        if (beginning < parms.ledgerMIN_CONSENSUS)
        {
            adaptor_.setTimerDelay(parms.ledgerMIN_CONSENSUS - beginning);
            return;
        }
    }
 
    updateOurPositions(true);
 
    // Nothing to do if too many laggards or we don't have consensus.
    if (shouldPause() || !haveConsensus())
        return;
 
    if (!haveCloseTimeConsensus_)
    {
        JLOG(j_.info()) << "We have TX consensus but not CT consensus";
        return;
    }
 
    JLOG(j_.info()) << "Converge cutoff (" << currPeerPositions_.size()
                    << " participants)";
    adaptor_.updateOperatingMode(currPeerPositions_.size());
    prevProposers_ = currPeerPositions_.size();
    prevRoundTime_ = result_->roundTime.read();
    phase_ = ConsensusPhase::accepted;
    JLOG(j_.debug()) << "transitioned to ConsensusPhase::accepted";
 
    std::optional<std::pair<
        typename Adaptor::CanonicalTxSet_t,
        typename Adaptor::Ledger_t>>
        txsBuilt;
    // Track time spent retrying new ledger validation.
    std::optional<std::chrono::time_point<std::chrono::steady_clock>>
        startDelay;
    // Amount of time to pause checking for ledger to become validated.
    static auto const validationWait = std::chrono::milliseconds(100);
 
    // Make a copy of the result_ because it may be reset during the accept
    // phase if ledgers are switched and a new round is started.
    assert(result_.has_value());
    std::optional<Result const> result{result_};
    // Building the new ledger is time-consuming and safe to not lock, but
    // the rest of the logic below needs to be locked, until
    // finishing (onAccept).
    std::unique_lock<std::recursive_mutex> lock(adaptor_.peekMutex());
    do
    {
        if (!result_.has_value() ||
            result_->position.prevLedger() != result->position.prevLedger())
        {
            JLOG(j_.debug()) << "A new consensus round has started based on "
                                "a different ledger.";
            return;
        }
        if (txsBuilt)
        {
            if (!startDelay)
                startDelay = std::chrono::steady_clock::now();
 
            // Only send a single validation per round.
            adaptor_.clearValidating();
            // Check if a better proposal has been shared by the network.
            auto prevProposal = result_->position;
            updateOurPositions(false);
 
            if (prevProposal == result_->position)
            {
                JLOG(j_.debug())
                    << "old and new positions "
                       "match: "
                    << prevProposal.position() << " delay so far "
                    << std::chrono::duration_cast<std::chrono::milliseconds>(
                           std::chrono::steady_clock::now() - *startDelay)
                           .count()
                    << "ms. pausing";
                adaptor_.getLedgerMaster().waitForValidated(validationWait);
                continue;
            }
            JLOG(j_.debug()) << "retrying buildAndValidate with "
                                "new position: "
                             << result_->position.position();
            // Update the result used for the remainder of this Consensus round.
            assert(result_.has_value());
            result.emplace(*result_);
        }
        lock.unlock();
 
        // This is time-consuming and safe to not have under mutex.
        assert(result.has_value());
        txsBuilt = adaptor_.buildAndValidate(
            *result,
            previousLedger_,
            closeResolution_,
            mode_.get(),
            getJson(true));
        lock.lock();
    } while (adaptor_.retryAccept(txsBuilt->second, startDelay));
 
    if (startDelay)
    {
        auto const delay =
            std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::steady_clock::now() - *startDelay);
        JLOG(j_.debug()) << "validationDelay will be " << delay.count() << "ms";
        adaptor_.setValidationDelay(delay);
    }
 
    lock.unlock();
 
    assert(result.has_value());
    adaptor_.onAccept(
        *result,
        rawCloseTimes_,
        mode_.get(),
        getJson(true),
        std::move(*txsBuilt));
}
 
template <class Adaptor>
void
Consensus<Adaptor>::closeLedger()
{
    // We should not be closing if we already have a position
    assert(!result_);
 
    phase_ = ConsensusPhase::establish;
    JLOG(j_.debug()) << "transitioned to ConsensusPhase::establish";
    rawCloseTimes_.self = now_;
 
    result_.emplace(
        adaptor_.onClose(previousLedger_, now_, mode_.get(), clock_));
    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);
 
    if (mode_.get() == 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);
    }
    // There's no reason to pause, especially if we have fallen behind and
    // can possible agree to a consensus proposal already.
    timerEntry(now_);
}
 
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(bool const share)
{
    // We must have a position if we are updating it
    assert(result_);
    ConsensusParms const& parms = adaptor_.parms();
 
    // Compute a cutoff time
    auto const peerCutoff = now_ - parms.proposeFRESHNESS;
    auto const ourCutoff = now_ - parms.proposeINTERVAL;
 
    // 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);
                // Remove from acquired_ or else it will consume space for
                // awhile. beast::aged_unordered_map::erase by key is broken and
                // is not used anywhere in the existing codebase.
                if (auto found = acquired_.find(peerProp.position());
                    found != acquired_.end())
                {
                    acquiredPurge_.push(peerProp.position());
                }
                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
    {
        int neededWeight;
 
        if (convergePercent_ < parms.avMID_CONSENSUS_TIME)
            neededWeight = parms.avINIT_CONSENSUS_PCT;
        else if (convergePercent_ < parms.avLATE_CONSENSUS_TIME)
            neededWeight = parms.avMID_CONSENSUS_PCT;
        else if (convergePercent_ < parms.avSTUCK_CONSENSUS_TIME)
            neededWeight = parms.avLATE_CONSENSUS_PCT;
        else
            neededWeight = parms.avSTUCK_CONSENSUS_PCT;
 
        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);
 
        JLOG(j_.info()) << "Proposers:" << currPeerPositions_.size()
                        << " nw:" << neededWeight << " thrV:" << threshVote
                        << " thrC:" << threshConsensus;
 
        // An impasse is possible unless a validator pretends to change
        // its close time vote. Imagine 5 validators. 3 have positions
        // for close time t1, and 2 with t2. That's an impasse because
        // 75% will never be met. However, if one of the validators voting
        // for t2 switches to t1, then that will be 80% and sufficient
        // to break the impasse. It's also OK for those agreeing
        // with the 3 to pretend to vote for the one with 2, because
        // that will never exceed the threshold of 75%, even with as
        // few as 3 validators. The most it can achieve is 2/3.
        for (auto& [t, v] : closeTimeVotes)
        {
            if (adaptor_.validating() &&
                t != asCloseTime(result_->position.closeTime()))
            {
                JLOG(j_.debug()) << "Others have voted for a close time "
                                    "different than ours. Adding our vote "
                                    "to this one in case it is necessary "
                                    "to break an impasse.";
                ++v;
            }
            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;
                    // Make sure that the winning close time is the one
                    // that propagates to the rest of the function.
                    break;
                }
            }
        }
 
        if (!haveCloseTimeConsensus_)
        {
            JLOG(j_.debug())
                << "No CT consensus:"
                << " Proposers:" << currPeerPositions_.size()
                << " Mode:" << to_string(mode_.get())
                << " Thresh:" << threshConsensus
                << " Pos:" << consensusCloseTime.time_since_epoch().count();
        }
    }
 
    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);
 
        JLOG(j_.info()) << "Position change: CTime "
                        << consensusCloseTime.time_since_epoch().count()
                        << ", tx " << newID;
 
        result_->position.changePosition(newID, consensusCloseTime, now_);
 
        // Share our new transaction set and update disputes
        // if we haven't already received it. Unless we have already
        // accepted a position, but are recalculating because it didn't
        // validate.
        if (acquired_.emplace(newID, result_->txns).second && share)
        {
            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,
        // unless we have already accepted a position but are recalculating
        // because it didn't validate.
        if (!result_->position.isBowOut() &&
            (mode_.get() == ConsensusMode::proposing) && share)
            adaptor_.propose(result_->position);
    }
}
 
template <class Adaptor>
bool
Consensus<Adaptor>::haveConsensus()
{
    // Must have a stance if we are checking for consensus
    assert(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
            ++disagree;
    }
    auto currentFinished =
        adaptor_.proposersFinished(previousLedger_, prevLedgerID_);
 
    JLOG(j_.debug()) << "Checking for TX consensus: agree=" << agree
                     << ", disagree=" << disagree;
 
    // Determine if we actually have consensus or not
    result_->state = checkConsensus(
        prevProposers_,
        agree + disagree,
        agree,
        currentFinished,
        prevRoundTime_,
        result_->roundTime.read(),
        adaptor_.parms(),
        mode_.get() == ConsensusMode::proposing,
        j_);
 
    if (result_->state == ConsensusState::No)
        return false;
 
    // 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 MovedOn: "
                         << Json::Compact{getJson(true)};
    }
 
    return true;
}
 
template <class Adaptor>
void
Consensus<Adaptor>::leaveConsensus()
{
    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";
    }
}
 
template <class Adaptor>
void
Consensus<Adaptor>::createDisputes(TxSet_t const& o)
{
    // Cannot create disputes without our stance
    assert(result_);
 
    // Only create disputes if this is a new set
    if (!result_->compares.emplace(o.id()).second)
        return;
 
    // Nothing to dispute if we agree
    if (result_->txns.id() == o.id())
        return;
 
    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)
        assert(
            (inThisSet && result_->txns.find(txId) && !o.find(txId)) ||
            (!inThisSet && !result_->txns.find(txId) && o.find(txId)));
 
        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_.trace()) << "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));
        }
        adaptor_.share(dtx.tx());
 
        result_->disputes.emplace(txID, std::move(dtx));
    }
    JLOG(j_.trace()) << dc << " differences found";
}
 
template <class Adaptor>
void
Consensus<Adaptor>::updateDisputes(NodeID_t const& node, TxSet_t const& other)
{
    // Cannot updateDisputes without our stance
    assert(result_);
 
    // 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;
        d.setVote(node, other.exists(d.tx().id()));
    }
}
 
template <class Adaptor>
NetClock::time_point
Consensus<Adaptor>::asCloseTime(NetClock::time_point raw) const
{
    return roundCloseTime(raw, closeResolution_);
}
 
}  // namespace ripple
 
#endif