#pragma once

#include <xrpl/basics/Log.h>
#include <xrpl/basics/UnorderedContainers.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/clock/abstract_clock.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/consensus/ConsensusParms.h>
#include <xrpl/consensus/ConsensusProposal.h>
#include <xrpl/consensus/ConsensusTypes.h>
#include <xrpl/json/json_value.h>
#include <xrpl/json/json_writer.h>
#include <xrpl/ledger/LedgerTiming.h>

#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <map>
#include <memory>
#include <optional>
#include <sstream>
#include <string>
#include <utility>

namespace xrpl {

/**
 * Determines whether the current ledger should close at this time.
 *
 * This function should be called when a ledger is open and there is no close
 * in progress, or when a transaction is received and no close is in progress.
 *
 * @param anyTransactions indicates whether any transactions have been received
 * @param prevProposers proposers in the last closing
 * @param proposersClosed proposers who have currently closed this ledger
 * @param proposersValidated proposers who have validated the last closed
 *                           ledger
 * @param prevRoundTime time for the previous ledger to reach consensus
 * @param timeSincePrevClose  time since the previous ledger's (possibly
 * rounded) close time
 * @param openTime     duration this ledger has been open
 * @param idleInterval the network's desired idle interval
 * @param parms        Consensus constant parameters
 * @param j            journal for logging
 * @param clog         log object to which to append
 */
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 = {});

/**
 * Determine whether the network reached consensus and whether we joined.
 *
 * @param prevProposers proposers in the last closing (not including us)
 * @param currentProposers proposers in this closing so far (not including us)
 * @param currentAgree proposers who agree with us
 * @param currentFinished proposers who have validated a ledger after this one
 * @param previousAgreeTime how long, in milliseconds, it took to agree on the
 *                          last ledger
 * @param currentAgreeTime how long, in milliseconds, we've been trying to
 *                         agree
 * @param stalled the network appears to be stalled, where
 *        neither we nor our peers have changed their vote on any disputes in a
 *        while. This is undesirable, and should be rare, and will cause us to
 *        end consensus without 80% agreement.
 * @param parms            Consensus constant parameters
 * @param proposing        whether we should count ourselves
 * @param j                journal for logging
 * @param clog             log object to which to append
 */
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 = {});

