20 #include <ripple/app/consensus/RCLConsensus.h>
21 #include <ripple/app/consensus/RCLValidations.h>
22 #include <ripple/app/ledger/BuildLedger.h>
23 #include <ripple/app/ledger/InboundLedgers.h>
24 #include <ripple/app/ledger/InboundTransactions.h>
25 #include <ripple/app/ledger/Ledger.h>
26 #include <ripple/app/ledger/LedgerMaster.h>
27 #include <ripple/app/ledger/LocalTxs.h>
28 #include <ripple/app/ledger/OpenLedger.h>
29 #include <ripple/app/misc/AmendmentTable.h>
30 #include <ripple/app/misc/HashRouter.h>
31 #include <ripple/app/misc/LoadFeeTrack.h>
32 #include <ripple/app/misc/NegativeUNLVote.h>
33 #include <ripple/app/misc/NetworkOPs.h>
34 #include <ripple/app/misc/TxQ.h>
35 #include <ripple/app/misc/ValidatorKeys.h>
36 #include <ripple/app/misc/ValidatorList.h>
37 #include <ripple/basics/random.h>
38 #include <ripple/beast/core/LexicalCast.h>
39 #include <ripple/consensus/LedgerTiming.h>
40 #include <ripple/nodestore/DatabaseShard.h>
41 #include <ripple/overlay/Overlay.h>
42 #include <ripple/overlay/predicates.h>
43 #include <ripple/protocol/BuildInfo.h>
44 #include <ripple/protocol/Feature.h>
45 #include <ripple/protocol/digest.h>
69 , consensus_(clock, adaptor_, journal)
83 , feeVote_(
std::move(feeVote))
86 , inboundTransactions_{inboundTransactions}
88 , validatorKeys_(validatorKeys)
94 , nUnlVote_(validatorKeys_.nodeID,
j_)
96 assert(valCookie_ != 0);
98 JLOG(
j_.
info()) <<
"Consensus engine started (cookie: " +
101 if (validatorKeys_.nodeID != beast::zero && validatorKeys_.keys)
105 JLOG(
j_.
info()) <<
"Validator identity: "
108 validatorKeys_.keys->masterPublicKey);
110 if (validatorKeys_.keys->masterPublicKey !=
111 validatorKeys_.keys->publicKey)
114 <<
"Validator ephemeral signing key: "
126 auto built = ledgerMaster_.getLedgerByHash(hash);
129 if (acquiringLedger_ != hash)
132 JLOG(
j_.
warn()) <<
"Need consensus ledger " << hash;
135 acquiringLedger_ = hash;
137 app_.getJobQueue().addJob(
138 jtADVANCE,
"getConsensusLedger", [
id = hash, &app = app_]() {
139 app.getInboundLedgers().
acquire(
146 assert(!built->open() && built->isImmutable());
147 assert(built->info().hash == hash);
150 inboundTransactions_.newRound(built->info().seq);
158 protocol::TMProposeSet prop;
162 prop.set_proposeseq(
proposal.proposeSeq());
163 prop.set_closetime(
proposal.closeTime().time_since_epoch().count());
165 prop.set_currenttxhash(
167 prop.set_previousledger(
171 prop.set_nodepubkey(pk.data(), pk.size());
174 prop.set_signature(sig.data(), sig.size());
183 if (app_.getHashRouter().shouldRelay(tx.
id()))
185 JLOG(
j_.
debug()) <<
"Relaying disputed tx " << tx.
id();
186 auto const slice = tx.
tx_->slice();
187 protocol::TMTransaction msg;
188 msg.set_rawtransaction(slice.data(), slice.size());
189 msg.set_status(protocol::tsNEW);
190 msg.set_receivetimestamp(
191 app_.timeKeeper().now().time_since_epoch().count());
193 app_.overlay().relay(tx.
id(), msg, skip);
197 JLOG(
j_.
debug()) <<
"Not relaying disputed tx " << tx.
id();
203 JLOG(
j_.
trace()) << (
proposal.isBowOut() ?
"We bow out: " :
"We propose: ")
207 protocol::TMProposeSet prop;
209 prop.set_currenttxhash(
211 prop.set_previousledger(
213 prop.set_proposeseq(
proposal.proposeSeq());
214 prop.set_closetime(
proposal.closeTime().time_since_epoch().count());
216 if (!validatorKeys_.keys)
218 JLOG(
j_.
warn()) <<
"RCLConsensus::Adaptor::propose: ValidatorKeys "
223 auto const& keys = *validatorKeys_.keys;
225 prop.set_nodepubkey(keys.publicKey.data(), keys.publicKey.size());
230 prop.set_signature(sig.data(), sig.size());
240 app_.getHashRouter().addSuppression(suppression);
242 app_.overlay().broadcast(prop);
248 inboundTransactions_.giveSet(txns.
id(), txns.
map_,
false);
254 if (
auto txns = inboundTransactions_.getSet(setId,
true))
264 return !app_.openLedger().empty();
270 return app_.getValidations().numTrustedForLedger(h);
292 ledgerMaster_.getValidLedgerIndex());
294 if (netLgr != ledgerID)
297 app_.getOPs().consensusViewChange();
314 notify(protocol::neCLOSING_LEDGER, ledger, !wrongLCL);
316 auto const& prevLedger = ledger.ledger_;
318 ledgerMaster_.applyHeldTransactions();
320 ledgerMaster_.setBuildingLedger(prevLedger->info().seq + 1);
322 auto initialLedger = app_.openLedger().current();
326 initialSet->setUnbacked();
329 for (
auto const& tx : initialLedger->txs)
331 JLOG(
j_.
trace()) <<
"Adding open ledger TX "
332 << tx.first->getTransactionID();
341 if (app_.config().standalone() || (
proposing && !wrongLCL))
343 if (prevLedger->isFlagLedger())
347 auto validations = app_.validators().negativeUNLFilter(
348 app_.getValidations().getTrustedForLedger(
349 prevLedger->info().parentHash, prevLedger->seq() - 1));
350 if (validations.size() >= app_.validators().quorum())
352 feeVote_->doVoting(prevLedger, validations, initialSet);
353 app_.getAmendmentTable().doVoting(
354 prevLedger, validations, initialSet);
358 prevLedger->isVotingLedger() &&
366 app_.validators().getTrustedMasterKeys(),
367 app_.getValidations(),
373 initialSet = initialSet->snapShot(
false);
377 LedgerIndex const seq = prevLedger->info().seq + 1;
380 initialSet->visitLeaves(
382 seq](boost::intrusive_ptr<SHAMapItem const>
const& item) {
386 censorshipDetector_.propose(std::move(proposed));
390 auto const setHash = initialSet->getHash().as_uint256();
393 std::move(initialSet),
395 initialLedger->info().parentHash,
399 app_.timeKeeper().closeTime(),
400 validatorKeys_.nodeID}};
418 std::move(consensusJson));
430 app_.getJobQueue().addJob(
433 [=,
this, cj = std::move(consensusJson)]()
mutable {
446 this->app_.getOPs().endConsensus();
462 bool closeTimeCorrect;
473 using namespace std::chrono_literals;
474 consensusCloseTime = prevLedger.
closeTime() + 1s;
475 closeTimeCorrect =
false;
481 consensusCloseTime, closeResolution, prevLedger.
closeTime());
482 closeTimeCorrect =
true;
486 <<
" val=" << (validating_ ?
"yes" :
"no")
487 <<
" corLCL=" << (haveCorrectLCL ?
"yes" :
"no")
488 <<
" fail=" << (consensusFail ?
"yes" :
"no");
489 JLOG(
j_.
debug()) <<
"Report: Prev = " << prevLedger.
id() <<
":"
501 JLOG(
j_.
debug()) <<
"Building canonical tx set: " << retriableTxs.key();
503 for (
auto const& item : *result.
txns.map_)
508 std::make_shared<STTx const>(
SerialIter{item.slice()}));
509 JLOG(
j_.
debug()) <<
" Tx: " << item.key();
513 failed.
insert(item.key());
515 <<
" Tx: " << item.key() <<
" throws: " << ex.
what();
519 auto built = buildLCL(
528 auto const newLCLHash = built.id();
529 JLOG(
j_.
debug()) <<
"Built ledger #" << built.seq() <<
": " << newLCLHash;
532 notify(protocol::neACCEPTED_LEDGER, built, haveCorrectLCL);
541 result.
txns.map_->visitLeaves(
542 [&
accepted](boost::intrusive_ptr<SHAMapItem const>
const& item) {
547 for (
auto const& r : retriableTxs)
548 failed.
insert(r.first.getTXID());
550 censorshipDetector_.check(
553 j = app_.journal(
"CensorshipDetector"),
555 if (failed.count(id))
558 auto const wait = curr - seq;
560 if (wait && (wait % censorshipWarnInternal == 0))
562 std::ostringstream ss;
563 ss <<
"Potential Censorship: Eligible tx " << id
564 <<
", which we are tracking since ledger " << seq
565 <<
" has not been included as of ledger " << curr <<
".";
567 JLOG(j.warn()) << ss.str();
575 validating_ = ledgerMaster_.isCompatible(
576 *built.ledger_,
j_.
warn(),
"Not validating");
578 if (validating_ && !consensusFail &&
579 app_.getValidations().canValidateSeq(built.seq()))
581 validate(built, result.txns, proposing);
582 JLOG(
j_.
info()) <<
"CNF Val " << newLCLHash;
585 JLOG(
j_.
info()) <<
"CNF buildLCL " << newLCLHash;
588 ledgerMaster_.consensusBuilt(
589 built.ledger_, result.txns.id(), std::move(consensusJson));
604 bool anyDisputes =
false;
605 for (
auto const& [_, dispute] : result.disputes)
608 if (!dispute.getOurVote())
614 <<
"Test applying disputed transaction that did"
615 <<
" not get in " << dispute.tx().id();
618 auto txn = std::make_shared<STTx const>(sit);
625 retriableTxs.insert(txn);
631 JLOG(
j_.
debug()) <<
"Failed to apply transaction we voted "
643 auto const lastVal = ledgerMaster_.getValidatedLedger();
648 rules.
emplace(app_.config().features);
649 app_.openLedger().accept(
653 localTxs_.getTxSet(),
660 return app_.getTxQ().accept(app_, view);
665 app_.getOPs().reportFeeChange();
670 ledgerMaster_.switchLCL(built.ledger_);
673 assert(ledgerMaster_.getClosedLedger()->info().hash == built.id());
674 assert(app_.openLedger().current()->info().parentHash == built.id());
685 auto closeTime = rawCloseTimes.self;
687 JLOG(
j_.
info()) <<
"We closed at "
688 << closeTime.time_since_epoch().count();
690 usec64_t closeTotal =
691 std::chrono::duration_cast<usec64_t>(closeTime.time_since_epoch());
694 for (
auto const& [t, v] : rawCloseTimes.peers)
700 std::chrono::duration_cast<usec64_t>(t.time_since_epoch()) * v;
703 closeTotal += usec64_t(closeCount / 2);
704 closeTotal /= closeCount;
709 auto offset = time_point{closeTotal} -
710 std::chrono::time_point_cast<duration>(closeTime);
711 JLOG(
j_.
info()) <<
"Our close offset is estimated at " << offset.count()
712 <<
" (" << closeCount <<
")";
714 app_.timeKeeper().adjustCloseTime(offset);
720 protocol::NodeEvent ne,
724 protocol::TMStatusChange s;
727 s.set_newevent(protocol::neLOST_SYNC);
731 s.set_ledgerseq(ledger.
seq());
732 s.set_networktime(app_.timeKeeper().now().time_since_epoch().count());
733 s.set_ledgerhashprevious(
740 if (!ledgerMaster_.getFullValidatedRange(uMin, uMax))
748 uMin =
std::max(uMin, ledgerMaster_.getEarliestFetch());
750 s.set_firstseq(uMin);
752 app_.overlay().foreach(
753 send_always(std::make_shared<Message>(s, protocol::mtSTATUS_CHANGE)));
754 JLOG(
j_.
trace()) <<
"send status change to peer";
762 bool closeTimeCorrect,
768 if (
auto const replayData = ledgerMaster_.releaseReplay())
770 assert(replayData->parent()->info().hash == previousLedger.
id());
785 using namespace std::chrono_literals;
786 app_.getTxQ().processClosedLedger(app_, *built, roundTime > 5s);
789 if (ledgerMaster_.storeLedger(built))
790 JLOG(
j_.
debug()) <<
"Consensus built ledger we already had";
791 else if (app_.getInboundLedgers().find(built->info().hash))
792 JLOG(
j_.
debug()) <<
"Consensus built ledger we were acquiring";
794 JLOG(
j_.
debug()) <<
"Consensus built new ledger";
804 using namespace std::chrono_literals;
806 auto validationTime = app_.timeKeeper().closeTime();
807 if (validationTime <= lastValidationTime_)
808 validationTime = lastValidationTime_ + 1s;
809 lastValidationTime_ = validationTime;
811 if (!validatorKeys_.keys)
813 JLOG(
j_.
warn()) <<
"RCLConsensus::Adaptor::validate: ValidatorKeys "
818 auto const& keys = *validatorKeys_.keys;
820 auto v = std::make_shared<STValidation>(
824 validatorKeys_.nodeID,
826 v.setFieldH256(sfLedgerHash, ledger.id());
827 v.setFieldH256(sfConsensusHash, txns.id());
829 v.setFieldU32(sfLedgerSequence, ledger.seq());
832 v.setFlag(vfFullValidation);
834 if (ledger.ledger_->rules().enabled(featureHardenedValidations))
839 if (auto const vl = ledgerMaster_.getValidatedLedger())
840 v.setFieldH256(sfValidatedHash, vl->info().hash);
842 v.setFieldU64(sfCookie, valCookie_);
845 if (ledger.ledger_->isVotingLedger())
847 sfServerVersion, BuildInfo::getEncodedVersion());
852 auto const& ft = app_.getFeeTrack();
853 auto const fee = std::max(ft.getLocalFee(), ft.getClusterFee());
854 if (fee > ft.getLoadBase())
855 v.setFieldU32(sfLoadFee, fee);
860 if (ledger.
ledger_->isVotingLedger())
863 feeVote_->doValidation(
864 ledger.ledger_->fees(), ledger.ledger_->rules(), v);
869 auto const amendments = app_.getAmendmentTable().doValidation(
870 getEnabledAmendments(*ledger.ledger_));
872 if (!amendments.empty())
874 sfAmendments, STVector256(sfAmendments, amendments));
878 auto const serialized = v->getSerialized();
886 protocol::TMValidation val;
887 val.set_validation(serialized.data(), serialized.size());
888 app_.overlay().broadcast(val);
891 app_.getOPs().pubValidation(v);
897 JLOG(
j_.
info()) <<
"Consensus mode change before=" <<
to_string(before)
905 censorshipDetector_.reset();
933 JLOG(
j_.
error()) <<
"During consensus timerEntry: " << mn.
what();
949 JLOG(
j_.
error()) <<
"During consensus gotTxSet: " << mn.
what();
971 return consensus_.peerProposal(now, newProposal);
981 validating_ = validatorKeys_.keys &&
982 prevLgr.
seq() >= app_.getMaxDisallowedLedger() &&
983 !app_.getOPs().isBlocked();
987 if (validating_ && !app_.config().standalone() && app_.validators().count())
989 auto const when = app_.validators().expires();
991 if (!when || *when < app_.timeKeeper().now())
993 JLOG(
j_.
error()) <<
"Voluntarily bowing out of consensus process "
994 "because of an expired validator list.";
1003 JLOG(
j_.
info()) <<
"Entering consensus process, validating, synced="
1004 << (synced ?
"yes" :
"no");
1009 JLOG(
j_.
info()) <<
"Entering consensus process, watching, synced="
1010 << (synced ?
"yes" :
"no");
1014 inboundTransactions_.newRound(prevLgr.
seq());
1018 !nowTrusted.
empty())
1019 nUnlVote_.newValidators(prevLgr.
seq() + 1, nowTrusted);
1022 return validating_ && synced;
1028 return ledgerMaster_.haveValidated();
1034 return ledgerMaster_.getValidLedgerIndex();
1040 return app_.validators().getQuorumKeys();
1048 return app_.getValidations().laggards(seq, trustedKeys);
1054 return validatorKeys_.keys.has_value();
1060 if (!positions && app_.getOPs().isFull())