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)
89 , valCookie_{rand_int<std::uint64_t>(
92 , nUnlVote_(validatorKeys_.nodeID,
j_)
94 assert(valCookie_ != 0);
96 JLOG(
j_.
info()) <<
"Consensus engine started (cookie: " +
99 if (validatorKeys_.nodeID != beast::zero)
103 JLOG(
j_.
info()) <<
"Validator identity: "
106 validatorKeys_.masterPublicKey);
108 if (validatorKeys_.masterPublicKey != validatorKeys_.publicKey)
111 <<
"Validator ephemeral signing key: "
122 auto built = ledgerMaster_.getLedgerByHash(hash);
125 if (acquiringLedger_ != hash)
128 JLOG(
j_.
warn()) <<
"Need consensus ledger " << hash;
131 acquiringLedger_ = hash;
133 app_.getJobQueue().addJob(
134 jtADVANCE,
"getConsensusLedger", [
id = hash, &app = app_]() {
135 app.getInboundLedgers().
acquire(
142 assert(!built->open() && built->isImmutable());
143 assert(built->info().hash == hash);
146 inboundTransactions_.newRound(built->info().seq);
154 protocol::TMProposeSet prop;
158 prop.set_proposeseq(
proposal.proposeSeq());
159 prop.set_closetime(
proposal.closeTime().time_since_epoch().count());
161 prop.set_currenttxhash(
163 prop.set_previousledger(
167 prop.set_nodepubkey(pk.data(), pk.size());
170 prop.set_signature(sig.data(), sig.size());
179 if (app_.getHashRouter().shouldRelay(tx.
id()))
181 JLOG(
j_.
debug()) <<
"Relaying disputed tx " << tx.
id();
183 protocol::TMTransaction msg;
184 msg.set_rawtransaction(slice.data(), slice.size());
185 msg.set_status(protocol::tsNEW);
186 msg.set_receivetimestamp(
187 app_.timeKeeper().now().time_since_epoch().count());
189 app_.overlay().relay(tx.
id(), msg, skip);
193 JLOG(
j_.
debug()) <<
"Not relaying disputed tx " << tx.
id();
199 JLOG(
j_.
trace()) << (
proposal.isBowOut() ?
"We bow out: " :
"We propose: ")
203 protocol::TMProposeSet prop;
205 prop.set_currenttxhash(
207 prop.set_previousledger(
209 prop.set_proposeseq(
proposal.proposeSeq());
210 prop.set_closetime(
proposal.closeTime().time_since_epoch().count());
212 validatorKeys_.publicKey.data(), validatorKeys_.publicKey.size());
217 proposal.closeTime().time_since_epoch().count(),
222 validatorKeys_.publicKey, validatorKeys_.secretKey, signingHash);
224 prop.set_signature(sig.data(), sig.size());
231 validatorKeys_.publicKey,
234 app_.getHashRouter().addSuppression(suppression);
236 app_.overlay().broadcast(prop);
242 inboundTransactions_.giveSet(txns.
id(), txns.
map_,
false);
248 if (
auto txns = inboundTransactions_.getSet(setId,
true))
258 return !app_.openLedger().empty();
264 return app_.getValidations().numTrustedForLedger(h);
286 ledgerMaster_.getValidLedgerIndex());
288 if (netLgr != ledgerID)
291 app_.getOPs().consensusViewChange();
308 notify(protocol::neCLOSING_LEDGER, ledger, !wrongLCL);
310 auto const& prevLedger = ledger.ledger_;
312 ledgerMaster_.applyHeldTransactions();
314 ledgerMaster_.setBuildingLedger(prevLedger->info().seq + 1);
316 auto initialLedger = app_.openLedger().current();
320 initialSet->setUnbacked();
323 for (
auto const& tx : initialLedger->txs)
325 JLOG(
j_.
trace()) <<
"Adding open ledger TX "
326 << tx.first->getTransactionID();
335 if (app_.config().standalone() || (
proposing && !wrongLCL))
337 if (prevLedger->isFlagLedger())
341 auto validations = app_.validators().negativeUNLFilter(
342 app_.getValidations().getTrustedForLedger(
343 prevLedger->info().parentHash));
344 if (validations.size() >= app_.validators().quorum())
346 feeVote_->doVoting(prevLedger, validations, initialSet);
347 app_.getAmendmentTable().doVoting(
348 prevLedger, validations, initialSet);
352 prevLedger->isVotingLedger() &&
360 app_.validators().getTrustedMasterKeys(),
361 app_.getValidations(),
367 initialSet = initialSet->snapShot(
false);
371 LedgerIndex const seq = prevLedger->info().seq + 1;
374 initialSet->visitLeaves(
379 censorshipDetector_.propose(std::move(proposed));
383 auto const setHash = initialSet->getHash().as_uint256();
386 std::move(initialSet),
388 initialLedger->info().parentHash,
392 app_.timeKeeper().closeTime(),
393 validatorKeys_.nodeID}};
411 std::move(consensusJson));
423 app_.getJobQueue().addJob(
424 jtACCEPT,
"acceptLedger", [=, cj = std::move(consensusJson)]()
mutable {
437 this->app_.getOPs().endConsensus();
453 bool closeTimeCorrect;
464 using namespace std::chrono_literals;
465 consensusCloseTime = prevLedger.
closeTime() + 1s;
466 closeTimeCorrect =
false;
472 consensusCloseTime, closeResolution, prevLedger.
closeTime());
473 closeTimeCorrect =
true;
477 <<
" val=" << (validating_ ?
"yes" :
"no")
478 <<
" corLCL=" << (haveCorrectLCL ?
"yes" :
"no")
479 <<
" fail=" << (consensusFail ?
"yes" :
"no");
480 JLOG(
j_.
debug()) <<
"Report: Prev = " << prevLedger.
id() <<
":"
492 JLOG(
j_.
debug()) <<
"Building canonical tx set: " << retriableTxs.key();
494 for (
auto const& item : *result.
txns.map_)
499 std::make_shared<STTx const>(
SerialIter{item.slice()}));
500 JLOG(
j_.
debug()) <<
" Tx: " << item.key();
504 failed.
insert(item.key());
505 JLOG(
j_.
warn()) <<
" Tx: " << item.key() <<
" throws!";
509 auto built = buildLCL(
518 auto const newLCLHash = built.id();
519 JLOG(
j_.
debug()) <<
"Built ledger #" << built.seq() <<
": " << newLCLHash;
522 notify(protocol::neACCEPTED_LEDGER, built, haveCorrectLCL);
531 result.
txns.map_->visitLeaves(
537 for (
auto const& r : retriableTxs)
538 failed.
insert(r.first.getTXID());
540 censorshipDetector_.check(
543 j = app_.journal(
"CensorshipDetector"),
545 if (failed.count(id))
548 auto const wait = curr - seq;
550 if (wait && (wait % censorshipWarnInternal == 0))
552 std::ostringstream ss;
553 ss <<
"Potential Censorship: Eligible tx " << id
554 <<
", which we are tracking since ledger " << seq
555 <<
" has not been included as of ledger " << curr <<
".";
557 JLOG(j.warn()) << ss.str();
565 validating_ = ledgerMaster_.isCompatible(
566 *built.ledger_,
j_.
warn(),
"Not validating");
568 if (validating_ && !consensusFail &&
569 app_.getValidations().canValidateSeq(built.seq()))
571 validate(built, result.txns, proposing);
572 JLOG(
j_.
info()) <<
"CNF Val " << newLCLHash;
575 JLOG(
j_.
info()) <<
"CNF buildLCL " << newLCLHash;
578 ledgerMaster_.consensusBuilt(
579 built.ledger_, result.txns.id(), std::move(consensusJson));
594 bool anyDisputes =
false;
595 for (
auto const& [_, dispute] : result.disputes)
598 if (!dispute.getOurVote())
604 <<
"Test applying disputed transaction that did"
605 <<
" not get in " << dispute.tx().id();
608 auto txn = std::make_shared<STTx const>(sit);
615 retriableTxs.insert(txn);
622 <<
"Failed to apply transaction we voted NO on";
632 auto const lastVal = ledgerMaster_.getValidatedLedger();
635 rules.
emplace(*lastVal, app_.config().features);
637 rules.
emplace(app_.config().features);
638 app_.openLedger().accept(
642 localTxs_.getTxSet(),
649 return app_.getTxQ().accept(app_, view);
654 app_.getOPs().reportFeeChange();
659 ledgerMaster_.switchLCL(built.ledger_);
662 assert(ledgerMaster_.getClosedLedger()->info().hash == built.id());
663 assert(app_.openLedger().current()->info().parentHash == built.id());
674 auto closeTime = rawCloseTimes.self;
676 JLOG(
j_.
info()) <<
"We closed at "
677 << closeTime.time_since_epoch().count();
679 usec64_t closeTotal =
680 std::chrono::duration_cast<usec64_t>(closeTime.time_since_epoch());
683 for (
auto const& [t, v] : rawCloseTimes.peers)
689 std::chrono::duration_cast<usec64_t>(t.time_since_epoch()) * v;
692 closeTotal += usec64_t(closeCount / 2);
693 closeTotal /= closeCount;
698 auto offset = time_point{closeTotal} -
699 std::chrono::time_point_cast<duration>(closeTime);
700 JLOG(
j_.
info()) <<
"Our close offset is estimated at " << offset.count()
701 <<
" (" << closeCount <<
")";
703 app_.timeKeeper().adjustCloseTime(offset);
709 protocol::NodeEvent ne,
713 protocol::TMStatusChange s;
716 s.set_newevent(protocol::neLOST_SYNC);
720 s.set_ledgerseq(ledger.
seq());
721 s.set_networktime(app_.timeKeeper().now().time_since_epoch().count());
722 s.set_ledgerhashprevious(
729 if (!ledgerMaster_.getFullValidatedRange(uMin, uMax))
737 uMin =
std::max(uMin, ledgerMaster_.getEarliestFetch());
739 s.set_firstseq(uMin);
741 app_.overlay().foreach(
742 send_always(std::make_shared<Message>(s, protocol::mtSTATUS_CHANGE)));
743 JLOG(
j_.
trace()) <<
"send status change to peer";
751 bool closeTimeCorrect,
757 if (
auto const replayData = ledgerMaster_.releaseReplay())
759 assert(replayData->parent()->info().hash == previousLedger.
id());
774 using namespace std::chrono_literals;
775 app_.getTxQ().processClosedLedger(app_, *built, roundTime > 5s);
778 if (ledgerMaster_.storeLedger(built))
779 JLOG(
j_.
debug()) <<
"Consensus built ledger we already had";
780 else if (app_.getInboundLedgers().find(built->info().hash))
781 JLOG(
j_.
debug()) <<
"Consensus built ledger we were acquiring";
783 JLOG(
j_.
debug()) <<
"Consensus built new ledger";
793 using namespace std::chrono_literals;
795 auto validationTime = app_.timeKeeper().closeTime();
796 if (validationTime <= lastValidationTime_)
797 validationTime = lastValidationTime_ + 1s;
798 lastValidationTime_ = validationTime;
800 auto v = std::make_shared<STValidation>(
802 validatorKeys_.publicKey,
803 validatorKeys_.secretKey,
804 validatorKeys_.nodeID,
806 v.setFieldH256(sfLedgerHash, ledger.id());
807 v.setFieldH256(sfConsensusHash, txns.id());
809 v.setFieldU32(sfLedgerSequence, ledger.seq());
812 v.setFlag(vfFullValidation);
814 if (ledger.ledger_->rules().enabled(featureHardenedValidations))
819 if (auto const vl = ledgerMaster_.getValidatedLedger())
820 v.setFieldH256(sfValidatedHash, vl->info().hash);
822 v.setFieldU64(sfCookie, valCookie_);
825 if (ledger.ledger_->isVotingLedger())
827 sfServerVersion, BuildInfo::getEncodedVersion());
832 auto const& ft = app_.getFeeTrack();
833 auto const fee = std::max(ft.getLocalFee(), ft.getClusterFee());
834 if (fee > ft.getLoadBase())
835 v.setFieldU32(sfLoadFee, fee);
840 if (ledger.
ledger_->isVotingLedger())
843 feeVote_->doValidation(ledger.ledger_->fees(), v);
848 auto const amendments = app_.getAmendmentTable().doValidation(
849 getEnabledAmendments(*ledger.ledger_));
851 if (!amendments.empty())
853 sfAmendments, STVector256(sfAmendments, amendments));
857 auto const serialized = v->getSerialized();
865 protocol::TMValidation val;
866 val.set_validation(serialized.data(), serialized.size());
867 app_.overlay().broadcast(val);
870 app_.getOPs().pubValidation(v);
876 JLOG(
j_.
info()) <<
"Consensus mode change before=" <<
to_string(before)
884 censorshipDetector_.reset();
912 JLOG(
j_.
error()) <<
"During consensus timerEntry: " << mn.
what();
928 JLOG(
j_.
error()) <<
"During consensus gotTxSet: " << mn.
what();
950 return consensus_.peerProposal(now, newProposal);
960 validating_ = validatorKeys_.publicKey.size() != 0 &&
961 prevLgr.
seq() >= app_.getMaxDisallowedLedger() &&
962 !app_.getOPs().isBlocked();
966 if (validating_ && !app_.config().standalone() && app_.validators().count())
968 auto const when = app_.validators().expires();
970 if (!when || *when < app_.timeKeeper().now())
972 JLOG(
j_.
error()) <<
"Voluntarily bowing out of consensus process "
973 "because of an expired validator list.";
982 JLOG(
j_.
info()) <<
"Entering consensus process, validating, synced="
983 << (synced ?
"yes" :
"no");
988 JLOG(
j_.
info()) <<
"Entering consensus process, watching, synced="
989 << (synced ?
"yes" :
"no");
993 inboundTransactions_.newRound(prevLgr.
seq());
998 nUnlVote_.newValidators(prevLgr.
seq() + 1, nowTrusted);
1001 return validating_ && synced;
1007 return ledgerMaster_.haveValidated();
1013 return ledgerMaster_.getValidLedgerIndex();
1019 return app_.validators().getQuorumKeys();
1027 return app_.getValidations().laggards(seq, trustedKeys);
1033 return !validatorKeys_.publicKey.empty();
1039 if (!positions && app_.getOPs().isFull())