/**
 * Generic implementation of consensus algorithm.
 *
 * Achieves consensus on the next ledger.
 *
 * Two things need consensus:
 *
 *   1.  The set of transactions included in the ledger.
 *   2.  The close time for the ledger.
 *
 * The basic flow:
 *
 *   1. A call to `startRound` places the node in the `Open` phase.  In this
 *      phase, the node is waiting for transactions to include in its open
 *      ledger.
 *   2. Successive calls to `timerEntry` check if the node can close the ledger.
 *      Once the node `Close`s the open ledger, it transitions to the
 *      `Establish` phase.  In this phase, the node shares/receives peer
 *      proposals on which transactions should be accepted in the closed ledger.
 *   3. During a subsequent call to `timerEntry`, the node determines it has
 *      reached consensus with its peers on which transactions to include. It
 *      transitions to the `Accept` phase. In this phase, the node works on
 *      applying the transactions to the prior ledger to generate a new closed
 *      ledger. Once the new ledger is completed, the node shares the validated
 *      ledger with the network, does some book-keeping, then makes a call to
 *      `startRound` to start the cycle again.
 *
 * This class uses a generic interface to allow adapting Consensus for specific
 * applications. The Adaptor template implements a set of helper functions that
 * plug the consensus algorithm into a specific application.  It also identifies
 * the types that play important roles in Consensus (transactions, ledgers, ...).
 * The code stubs below outline the interface and type requirements.  The traits
 * types must be copy constructible and assignable.
 *
 * @warning The generic implementation is not thread safe and the public methods
 * are not intended to be run concurrently.  When in a concurrent environment,
 * the application is responsible for ensuring thread-safety.  Simply locking
 * whenever touching the Consensus instance is one option.
 *
 * @code
 * // A single transaction
 * struct Tx
 * {
 *   // Unique identifier of transaction
 *   using ID = ...;
 *
 *   ID id() const;
 *
 * };
 *
 * // A set of transactions
 * struct TxSet
 * {
 *   // Unique ID of TxSet (not of Tx)
 *   using ID = ...;
 *   // Type of individual transaction comprising the TxSet
 *   using Tx = Tx;
 *
 *   bool exists(Tx::ID const &) const;
 *   // Return value should have semantics like Tx const *
 *   Tx const * find(Tx::ID const &) const ;
 *   ID const & id() const;
 *
 *   // Return set of transactions that are not common to this set or other
 *   // boolean indicates which set it was in
 *   std::map<Tx::ID, bool> compare(TxSet const & other) const;
 *
 *   // A mutable view of transactions
 *   struct MutableTxSet
 *   {
 *       MutableTxSet(TxSet const &);
 *       bool insert(Tx const &);
 *       bool erase(Tx::ID const &);
 *   };
 *
 *   // Construct from a mutable view.
 *   TxSet(MutableTxSet const &);
 *
 *   // Alternatively, if the TxSet is itself mutable
 *   // just alias MutableTxSet = TxSet
 *
 * };
 *
 * // Agreed upon state that consensus transactions will modify
 * struct Ledger
 * {
 *   using ID = ...;
 *   using Seq = ...;
 *
 *   // Unique identifier of ledger
 *   ID const id() const;
 *   Seq seq() const;
 *   auto closeTimeResolution() const;
 *   auto closeAgree() const;
 *   auto closeTime() const;
 *   auto parentCloseTime() const;
 *   json::Value getJson() const;
 * };
 *
 * // Wraps a peer's ConsensusProposal
 * struct PeerPosition
 * {
 *   ConsensusProposal<
 *       std::uint32_t, //NodeID,
 *       typename Ledger::ID,
 *       typename TxSet::ID> const &
 *   proposal() const;
 *
 * };
 *
 *
 * class Adaptor
 * {
 * public:
 *     //-----------------------------------------------------------------------
 *     // Define consensus types
 *     using Ledger_t = Ledger;
 *     using NodeID_t = std::uint32_t;
 *     using TxSet_t = TxSet;
 *     using PeerPosition_t = PeerPosition;
 *
 *     //-----------------------------------------------------------------------
 *     //
 *     // Attempt to acquire a specific ledger.
 *     std::optional<Ledger> acquireLedger(Ledger::ID const & ledgerID);
 *
 *     // Acquire the transaction set associated with a proposed position.
 *     std::optional<TxSet> acquireTxSet(TxSet::ID const & setID);
 *
 *     // Whether any transactions are in the open ledger
 *     bool hasOpenTransactions() const;
 *
 *     // Number of proposers that have validated the given ledger
 *     std::size_t proposersValidated(Ledger::ID const & prevLedger) const;
 *
 *     // Number of proposers that have validated a ledger descended from the
 *     // given ledger; if prevLedger.id() != prevLedgerID, use prevLedgerID
 *     // for the determination
 *     std::size_t proposersFinished(Ledger const & prevLedger,
 *                                   Ledger::ID const & prevLedger) const;
 *
 *     // Return the ID of the last closed (and validated) ledger that the
 *     // application thinks consensus should use as the prior ledger.
 *     Ledger::ID getPrevLedger(Ledger::ID const & prevLedgerID,
 *                     Ledger const & prevLedger,
 *                     Mode mode);
 *
 *     // Called whenever consensus operating mode changes
 *     void onModeChange(ConsensusMode before, ConsensusMode after);
 *
 *     // Called when ledger closes
 *     Result onClose(Ledger const &, Ledger const & prev, Mode mode);
 *
 *     // Called when ledger is accepted by consensus
 *     void onAccept(Result const & result,
 *       RCLCxLedger const & prevLedger,
 *       NetClock::duration closeResolution,
 *       CloseTimes const & rawCloseTimes,
 *       Mode const & mode);
 *
 *     // Called when ledger was forcibly accepted by consensus via the simulate
 *     // function.
 *     void onForceAccept(Result const & result,
 *       RCLCxLedger const & prevLedger,
 *       NetClock::duration closeResolution,
 *       CloseTimes const & rawCloseTimes,
 *       Mode const & mode);
 *
 *     // Propose the position to peers.
 *     void propose(ConsensusProposal<...> const & pos);
 *
 *     // Share a received peer proposal with other peer's.
 *     void share(PeerPosition_t const & prop);
 *
 *     // Share a disputed transaction with peers
 *     void share(Txn const & tx);
 *
 *     // Share given transaction set with peers
 *     void share(TxSet const &s);
 *
 *     // Consensus timing parameters and constants
 *     ConsensusParms const &
 *     parms() const;
 * };
 * @endcode
 *
 * @tparam Adaptor Defines types and provides helper functions needed to adapt
 *                 Consensus to the larger application.
 */
