Several changes to improve Consensus stability:

 * Verify accepted ledger becomes validated, and retry
   with a new consensus transaction set if not.
 * Always store proposals.
 * Track proposals by ledger sequence. This helps slow peers catch
   up with the rest of the network.
 * Acquire transaction sets for proposals with future ledger sequences.
   This also helps slow peers catch up.
 * Optimize timer delay for establish phase to wait based on how
   long validators have been sending proposals. This also helps slow
   peers to catch up.
 * Fix impasse achieving close time consensus.
 * Don't wait between open and establish phases.

Author: Mark Travis

src/ripple/app/consensus/RCLConsensus.cpp

@@ -55,7 +55,7 @@ RCLConsensus::RCLConsensus(
     LedgerMaster& ledgerMaster,
     LocalTxs& localTxs,
     InboundTransactions& inboundTransactions,
-    Consensus<Adaptor>::clock_type const& clock,
+    Consensus<Adaptor>::clock_type& clock,
     ValidatorKeys const& validatorKeys,
     beast::Journal journal)
     : adaptor_(
@@ -171,6 +171,9 @@ RCLConsensus::Adaptor::share(RCLCxPeerPos const& peerPos)
     auto const sig = peerPos.signature();
     prop.set_signature(sig.data(), sig.size());
 
+    if (proposal.ledgerSeq().has_value())
+        prop.set_ledgerseq(*proposal.ledgerSeq());
+
     app_.overlay().relay(prop, peerPos.suppressionID(), peerPos.publicKey());
 }
 
