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(
135 "getConsensusLedger",
136 [
id = hash, &app = app_](
Job&) {
137 app.getInboundLedgers().
acquire(
144 assert(!built->open() && built->isImmutable());
145 assert(built->info().hash == hash);
148 inboundTransactions_.newRound(built->info().seq);
156 protocol::TMProposeSet prop;
160 prop.set_proposeseq(
proposal.proposeSeq());
161 prop.set_closetime(
proposal.closeTime().time_since_epoch().count());
163 prop.set_currenttxhash(
165 prop.set_previousledger(
169 prop.set_nodepubkey(pk.data(), pk.size());
172 prop.set_signature(sig.data(), sig.size());
181 if (app_.getHashRouter().shouldRelay(tx.
id()))
183 JLOG(
j_.
debug()) <<
"Relaying disputed tx " << tx.
id();
185 protocol::TMTransaction msg;
186 msg.set_rawtransaction(slice.data(), slice.size());
187 msg.set_status(protocol::tsNEW);
188 msg.set_receivetimestamp(
189 app_.timeKeeper().now().time_since_epoch().count());
191 app_.overlay().relay(tx.
id(), msg, skip);
195 JLOG(
j_.
debug()) <<
"Not relaying disputed tx " << tx.
id();
201 JLOG(
j_.
trace()) << (
proposal.isBowOut() ?
"We bow out: " :
"We propose: ")
205 protocol::TMProposeSet prop;
207 prop.set_currenttxhash(
209 prop.set_previousledger(
211 prop.set_proposeseq(
proposal.proposeSeq());
212 prop.set_closetime(
proposal.closeTime().time_since_epoch().count());
214 validatorKeys_.publicKey.data(), validatorKeys_.publicKey.size());
219 proposal.closeTime().time_since_epoch().count(),
224 validatorKeys_.publicKey, validatorKeys_.secretKey, signingHash);
226 prop.set_signature(sig.data(), sig.size());
233 validatorKeys_.publicKey,
236 app_.getHashRouter().addSuppression(suppression);
238 app_.overlay().broadcast(prop);
244 inboundTransactions_.giveSet(txns.
id(), txns.
map_,
false);
250 if (
auto txns = inboundTransactions_.getSet(setId,
true))
260 return !app_.openLedger().empty();
266 return app_.getValidations().numTrustedForLedger(h);
288 ledgerMaster_.getValidLedgerIndex());
290 if (netLgr != ledgerID)
293 app_.getOPs().consensusViewChange();
310 notify(protocol::neCLOSING_LEDGER, ledger, !wrongLCL);
312 auto const& prevLedger = ledger.ledger_;
314 ledgerMaster_.applyHeldTransactions();
316 ledgerMaster_.setBuildingLedger(prevLedger->info().seq + 1);
318 auto initialLedger = app_.openLedger().current();
322 initialSet->setUnbacked();
325 for (
auto const& tx : initialLedger->txs)
327 JLOG(
j_.
trace()) <<
"Adding open ledger TX "
328 << tx.first->getTransactionID();
337 if (app_.config().standalone() || (
proposing && !wrongLCL))
339 if (prevLedger->isFlagLedger())
343 auto validations = app_.validators().negativeUNLFilter(
344 app_.getValidations().getTrustedForLedger(
345 prevLedger->info().parentHash));
346 if (validations.size() >= app_.validators().quorum())
348 feeVote_->doVoting(prevLedger, validations, initialSet);
349 app_.getAmendmentTable().doVoting(
350 prevLedger, validations, initialSet);
354 prevLedger->isVotingLedger() &&
362 app_.validators().getTrustedMasterKeys(),
363 app_.getValidations(),
369 initialSet = initialSet->snapShot(
false);
373 LedgerIndex const seq = prevLedger->info().seq + 1;
376 initialSet->visitLeaves(
381 censorshipDetector_.propose(std::move(proposed));
385 auto const setHash = initialSet->getHash().as_uint256();
388 std::move(initialSet),
390 initialLedger->info().parentHash,
394 app_.timeKeeper().closeTime(),
395 validatorKeys_.nodeID}};
413 std::move(consensusJson));
425 app_.getJobQueue().addJob(
428 [=, cj = std::move(consensusJson)](
auto&)
mutable {
441 this->app_.getOPs().endConsensus();
457 bool closeTimeCorrect;
468 using namespace std::chrono_literals;
469 consensusCloseTime = prevLedger.
closeTime() + 1s;
470 closeTimeCorrect =
false;
476 consensusCloseTime, closeResolution, prevLedger.
closeTime());
477 closeTimeCorrect =
true;
481 <<
" val=" << (validating_ ?
"yes" :
"no")
482 <<
" corLCL=" << (haveCorrectLCL ?
"yes" :
"no")
483 <<
" fail=" << (consensusFail ?
"yes" :
"no");
484 JLOG(
j_.
debug()) <<
"Report: Prev = " << prevLedger.
id() <<
":"
496 JLOG(
j_.
debug()) <<
"Building canonical tx set: " << retriableTxs.key();
498 for (
auto const& item : *result.
txns.map_)
503 std::make_shared<STTx const>(
SerialIter{item.slice()}));
504 JLOG(
j_.
debug()) <<
" Tx: " << item.key();
508 failed.
insert(item.key());
509 JLOG(
j_.
warn()) <<
" Tx: " << item.key() <<
" throws!";
513 auto built = buildLCL(
522 auto const newLCLHash = built.id();
523 JLOG(
j_.
debug()) <<
"Built ledger #" << built.seq() <<
": " << newLCLHash;
526 notify(protocol::neACCEPTED_LEDGER, built, haveCorrectLCL);
535 result.
txns.map_->visitLeaves(
541 for (
auto const& r : retriableTxs)
542 failed.
insert(r.first.getTXID());
544 censorshipDetector_.check(
547 j = app_.journal(
"CensorshipDetector"),
549 if (failed.count(id))
552 auto const wait = curr - seq;
554 if (wait && (wait % censorshipWarnInternal == 0))
556 std::ostringstream ss;
557 ss <<
"Potential Censorship: Eligible tx " << id
558 <<
", which we are tracking since ledger " << seq
559 <<
" has not been included as of ledger " << curr <<
".";
561 JLOG(j.warn()) << ss.str();
569 validating_ = ledgerMaster_.isCompatible(
570 *built.ledger_,
j_.
warn(),
"Not validating");
572 if (validating_ && !consensusFail &&
573 app_.getValidations().canValidateSeq(built.seq()))
575 validate(built, result.txns, proposing);
576 JLOG(
j_.
info()) <<
"CNF Val " << newLCLHash;
579 JLOG(
j_.
info()) <<
"CNF buildLCL " << newLCLHash;
582 ledgerMaster_.consensusBuilt(
583 built.ledger_, result.txns.id(), std::move(consensusJson));
598 bool anyDisputes =
false;
599 for (
auto const& [_, dispute] : result.disputes)
602 if (!dispute.getOurVote())
608 <<
"Test applying disputed transaction that did"
609 <<
" not get in " << dispute.tx().id();
612 auto txn = std::make_shared<STTx const>(sit);
619 retriableTxs.insert(txn);
626 <<
"Failed to apply transaction we voted NO on";
636 auto const lastVal = ledgerMaster_.getValidatedLedger();
639 rules.
emplace(*lastVal, app_.config().features);
641 rules.
emplace(app_.config().features);
642 app_.openLedger().accept(
646 localTxs_.getTxSet(),
653 return app_.getTxQ().accept(app_, view);
658 app_.getOPs().reportFeeChange();
663 ledgerMaster_.switchLCL(built.ledger_);
666 assert(ledgerMaster_.getClosedLedger()->info().hash == built.id());
667 assert(app_.openLedger().current()->info().parentHash == built.id());
678 auto closeTime = rawCloseTimes.self;
680 JLOG(
j_.
info()) <<
"We closed at "
681 << closeTime.time_since_epoch().count();
683 usec64_t closeTotal =
684 std::chrono::duration_cast<usec64_t>(closeTime.time_since_epoch());
687 for (
auto const& [t, v] : rawCloseTimes.peers)
693 std::chrono::duration_cast<usec64_t>(t.time_since_epoch()) * v;
696 closeTotal += usec64_t(closeCount / 2);
697 closeTotal /= closeCount;
702 auto offset = time_point{closeTotal} -
703 std::chrono::time_point_cast<duration>(closeTime);
704 JLOG(
j_.
info()) <<
"Our close offset is estimated at " << offset.count()
705 <<
" (" << closeCount <<
")";
707 app_.timeKeeper().adjustCloseTime(offset);
713 protocol::NodeEvent ne,
717 protocol::TMStatusChange s;
720 s.set_newevent(protocol::neLOST_SYNC);
724 s.set_ledgerseq(ledger.
seq());
725 s.set_networktime(app_.timeKeeper().now().time_since_epoch().count());
726 s.set_ledgerhashprevious(
733 if (!ledgerMaster_.getFullValidatedRange(uMin, uMax))
741 uMin =
std::max(uMin, ledgerMaster_.getEarliestFetch());
743 s.set_firstseq(uMin);
745 app_.overlay().foreach(
746 send_always(std::make_shared<Message>(s, protocol::mtSTATUS_CHANGE)));
747 JLOG(
j_.
trace()) <<
"send status change to peer";
755 bool closeTimeCorrect,
761 if (
auto const replayData = ledgerMaster_.releaseReplay())
763 assert(replayData->parent()->info().hash == previousLedger.
id());
778 using namespace std::chrono_literals;
779 app_.getTxQ().processClosedLedger(app_, *built, roundTime > 5s);
782 if (ledgerMaster_.storeLedger(built))
783 JLOG(
j_.
debug()) <<
"Consensus built ledger we already had";
784 else if (app_.getInboundLedgers().find(built->info().hash))
785 JLOG(
j_.
debug()) <<
"Consensus built ledger we were acquiring";
787 JLOG(
j_.
debug()) <<
"Consensus built new ledger";
797 using namespace std::chrono_literals;
799 auto validationTime = app_.timeKeeper().closeTime();
800 if (validationTime <= lastValidationTime_)
801 validationTime = lastValidationTime_ + 1s;
802 lastValidationTime_ = validationTime;
804 auto v = std::make_shared<STValidation>(
806 validatorKeys_.publicKey,
807 validatorKeys_.secretKey,
808 validatorKeys_.nodeID,
810 v.setFieldH256(sfLedgerHash, ledger.id());
811 v.setFieldH256(sfConsensusHash, txns.id());
813 v.setFieldU32(sfLedgerSequence, ledger.seq());
816 v.setFlag(vfFullValidation);
818 if (ledger.ledger_->rules().enabled(featureHardenedValidations))
823 if (auto const vl = ledgerMaster_.getValidatedLedger())
824 v.setFieldH256(sfValidatedHash, vl->info().hash);
826 v.setFieldU64(sfCookie, valCookie_);
829 if (ledger.ledger_->isVotingLedger())
831 sfServerVersion, BuildInfo::getEncodedVersion());
836 auto const& ft = app_.getFeeTrack();
837 auto const fee = std::max(ft.getLocalFee(), ft.getClusterFee());
838 if (fee > ft.getLoadBase())
839 v.setFieldU32(sfLoadFee, fee);
844 if (ledger.
ledger_->isVotingLedger())
847 feeVote_->doValidation(ledger.ledger_->fees(), v);
852 auto const amendments = app_.getAmendmentTable().doValidation(
853 getEnabledAmendments(*ledger.ledger_));
855 if (!amendments.empty())
857 sfAmendments, STVector256(sfAmendments, amendments));
861 auto const serialized = v->getSerialized();
869 protocol::TMValidation val;
870 val.set_validation(serialized.data(), serialized.size());
871 app_.overlay().broadcast(val);
874 app_.getOPs().pubValidation(v);
880 JLOG(
j_.
info()) <<
"Consensus mode change before=" <<
to_string(before)
888 censorshipDetector_.reset();
916 JLOG(
j_.
error()) <<
"During consensus timerEntry: " << mn.
what();
932 JLOG(
j_.
error()) <<
"During consensus gotTxSet: " << mn.
what();
954 return consensus_.peerProposal(now, newProposal);
964 validating_ = validatorKeys_.publicKey.size() != 0 &&
965 prevLgr.
seq() >= app_.getMaxDisallowedLedger() &&
966 !app_.getOPs().isBlocked();
970 if (validating_ && !app_.config().standalone() && app_.validators().count())
972 auto const when = app_.validators().expires();
974 if (!when || *when < app_.timeKeeper().now())
976 JLOG(
j_.
error()) <<
"Voluntarily bowing out of consensus process "
977 "because of an expired validator list.";
986 JLOG(
j_.
info()) <<
"Entering consensus process, validating, synced="
987 << (synced ?
"yes" :
"no");
992 JLOG(
j_.
info()) <<
"Entering consensus process, watching, synced="
993 << (synced ?
"yes" :
"no");
997 inboundTransactions_.newRound(prevLgr.
seq());
1001 !nowTrusted.
empty())
1002 nUnlVote_.newValidators(prevLgr.
seq() + 1, nowTrusted);
1005 return validating_ && synced;
1011 return ledgerMaster_.haveValidated();
1017 return ledgerMaster_.getValidLedgerIndex();
1023 return app_.validators().getQuorumKeys();
1031 return app_.getValidations().laggards(seq, trustedKeys);
1037 return !validatorKeys_.publicKey.empty();
1043 if (!positions && app_.getOPs().isFull())