template <class Adaptor>
class Consensus
{
    using Ledger_t = Adaptor::Ledger_t;
    using TxSet_t = Adaptor::TxSet_t;
    using NodeID_t = Adaptor::NodeID_t;
    using Tx_t = TxSet_t::Tx;
    using PeerPosition_t = 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}
        {
        }
        [[nodiscard]] ConsensusMode
        get() const
        {
            return mode_;
        }

        void
        set(ConsensusMode mode, Adaptor& a)
        {
            a.onModeChange(mode_, mode);
            mode_ = mode;
        }
    };

public:
    /**
     * Clock type for measuring time within the consensus code
     */
    using clock_type = beast::AbstractClock<std::chrono::steady_clock>;

    Consensus(Consensus&&) noexcept = default;

    /**
     * Constructor.
     *
     * @param clock The clock used to internally sample consensus progress
     * @param adaptor The instance of the adaptor class
     * @param j The journal to log debug output
     */
    Consensus(clock_type const& clock, Adaptor& adaptor, beast::Journal j);

    /**
     * Kick-off the next round of consensus.
     *
     * Called by the client code to start each round of consensus.
     *
     * @param now          The network adjusted time
     * @param prevLedgerID the ID of the last ledger
     * @param prevLedger   The last ledger
     * @param nowUntrusted ID of nodes that are newly untrusted this round
     * @param proposing    Whether we want to send proposals to peers this
     * round.
     * @param clog         log object to which to append
     *
     * @note @b prevLedgerID is not required to the ID of @b prevLedger since
     * the ID may be known locally before the contents of the ledger arrive
     */
    void
    startRound(
        NetClock::time_point const& now,
        Ledger_t::ID const& prevLedgerID,
        Ledger_t prevLedger,
        hash_set<NodeID_t> const& nowUntrusted,
        bool proposing,
        std::unique_ptr<std::stringstream> const& clog = {});

    /**
     * A peer has proposed a new position, adjust our tracking.
     *
     * @param now The network adjusted time
     * @param newProposal The new proposal from a peer
     * @return Whether we should do delayed relay of this proposal.
     */
    bool
    peerProposal(NetClock::time_point const& now, PeerPosition_t const& newProposal);

    /**
     * Call periodically to drive consensus forward.
     *
     * @param now  The network adjusted time
     * @param clog log object to which to append
     */
    void
    timerEntry(
        NetClock::time_point const& now,
        std::unique_ptr<std::stringstream> const& clog = {});

    /**
     * Process a transaction set acquired from the network
     *
     * @param now The network adjusted time
     * @param txSet the transaction set
     */
    void
    gotTxSet(NetClock::time_point const& now, TxSet_t const& txSet);

    /**
     * Simulate the consensus process without any network traffic.
     *
     * The end result, is that consensus begins and completes as if everyone
     * had agreed with whatever we propose.
     *
     * This function is only called from the rpc "ledger_accept" path with the
     * server in standalone mode and SHOULD NOT be used during the normal
     * consensus process.
     *
     * Simulate will call onForceAccept since clients are manually driving
     * consensus to the accept phase.
     *
     * @param now The current network adjusted time.
     * @param consensusDelay Duration to delay between closing and accepting the
     *                       ledger. Uses 100ms if unspecified.
     */
    void
    simulate(
        NetClock::time_point const& now,
        std::optional<std::chrono::milliseconds> consensusDelay);

    /**
     * Get the previous ledger ID.
     *
     * The previous ledger is the last ledger seen by the consensus code and
     * should correspond to the most recent validated ledger seen by this peer.
     *
     * @return ID of previous ledger
     */
    Ledger_t::ID
    prevLedgerID() const
    {
        return prevLedgerID_;
    }

    [[nodiscard]] ConsensusPhase
    phase() const
    {
        return phase_;
    }

    /**
     * Get the Json state of the consensus process.
     *
     * Called by the consensus_info RPC.
     *
     * @param full True if verbose response desired.
     * @return     The Json state.
     */
    [[nodiscard]] json::Value
    getJson(bool full) const;