@@ -180,7 +183,7 @@ RCLConsensus::Adaptor::share(RCLCxTx const& tx)
     // If we didn't relay this transaction recently, relay it to all peers
     if (app_.getHashRouter().shouldRelay(tx.id()))
     {
-        JLOG(j_.debug()) << "Relaying disputed tx " << tx.id();
+        JLOG(j_.trace()) << "Relaying disputed tx " << tx.id();
         auto const slice = tx.tx_->slice();
         protocol::TMTransaction msg;
         msg.set_rawtransaction(slice.data(), slice.size());
@@ -192,13 +195,13 @@ RCLConsensus::Adaptor::share(RCLCxTx const& tx)
     }
     else
     {
-        JLOG(j_.debug()) << "Not relaying disputed tx " << tx.id();
+        JLOG(j_.trace()) << "Not relaying disputed tx " << tx.id();
     }
 }
 void
 RCLConsensus::Adaptor::propose(RCLCxPeerPos::Proposal const& proposal)
 {
-    JLOG(j_.trace()) << (proposal.isBowOut() ? "We bow out: " : "We propose: ")
+    JLOG(j_.debug()) << (proposal.isBowOut() ? "We bow out: " : "We propose: ")
                      << ripple::to_string(proposal.prevLedger()) << " -> "
                      << ripple::to_string(proposal.position());
 
@@ -212,6 +215,7 @@ RCLConsensus::Adaptor::propose(RCLCxPeerPos::Proposal const& proposal)
     prop.set_closetime(proposal.closeTime().time_since_epoch().count());
     prop.set_nodepubkey(
         validatorKeys_.publicKey.data(), validatorKeys_.publicKey.size());
+    prop.set_ledgerseq(*proposal.ledgerSeq());
 
     auto sig = signDigest(
         validatorKeys_.publicKey,
@@ -297,7 +301,8 @@ auto
 RCLConsensus::Adaptor::onClose(
     RCLCxLedger const& ledger,
     NetClock::time_point const& closeTime,
-    ConsensusMode mode) -> Result
+    ConsensusMode mode,
+    clock_type& clock) -> Result
 {
     const bool wrongLCL = mode == ConsensusMode::wrongLedger;
     const bool proposing = mode == ConsensusMode::proposing;
@@ -379,7 +384,7 @@ RCLConsensus::Adaptor::onClose(
 
     // Needed because of the move below.
     auto const setHash = initialSet->getHash().as_uint256();
-
+    initialLedger->info().seq;
     return Result{
         std::move(initialSet),
         RCLCxPeerPos::Proposal{
@@ -388,7 +393,9 @@ RCLConsensus::Adaptor::onClose(
             setHash,
             closeTime,
             app_.timeKeeper().closeTime(),
-            validatorKeys_.nodeID}};
+            validatorKeys_.nodeID,
+            initialLedger->info().seq,
+            clock}};
 }
 
 void
@@ -400,50 +407,43 @@ RCLConsensus::Adaptor::onForceAccept(
     ConsensusMode const& mode,
     Json::Value&& consensusJson)
 {
-    doAccept(
-        result,
-        prevLedger,
-        closeResolution,
-        rawCloseTimes,
-        mode,
-        std::move(consensusJson));
+    auto txsBuilt = buildAndValidate(
+        result, prevLedger, closeResolution, mode, std::move(consensusJson));
+    prepareOpenLedger(std::move(txsBuilt), result, rawCloseTimes, mode);
 }
 
 void
 RCLConsensus::Adaptor::onAccept(
     Result const& result,
-    RCLCxLedger const& prevLedger,
-    NetClock::duration const& closeResolution,
     ConsensusCloseTimes const& rawCloseTimes,
     ConsensusMode const& mode,
-    Json::Value&& consensusJson)
+    Json::Value&& consensusJson,
+    std::pair<CanonicalTxSet_t, Ledger_t>&& tb)
 {
     app_.getJobQueue().addJob(
         jtACCEPT,
         "acceptLedger",
-        [=, this, cj = std::move(consensusJson)]() mutable {
+        [=,
+         this,
+         cj = std::move(consensusJson),
+         txsBuilt = std::move(tb)]() mutable {
             // Note that no lock is held or acquired during this job.
             // This is because generic Consensus guarantees that once a ledger
             // is accepted, the consensus results and capture by reference state
             // will not change until startRound is called (which happens via
             // endConsensus).
-            this->doAccept(
-                result,
-                prevLedger,
-                closeResolution,
-                rawCloseTimes,
-                mode,
-                std::move(cj));
+            prepareOpenLedger(std::move(txsBuilt), result, rawCloseTimes, mode);
             this->app_.getOPs().endConsensus();
         });
 }
 
-void
-RCLConsensus::Adaptor::doAccept(
+std::pair<
+    RCLConsensus::Adaptor::CanonicalTxSet_t,
+    RCLConsensus::Adaptor::Ledger_t>
+RCLConsensus::Adaptor::buildAndValidate(
     Result const& result,
-    RCLCxLedger const& prevLedger,
-    NetClock::duration closeResolution,
-    ConsensusCloseTimes const& rawCloseTimes,
+    Ledger_t const& prevLedger,
+    NetClock::duration const& closeResolution,
     ConsensusMode const& mode,
     Json::Value&& consensusJson)
 {
@@ -497,12 +497,12 @@ RCLConsensus::Adaptor::doAccept(
         {
             retriableTxs.insert(
                 std::make_shared<STTx const>(SerialIter{item.slice()}));
-            JLOG(j_.debug()) << "    Tx: " << item.key();
+            JLOG(j_.trace()) << "    Tx: " << item.key();
         }
         catch (std::exception const& ex)
         {
             failed.insert(item.key());
-            JLOG(j_.warn())
+            JLOG(j_.trace())
                 << "    Tx: " << item.key() << " throws: " << ex.what();
         }
     }
@@ -579,6 +579,19 @@ RCLConsensus::Adaptor::doAccept(
     ledgerMaster_.consensusBuilt(
         built.ledger_, result.txns.id(), std::move(consensusJson));
 
+    return {retriableTxs, built};
+}
+
+void
+RCLConsensus::Adaptor::prepareOpenLedger(
+    std::pair<CanonicalTxSet_t, Ledger_t>&& txsBuilt,
+    Result const& result,
+    ConsensusCloseTimes const& rawCloseTimes,
+    ConsensusMode const& mode)
+{
+    auto& retriableTxs = txsBuilt.first;
+    auto const& built = txsBuilt.second;
+
     //-------------------------------------------------------------------------
     {
         // Apply disputed transactions that didn't get in
@@ -601,7 +614,7 @@ RCLConsensus::Adaptor::doAccept(
                 // we voted NO
                 try
                 {
-                    JLOG(j_.debug())
+                    JLOG(j_.trace())
                         << "Test applying disputed transaction that did"
                         << " not get in " << dispute.tx().id();
 
@@ -619,7 +632,7 @@ RCLConsensus::Adaptor::doAccept(
                 }
                 catch (std::exception const& ex)
                 {
-                    JLOG(j_.debug()) << "Failed to apply transaction we voted "
+                    JLOG(j_.trace()) << "Failed to apply transaction we voted "
                                         "NO on. Exception: "
                                      << ex.what();
                 }
@@ -669,6 +682,7 @@ RCLConsensus::Adaptor::doAccept(
     // we entered the round with the network,
     // see how close our close time is to other node's
     //  close time reports, and update our clock.
+    const bool consensusFail = result.state == ConsensusState::MovedOn;
     if ((mode == ConsensusMode::proposing ||
          mode == ConsensusMode::observing) &&
         !consensusFail)
@@ -889,12 +903,32 @@ RCLConsensus::Adaptor::onModeChange(ConsensusMode before, ConsensusMode after)
     mode_ = after;
 }
 
+bool
+RCLConsensus::Adaptor::retryAccept(
+    Ledger_t const& newLedger,
+    std::optional<std::chrono::time_point<std::chrono::steady_clock>>& start)
+    const
+{
+    static bool const standalone = ledgerMaster_.standalone();
+    auto const& validLedger = ledgerMaster_.getValidatedLedger();
+
+    return (app_.getOPs().isFull() && !standalone &&
+            (validLedger && (newLedger.id() != validLedger->info().hash) &&
+             (newLedger.seq() >= validLedger->info().seq))) &&
+        (!start ||
+         std::chrono::duration_cast<std::chrono::seconds>(
+             std::chrono::steady_clock::now() - *start)
+                 .count() < 5);
+}
+
+//-----------------------------------------------------------------------------
+
 Json::Value
 RCLConsensus::getJson(bool full) const
 {
     Json::Value ret;
     {
-        std::lock_guard _{mutex_};
+        std::lock_guard _{adaptor_.peekMutex()};
         ret = consensus_.getJson(full);
     }
     ret["validating"] = adaptor_.validating();
@@ -906,7 +940,7 @@ RCLConsensus::timerEntry(NetClock::time_point const& now)
 {
     try
     {
-        std::lock_guard _{mutex_};
+        std::lock_guard _{adaptor_.peekMutex()};
         consensus_.timerEntry(now);
     }
     catch (SHAMapMissingNode const& mn)
@@ -922,7 +956,7 @@ RCLConsensus::gotTxSet(NetClock::time_point const& now, RCLTxSet const& txSet)
 {
     try
     {
-        std::lock_guard _{mutex_};
+        std::lock_guard _{adaptor_.peekMutex()};
         consensus_.gotTxSet(now, txSet);
     }
     catch (SHAMapMissingNode const& mn)
@@ -940,7 +974,7 @@ RCLConsensus::simulate(
     NetClock::time_point const& now,
     std::optional<std::chrono::milliseconds> consensusDelay)
 {
-    std::lock_guard _{mutex_};
+    std::lock_guard _{adaptor_.peekMutex()};
     consensus_.simulate(now, consensusDelay);
 }
 
@@ -949,7 +983,7 @@ RCLConsensus::peerProposal(
     NetClock::time_point const& now,
     RCLCxPeerPos const& newProposal)
 {
-    std::lock_guard _{mutex_};
+    std::lock_guard _{adaptor_.peekMutex()};
     return consensus_.peerProposal(now, newProposal);
 }
 
@@ -1051,12 +1085,13 @@ RCLConsensus::startRound(
     hash_set<NodeID> const& nowUntrusted,
     hash_set<NodeID> const& nowTrusted)
 {
-    std::lock_guard _{mutex_};
+    std::lock_guard _{adaptor_.peekMutex()};
     consensus_.startRound(
         now,
         prevLgrId,
         prevLgr,
         nowUntrusted,
         adaptor_.preStartRound(prevLgr, nowTrusted));
 }
+
 }  // namespace ripple