Prevent consensus from getting stuck in the establish phase (#5277)

- Detects if the consensus process is "stalled". If it is, then we can declare a 
  consensus and end successfully even if we do not have 80% agreement on
  our proposal.
  - "Stalled" is defined as:
    - We have a close time consensus
    - Each disputed transaction is individually stalled:
      - It has been in the final "stuck" 95% requirement for at least 2
        (avMIN_ROUNDS) "inner rounds" of phaseEstablish,
      - and either all of the other trusted proposers or this validator, if proposing,
        have had the same vote(s) for at least 4 (avSTALLED_ROUNDS) "inner
        rounds", and at least 80% of the validators (including this one, if
        appropriate) agree about the vote (whether yes or no).
- If we have been in the establish phase for more than 10x the previous
  consensus establish phase's time, then consensus is considered "expired",
  and we will leave the round, which sends a partial validation (indicating
  that the node is moving on without validating). Two restrictions avoid
  prematurely exiting, or having an extended exit in extreme situations.
  - The 10x time is clamped to be within a range of 15s
    (ledgerMAX_CONSENSUS) to 120s (ledgerABANDON_CONSENSUS).
  - If consensus has not had an opportunity to walk through all avalanche
    states (defined as not going through 8 "inner rounds" of phaseEstablish),
    then ConsensusState::Expired is treated as ConsensusState::No.
- When enough nodes leave the round, any remaining nodes will see they've
  fallen behind, and move on, too, generally before hitting the timeout. Any
  validations or partial validations sent during this time will help the
  consensus process bring the nodes back together.

Author: ximinez

Builds/levelization/results/ordering.txt

@@ -43,6 +43,7 @@ test.consensus > xrpl.basics
 test.consensus > xrpld.app
 test.consensus > xrpld.consensus
 test.consensus > xrpld.ledger
+test.consensus > xrpl.json
 test.core > test.jtx
 test.core > test.toplevel
 test.core > test.unit_test

docs/consensus.md

@@ -558,7 +558,7 @@ struct ConsensusResult
     ConsensusTimer roundTime;
 
     // Indicates state in which consensus ended.  Once in the accept phase
-    // will be either Yes or MovedOn
+    // will be either Yes or MovedOn or Expired
     ConsensusState state = ConsensusState::No;
 };
 

src/test/consensus/Consensus_test.cpp

@@ -52,7 +52,7 @@ class Tx
     {
     }
 
-    ID
+    ID const&
     id() const
     {
         return id_;

src/test/csf/Tx.h

@@ -35,6 +35,7 @@
 #include <xrpld/app/misc/NetworkOPs.h>
 #include <xrpld/app/misc/Transaction.h>
 #include <xrpld/app/misc/TxQ.h>
+#include <xrpld/app/misc/ValidatorKeys.h>
 #include <xrpld/app/misc/ValidatorList.h>
 #include <xrpld/app/misc/detail/AccountTxPaging.h>
 #include <xrpld/app/rdb/backend/SQLiteDatabase.h>
@@ -249,6 +250,12 @@ class NetworkOPsImp final : public NetworkOPs
               beast::get_abstract_clock<std::chrono::steady_clock>(),
               validatorKeys,
               app_.logs().journal("LedgerConsensus"))
+        , validatorPK_(
+              validatorKeys.keys ? validatorKeys.keys->publicKey
+                                 : decltype(validatorPK_){})
+        , validatorMasterPK_(
+              validatorKeys.keys ? validatorKeys.keys->masterPublicKey
+                                 : decltype(validatorMasterPK_){})
         , m_ledgerMaster(ledgerMaster)
         , m_job_queue(job_queue)
         , m_standalone(standalone)
@@ -732,6 +739,9 @@ class NetworkOPsImp final : public NetworkOPs
 
     RCLConsensus mConsensus;
 
+    std::optional<PublicKey> const validatorPK_;
+    std::optional<PublicKey> const validatorMasterPK_;
+
     ConsensusPhase mLastConsensusPhase;
 
     LedgerMaster& m_ledgerMaster;