private:
    void
    startRoundInternal(
        NetClock::time_point const& now,
        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(Ledger_t::ID const& lgrId, std::unique_ptr<std::stringstream> const& clog);

    /**
     * Check if our previous ledger matches the network's.
     *
     * If the previous ledger differs, we are no longer in sync with
     * the network and need to bow out/switch modes.
     */
    void
    checkLedger(std::unique_ptr<std::stringstream> const& clog);

    /**
     * If we radically changed our consensus context for some reason,
     * we need to replay recent proposals so that they're not lost.
     */
    void
    playbackProposals();

    /**
     * Handle a replayed or a new peer proposal.
     */
    bool
    peerProposalInternal(NetClock::time_point const& now, PeerPosition_t const& newProposal);

    /**
     * Handle pre-close phase.
     *
     * In the pre-close phase, the ledger is open as we wait for new
     * transactions.  After enough time has elapsed, we will close the ledger,
     * switch to the establish phase and start the consensus process.
     */
    void
    phaseOpen(std::unique_ptr<std::stringstream> const& clog);

    /**
     * Handle establish phase.
     *
     * In the establish phase, the ledger has closed and we work with peers
     * to reach consensus. Update our position only on the timer, and in this
     * phase.
     *
     * If we have consensus, move to the accepted phase.
     */
    void
    phaseEstablish(std::unique_ptr<std::stringstream> const& clog);

    /**
     * Evaluate whether pausing increases likelihood of validation.
     *
     *  As a validator that has previously synced to the network, if our most
     *  recent locally-validated ledger did not also achieve
     *  full validation, then consider pausing for awhile based on
     *  the state of other validators.
     *
     *  Pausing may be beneficial in this situation if at least one validator
     *  is known to be on a sequence number earlier than ours. The minimum
     *  number of validators on the same sequence number does not guarantee
     *  consensus, and waiting for all validators may be too time-consuming.
     *  Therefore, a variable threshold is enacted based on the number
     *  of ledgers past network validation that we are on. For the first phase,
     *  the threshold is also the minimum required for quorum. For the last,
     *  no online validators can have a lower sequence number. For intermediate
     *  phases, the threshold is linear between the minimum required for
     *  quorum and 100%. For example, with 3 total phases and a quorum of
     *  80%, the 2nd phase would be 90%. Once the final phase is reached,
     *  if consensus still fails to occur, the cycle is begun again at phase 1.
     *
     * @return Whether to pause to wait for lagging proposers.
     */
    [[nodiscard]] 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
    [[nodiscard]] 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_ = kLedgerDefaultTimeResolution;

    ConsensusParms::AvalancheState closeTimeAvalancheState_ = ConsensusParms::AvalancheState::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
    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,
    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().ledgerIdleInterval;
        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,
    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::AvalancheState::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({});
    // NOLINTBEGIN(bugprone-unchecked-optional-access) closeLedger sets result_
    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));
    // NOLINTEND(bugprone-unchecked-optional-access)
    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::ValueType::Object);

    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::ValueType::Object);

            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::ValueType::Array);
            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::ValueType::Object);
            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::ValueType::Object);
            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::ValueType::Array);
            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(
    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 const 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().ledgerIdleInterval, 2 * previousLedger_.closeTimeResolution());
    CLOG(clog) << "idle interval set to " << idleInterval.count() << "ms based on "
               << "ledgerIDLE_INTERVAL: " << adaptor_.parms().ledgerIdleInterval.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 const 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()
         << ", "
         // NOLINTBEGIN(bugprone-unchecked-optional-access) result_ is always set when shouldPause
         // is called (from phaseEstablish after assert)
         << "roundTime: " << result_->roundTime.read().count()
         << ", "
         // NOLINTEND(bugprone-unchecked-optional-access)
         << "max consensus time: " << parms.ledgerMaxConsensus.count() << ", "
         << "validators: " << totalValidators << ", "
         << "laggards: " << laggards << ", "
         << "offline: " << offline << ", "
         << "quorum: " << quorum << ")";

    if ((ahead == 0u) || (laggards == 0u) || (totalValidators == 0u) || !adaptor_.validator() ||
        !adaptor_.haveValidated() ||
        // NOLINTNEXTLINE(bugprone-unchecked-optional-access) result_ set as shouldPause called
        result_->roundTime.read() > parms.ledgerMaxConsensus)
    {
        j_.debug() << "not pausing (early)" << vars.str();
        CLOG(clog) << "Not pausing (early). ";
        return false;
    }

    bool willPause = false;

    /**
     * Maximum phase with distinct thresholds to determine how
     *  many validators must be on our same ledger sequence number.
     *  The threshold for the 1st (0) phase is >= the minimum number that
     *  can achieve quorum. Threshold for the maximum phase is 100%
     *  of all trusted validators. Progression from min to max phase is
     *  simply linear. If there are 5 phases (maxPausePhase = 4)
     *  and minimum quorum is 80%, then thresholds progress as follows:
     *  0: >=80%
     *  1: >=85%
     *  2: >=90%
     *  3: >=95%
     *  4: =100%
     */
    static constexpr std::size_t kMaxPausePhase = 4;

    /**
     * No particular threshold guarantees consensus. Lower thresholds
     * are easier to achieve than higher, but higher thresholds are
     * more likely to reach consensus. Cycle through the phases if
     * lack of synchronization continues.
     *
     * Current information indicates that no phase is likely to be intrinsically
     * better than any other: the lower the threshold, the less likely that
     * up-to-date nodes will be able to reach consensus without the laggards.
     * But the higher the threshold, the longer the likely resulting pause.
     * 100% is slightly less desirable in the long run because the potential
     * of a single dawdling peer to slow down everything else. So if we
     * accept that no phase is any better than any other phase, but that
     * all of them will potentially enable us to arrive at consensus, cycling
     * through them seems to be appropriate. Further, if we do reach the
     * point of having to cycle back around, then it's likely that something
     * else out of the scope of this delay mechanism is wrong with the
     * network.
     */
    std::size_t const phase = (ahead - 1) % (kMaxPausePhase + 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 kMaxPausePhase:
            // 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) / kMaxPausePhase;

            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");
    // NOLINTBEGIN(bugprone-unchecked-optional-access) assert above

    ++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.avMinConsensusTime);
    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.avMinConsensusTime.count() << "ms. ";

    // Give everyone a chance to take an initial position
    if (result_->roundTime.read() < parms.ledgerMinConsensus)
    {
        CLOG(clog) << "ledgerMIN_CONSENSUS not reached: " << parms.ledgerMinConsensus.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());
    // NOLINTEND(bugprone-unchecked-optional-access)
}

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);
    }
}

/**
 * How many of the participants must agree to reach a given threshold?
 *
 * Note that the number may not precisely yield the requested percentage.
 * For example, with with size = 5 and percent = 70, we return 3, but
 * 3 out of 5 works out to 60%. There are no security implications to
 * this.
 *
 * @param participants The number of participants (i.e. validators)
 * @param percent The percent that we want to reach
 *
 * @return the number of participants which must agree
 */
inline int
participantsNeeded(int participants, int percent)
{
    int const 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");
    // NOLINTBEGIN(bugprone-unchecked-optional-access) assert above
    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.avCtConsensusPct);

        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);
    }
    // NOLINTEND(bugprone-unchecked-optional-access)
}

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");
    // NOLINTBEGIN(bugprone-unchecked-optional-access) assert above

    // 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 kMinimumCounter = parms.avalancheCutoffs.size() * parms.avMinRounds;
        std::stringstream ss;
        if (establishCounter_ < kMinimumCounter)
        {
            // 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 " << kMinimumCounter << ". "
               << 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. ";
    // NOLINTEND(bugprone-unchecked-optional-access)
    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");
    // NOLINTBEGIN(bugprone-unchecked-optional-access) assert above

    // 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 const 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 << ". ";
    // NOLINTEND(bugprone-unchecked-optional-access)
}

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");
    // NOLINTBEGIN(bugprone-unchecked-optional-access) assert above

    // 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;
    }
    // NOLINTEND(bugprone-unchecked-optional-access)
}

template <class Adaptor>
NetClock::time_point
Consensus<Adaptor>::asCloseTime(NetClock::time_point raw) const
{
    return roundCloseTime(raw, closeResolution_);
}

}  // namespace xrpl