@@ -1917,6 +1927,23 @@ NetworkOPsImp::beginConsensus(
 bool
 NetworkOPsImp::processTrustedProposal(RCLCxPeerPos peerPos)
 {
+    auto const& peerKey = peerPos.publicKey();
+    if (validatorPK_ == peerKey || validatorMasterPK_ == peerKey)
+    {
+        // Could indicate a operator misconfiguration where two nodes are
+        // running with the same validator key configured, so this isn't fatal,
+        // and it doesn't necessarily indicate peer misbehavior. But since this
+        // is a trusted message, it could be a very big deal. Either way, we
+        // don't want to relay the proposal. Note that the byzantine behavior
+        // detection in handleNewValidation will notify other peers.
+        UNREACHABLE(
+            "ripple::NetworkOPsImp::processTrustedProposal : received own "
+            "proposal");
+        JLOG(m_journal.error())
+            << "Received a TRUSTED proposal signed with my key from a peer";
+        return false;
+    }
+
     return mConsensus.peerProposal(app_.timeKeeper().closeTime(), peerPos);
 }
 

src/xrpld/app/misc/NetworkOPs.cpp

@@ -109,6 +109,7 @@ checkConsensusReached(
     bool count_self,
     std::size_t minConsensusPct,
     bool reachedMax,
+    bool stalled,
     std::unique_ptr<std::stringstream> const& clog)
 {
     CLOG(clog) << "checkConsensusReached params: agreeing: " << agreeing
@@ -138,6 +139,17 @@ checkConsensusReached(
         return false;
     }
 
+    // We only get stalled when every disputed transaction unequivocally has 80%
+    // (minConsensusPct) agreement, either for or against. That is: either under
+    // 20% or over 80% consensus (repectively "nay" or "yay"). This prevents
+    // manipulation by a minority of byzantine peers of which transactions make
+    // the cut to get into the ledger.
+    if (stalled)
+    {
+        CLOG(clog) << "consensus stalled. ";
+        return true;
+    }
+
     if (count_self)
     {
         ++agreeing;
@@ -147,6 +159,7 @@ checkConsensusReached(
     }
 
     std::size_t currentPercentage = (agreeing * 100) / total;
+
     CLOG(clog) << "currentPercentage: " << currentPercentage;
     bool const ret = currentPercentage >= minConsensusPct;
     if (ret)
@@ -168,6 +181,7 @@ checkConsensus(
     std::size_t currentFinished,
     std::chrono::milliseconds previousAgreeTime,
     std::chrono::milliseconds currentAgreeTime,
+    bool stalled,
     ConsensusParms const& parms,
     bool proposing,
     beast::Journal j,
@@ -181,7 +195,7 @@ checkConsensus(
                << " minimum duration to reach consensus: "
                << parms.ledgerMIN_CONSENSUS.count() << "ms"
                << " max consensus time " << parms.ledgerMAX_CONSENSUS.count()
-               << "s"
+               << "ms"
                << " minimum consensus percentage: " << parms.minCONSENSUS_PCT
                << ". ";
 
@@ -211,10 +225,12 @@ checkConsensus(
             proposing,
             parms.minCONSENSUS_PCT,
             currentAgreeTime > parms.ledgerMAX_CONSENSUS,
+            stalled,
             clog))
     {
-        JLOG(j.debug()) << "normal consensus";
-        CLOG(clog) << "reached. ";
+        JLOG((stalled ? j.warn() : j.debug()))
+            << "normal consensus" << (stalled ? ", but stalled" : "");
+        CLOG(clog) << "reached" << (stalled ? ", but stalled." : ".");
         return ConsensusState::Yes;
     }
 
@@ -226,13 +242,27 @@ checkConsensus(
             false,
             parms.minCONSENSUS_PCT,
             currentAgreeTime > parms.ledgerMAX_CONSENSUS,
+            false,
             clog))
     {
         JLOG(j.warn()) << "We see no consensus, but 80% of nodes have moved on";
         CLOG(clog) << "We see no consensus, but 80% of nodes have moved on";
         return ConsensusState::MovedOn;
     }
 
+    std::chrono::milliseconds const maxAgreeTime =
+        previousAgreeTime * parms.ledgerABANDON_CONSENSUS_FACTOR;
+    if (currentAgreeTime > std::clamp(
+                               maxAgreeTime,
+                               parms.ledgerMAX_CONSENSUS,
+                               parms.ledgerABANDON_CONSENSUS))
+    {
+        JLOG(j.warn()) << "consensus taken too long";
+        CLOG(clog) << "Consensus taken too long. ";
+        // Note the Expired result may be overridden by the caller.
+        return ConsensusState::Expired;
+    }
+
     // no consensus yet
     JLOG(j.trace()) << "no consensus";
     CLOG(clog) << "No consensus. ";

src/xrpld/consensus/Consensus.cpp

@@ -31,6 +31,7 @@
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/json/json_writer.h>
 
+#include <algorithm>
 #include <chrono>
 #include <deque>
 #include <optional>
@@ -81,6 +82,10 @@ shouldCloseLedger(
                              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 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
@@ -94,6 +99,7 @@ checkConsensus(
     std::size_t currentFinished,
     std::chrono::milliseconds previousAgreeTime,
     std::chrono::milliseconds currentAgreeTime,
+    bool stalled,
     ConsensusParms const& parms,
     bool proposing,
     beast::Journal j,
@@ -574,6 +580,9 @@ class Consensus
 
     NetClock::duration closeResolution_ = ledgerDefaultTimeResolution;
 
+    ConsensusParms::AvalancheState closeTimeAvalancheState_ =
+        ConsensusParms::init;
+
     // Time it took for the last consensus round to converge
     std::chrono::milliseconds prevRoundTime_;
 
@@ -599,6 +608,13 @@ class Consensus
     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
 
@@ -696,6 +712,7 @@ Consensus<Adaptor>::startRoundInternal(
     previousLedger_ = prevLedger;
     result_.reset();
     convergePercent_ = 0;
+    closeTimeAvalancheState_ = ConsensusParms::init;
     haveCloseTimeConsensus_ = false;
     openTime_.reset(clock_.now());
     currPeerPositions_.clear();
@@ -1351,6 +1368,9 @@ Consensus<Adaptor>::phaseEstablish(
     // can only establish consensus if we already took a stance
     XRPL_ASSERT(result_, "ripple::Consensus::phaseEstablish : result is set");
 
+    ++peerUnchangedCounter_;
+    ++establishCounter_;
+
     using namespace std::chrono;
     ConsensusParms const& parms = adaptor_.parms();
 
@@ -1417,6 +1437,8 @@ Consensus<Adaptor>::closeLedger(std::unique_ptr<std::stringstream> const& clog)
     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());
@@ -1550,16 +1572,11 @@ Consensus<Adaptor>::updateOurPositions(
     }
     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;
+        // 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();
@@ -1681,7 +1698,8 @@ Consensus<Adaptor>::haveConsensus(
         }
         else
         {
-            JLOG(j_.debug()) << nodeId << " has " << peerProp.position();
+            JLOG(j_.debug()) << "Proposal disagreement: Peer " << nodeId
+                             << " has " << peerProp.position();
             ++disagree;
         }
     }
@@ -1691,6 +1709,17 @@ Consensus<Adaptor>::haveConsensus(
     JLOG(j_.debug()) << "Checking for TX consensus: agree=" << agree
                      << ", disagree=" << disagree;
 
+    ConsensusParms const& parms = adaptor_.parms();
+    // Stalling is BAD
+    bool const stalled = haveCloseTimeConsensus_ &&
+        std::ranges::all_of(result_->disputes,
+                            [this, &parms](auto const& dispute) {
+                                return dispute.second.stalled(
+                                    parms,
+                                    mode_.get() == ConsensusMode::proposing,
+                                    peerUnchangedCounter_);
+                            });
+
     // Determine if we actually have consensus or not
     result_->state = checkConsensus(
         prevProposers_,
@@ -1699,7 +1728,8 @@ Consensus<Adaptor>::haveConsensus(
         currentFinished,
         prevRoundTime_,
         result_->roundTime.read(),
-        adaptor_.parms(),
+        stalled,
+        parms,
         mode_.get() == ConsensusMode::proposing,
         j_,
         clog);
@@ -1710,6 +1740,33 @@ Consensus<Adaptor>::haveConsensus(
         return false;
     }
 
+    // Consensus has taken far too long. Drop out of the round.
+    if (result_->state == ConsensusState::Expired)
+    {
+        static auto const minimumCounter =
+            parms.avalancheCutoffs.size() * parms.avMIN_ROUNDS;
+        std::stringstream ss;
+        if (establishCounter_ < minimumCounter)
+        {
+            // If each round of phaseEstablish takes a very long time, we may
+            // "expire" before we've given consensus enough time at each
+            // avalanche level to actually come to a consensus. In that case,
+            // keep trying. This should only happen if there are an extremely
+            // large number of disputes such that each round takes an inordinate
+            // amount of time.
+
+            ss << "Consensus time has expired in round " << establishCounter_
+               << "; continue until round " << minimumCounter << ". "
+               << Json::Compact{getJson(false)};
+            JLOG(j_.error()) << ss.str();
+            CLOG(clog) << ss.str() << ". ";
+            return false;
+        }
+        ss << "Consensus expired. " << Json::Compact{getJson(true)};
+        JLOG(j_.error()) << ss.str();
+        CLOG(clog) << ss.str() << ". ";
+        leaveConsensus(clog);
+    }
     // There is consensus, but we need to track if the network moved on
     // without us.
     if (result_->state == ConsensusState::MovedOn)
@@ -1802,8 +1859,9 @@ Consensus<Adaptor>::createDisputes(
         {
             Proposal_t const& peerProp = peerPos.proposal();
             auto const cit = acquired_.find(peerProp.position());
-            if (cit != acquired_.end())
-                dtx.setVote(nodeId, cit->second.exists(txID));
+            if (cit != acquired_.end() &&
+                dtx.setVote(nodeId, cit->second.exists(txID)))
+                peerUnchangedCounter_ = 0;
         }
         adaptor_.share(dtx.tx());
 
@@ -1828,7 +1886,8 @@ Consensus<Adaptor>::updateDisputes(NodeID_t const& node, TxSet_t const& other)
     for (auto& it : result_->disputes)
     {
         auto& d = it.second;
-        d.setVote(node, other.exists(d.tx().id()));
+        if (d.setVote(node, other.exists(d.tx().id())))
+            peerUnchangedCounter_ = 0;
     }
 }