From 285fd6df905b84c35adcb63aa5adbf6bc0f40a86 Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Tue, 22 Sep 2026 15:55:42 +0700 Subject: [PATCH] fix(consensus): publish the heartbeat busy flag off the consensus lock The accept job no longer clears extension round state at the end of onPreBuild. The heartbeat reads a published atomic, updated where a contributing input changes, under a mutex that covers those fields. --- .../consensus/ConsensusExtensions_test.cpp | 185 ++++++++++++++++++ .../consensus/SteppingExtensions_test.cpp | 13 ++ src/test/csf/Peer.h | 19 ++ src/test/jtx/SteppingController.h | 13 ++ src/test/jtx/SteppingReplay.h | 14 ++ .../app/consensus/ConsensusExtensions.cpp | 114 ++++++++--- src/xrpld/app/consensus/ConsensusExtensions.h | 53 ++++- src/xrpld/app/consensus/RCLConsensus.cpp | 7 +- src/xrpld/consensus/ConsensusExtensionsTick.h | 8 +- 9 files changed, 386 insertions(+), 40 deletions(-) diff --git a/src/test/consensus/ConsensusExtensions_test.cpp b/src/test/consensus/ConsensusExtensions_test.cpp index e111db29e4..f9d1a41b94 100644 --- a/src/test/consensus/ConsensusExtensions_test.cpp +++ b/src/test/consensus/ConsensusExtensions_test.cpp @@ -459,6 +459,24 @@ struct FakeExtensions return exportOn; } + // The production tick calls this. The fake has no published flag. + void + publishBusy() + { + } + + void + publishEstState(EstablishState state) + { + estState_ = state; + } + + void + publishExportSigGateStarted() + { + exportSigGateStarted_ = true; + } + bool exportFinalizationViewAnchored() const { @@ -3182,6 +3200,10 @@ class ConsensusExtensions_test : public beast::unit_test::suite aligning.setMode(ConsensusMode::observing); aligning.startExportShareService(); BEAST_EXPECT(aligning.onExportShare(share, {}).isAccepted()); + BEAST_EXPECT( + aligning.busyPublished_.load(std::memory_order_relaxed) == + aligning.computeBusy()); + BEAST_EXPECT(aligning.extensionsBusy()); BEAST_EXPECT(!aligning.localIsActiveValidator()); ExtensionTickHarness observation; observation.mode = ConsensusMode::observing; @@ -3242,6 +3264,10 @@ class ConsensusExtensions_test : public beast::unit_test::suite BEAST_EXPECT(!aligning.acceptedExportSigSetHash_); BEAST_EXPECT(!aligning.hasEligiblePendingExports()); BEAST_EXPECT(!aligning.hasPendingExportSigs()); + BEAST_EXPECT( + aligning.busyPublished_.load(std::memory_order_relaxed) == + aligning.computeBusy()); + BEAST_EXPECT(!aligning.extensionsBusy()); // A local root is still not permission without peer alignment. The // old round's already-admitted share survives, but cannot self-count. @@ -5541,10 +5567,169 @@ class ConsensusExtensions_test : public beast::unit_test::suite env.app().getInboundTransactions().getSet(entropyHash, false)); } + void + testBusyFlagTransitions() + { + testcase("busy flag follows every contributing input"); + using namespace jtx; + Env env{ + *this, + envconfig(validator, ""), + supported_amendments() | featureExport, + nullptr}; + Account const alice{"alice"}; + env.fund(XRP(1000), alice); + env.close(); + env.app().getJobQueue().rendezvous(); + + ConsensusExtensions ce{env.app(), activeNoopJournal()}; + auto const check = [&](char const* step, bool busy) { + BEAST_EXPECT( + ce.busyPublished_.load(std::memory_order_relaxed) == + ce.computeBusy()); + BEAST_EXPECT(ce.extensionsBusy() == busy); + if (ce.extensionsBusy() != busy) + log << " busy step " << step + << " published=" << ce.extensionsBusy() << std::endl; + }; + check("initial", false); + ce.setExportEnabledThisRound(false); + check("export disabled", false); + ce.setExportEnabledThisRound(true); + check("export enabled idle", false); + ce.estState_ = EstablishState::ConvergingCommit; + ce.publishBusy(); + check("commit phase", true); + ce.estState_ = EstablishState::ConvergingReveal; + ce.publishBusy(); + check("reveal phase", true); + ce.estState_ = EstablishState::ConvergingTx; + ce.publishBusy(); + check("tx phase", false); + ce.resetSubState(); + check("reset substate", false); + + ce.setExportEnabledThisRound(true); + ce.exportSigGateStarted_ = true; + ce.publishBusy(); + check("gate started", true); + ce.setExportEnabledThisRound(false); + check("export disabled suppresses gate", false); + ce.setExportEnabledThisRound(true); + check("export enabled restores gate", true); + // Replay while Export is enabled keeps the gate. The disabled + // doAccept path is the one that clears it. + ce.onReplayBuild(); + check("replay keeps gate while export enabled", true); + ce.clearRngState(); + check("clear while export enabled", false); + ce.setExportEnabledThisRound(false); + ce.onReplayBuild(); + check("replay while export disabled", false); + + auto const parent = env.app().getLedgerMaster().getValidatedLedger(); + if (!BEAST_EXPECT(parent != nullptr)) + return; + auto validated = std::make_shared( + *parent, env.app().timeKeeper().closeTime()); + auto const deadline = validated->info().seq + 1; + auto const origin = makeHash("busy-flag-origin"); + auto latch = + std::make_shared(keylet::exportLatch(alice.id(), origin)); + latch->setAccountID(sfAccount, alice.id()); + latch->setFieldU32(sfTicketSequence, 1); + latch->setFieldH256(sfTransactionHash, origin); + latch->setFieldH256(sfDigest, makeHash("busy-flag-intent")); + latch->setFieldU32(sfLedgerSequence, validated->info().seq); + latch->setFieldH256( + sfExportCommitteeHash, validated->info().parentHash); + latch->setFieldU32(sfLastLedgerSequence, deadline); + Sandbox sandbox{validated.get(), tapNONE}; + BEAST_EXPECT(isTesSuccess(ExportLedgerOps::insertPendingExportLatch( + sandbox, sandbox, latch, env.journal))); + sandbox.apply(*validated); + validated->updateSkipList(); + validated->setAccepted( + validated->info().closeTime, + validated->info().closeTimeResolution, + true); + auto nextParent = std::make_shared( + *validated, env.app().timeKeeper().closeTime()); + env.app().getLedgerMaster().setFullLedger(validated, false, false); + + ce.onRoundStart(RCLCxLedger{validated}, {}); + BEAST_EXPECT(ce.exportEnabled()); + check("round start before admission", false); + + auto& collector = ce.postValidationExportSigCollector(); + auto const signer = randomKeyPair(KeyType::secp256k1).first; + std::uint8_t const signatureBytes[] = {1, 2, 3}; + Buffer const signature{signatureBytes, sizeof(signatureBytes)}; + BEAST_EXPECT(collector.registerOrigin(origin, deadline)); + auto admission = collector.beginAttributedAdmission( + origin, + ExportSigCollector::Contribution{0, signer, signature}, + deadline); + BEAST_EXPECT(admission.ticket); + if (!admission.ticket) + return; + BEAST_EXPECT( + collector + .admitContribution(std::move(*admission.ticket), true, deadline) + .result == ExportSigCollector::AdmitResult::accepted); + BEAST_EXPECT(ce.hasPendingExportSigs()); + // The collector write above is the same mutation admitExportShare + // publishes after. Publish here so this case checks the predicate. + // The witness-rebuild case drives admitExportShare itself. + ce.publishBusy(); + check("pending signature admitted", true); + ce.setExportEnabledThisRound(false); + check("export disabled suppresses pending", false); + ce.setExportEnabledThisRound(true); + check("export enabled restores pending", true); + ce.setExportEnabledThisRound(false); + ce.clearRngState(); + ce.setExportEnabledThisRound(true); + BEAST_EXPECT(!ce.hasPendingExportSigs()); + check("clear while export disabled drops pending", false); + + // clearRngState drops the round parent as well as the collector. + // Restore the same parent before checking admission again. + ce.onRoundStart(RCLCxLedger{validated}, {}); + BEAST_EXPECT(ce.exportEnabled()); + check("round restored before readmit", false); + + BEAST_EXPECT(collector.registerOrigin(origin, deadline)); + auto again = collector.beginAttributedAdmission( + origin, + ExportSigCollector::Contribution{0, signer, signature}, + deadline); + BEAST_EXPECT(again.ticket); + if (!again.ticket) + return; + BEAST_EXPECT( + collector + .admitContribution(std::move(*again.ticket), true, deadline) + .result == ExportSigCollector::AdmitResult::accepted); + ce.publishBusy(); + check("pending signature readmitted", true); + + nextParent->updateSkipList(); + nextParent->setAccepted( + nextParent->info().closeTime, + nextParent->info().closeTimeResolution, + true); + ce.onRoundStart(RCLCxLedger{nextParent}, {}); + BEAST_EXPECT(ce.exportEnabled()); + BEAST_EXPECT(!ce.hasPendingExportSigs()); + check("round parent moved past candidate", false); + } + public: void run() override { + testBusyFlagTransitions(); testSidecarPeerAlignmentHelper(); testHarnessEntropyRequiresStepping(); testSidecarSplitBrainEquivocationThreshold(); diff --git a/src/test/consensus/SteppingExtensions_test.cpp b/src/test/consensus/SteppingExtensions_test.cpp index 20852efc8c..db89fb8430 100644 --- a/src/test/consensus/SteppingExtensions_test.cpp +++ b/src/test/consensus/SteppingExtensions_test.cpp @@ -46,6 +46,7 @@ class SteppingExtensions_test : public beast::unit_test::suite // The first flag vote has fewer than 256 ancestors and only establishes // validation retention. The second can score a complete history window. static constexpr std::uint32_t warmLedger = 2 * FLAG_LEDGER_INTERVAL + 1; + std::uint64_t busyInvariantChecks_{0}; enum class Fault { none, @@ -5306,6 +5307,16 @@ public: void run() override { + busyInvariantChecks_ = 0; + steppingBusyProbe() = [this](SteppingNetwork& net, std::uint32_t id) { + if (!net.isLive(id)) + return; + auto& ce = net.node(id).app().getConsensusExtensions(); + ++busyInvariantChecks_; + BEAST_EXPECT( + ce.busyPublished_.load(std::memory_order_relaxed) == + ce.computeBusy()); + }; // Optional focused iteration, e.g. --unittest-arg=case=validator. // Keep replays=N available to the existing replay combinator. std::string filter; @@ -5578,6 +5589,8 @@ public: }); } BEAST_EXPECT(selected != 0); + log << " busy invariant checks=" << busyInvariantChecks_ << std::endl; + steppingBusyProbe() = {}; } }; diff --git a/src/test/csf/Peer.h b/src/test/csf/Peer.h index 460d612c28..9dba7be8c5 100644 --- a/src/test/csf/Peer.h +++ b/src/test/csf/Peer.h @@ -1155,6 +1155,25 @@ struct Peer } // --- Sub-state accessors --- + // The production tick calls this after a phase change. This sim still + // computes the predicate directly, so the call does not store a flag. + void + publishBusy() + { + } + + void + publishEstState(EstablishState state) + { + estState_ = state; + } + + void + publishExportSigGateStarted() + { + exportSigGateStarted_ = true; + } + bool extensionsBusy() const { diff --git a/src/test/jtx/SteppingController.h b/src/test/jtx/SteppingController.h index 7709e0c871..8a3b8fffa4 100644 --- a/src/test/jtx/SteppingController.h +++ b/src/test/jtx/SteppingController.h @@ -157,6 +157,8 @@ private: std::map, duration> jobLags_; std::map, duration> namedJobLags_; std::function beforeJob_; + // Survives observeJobs replacement. Scenario-wide invariants use this. + std::function alwaysBeforeJob_; [[nodiscard]] static char const* jobTypeName(JobType t) @@ -534,6 +536,15 @@ public: beforeJob_ = std::move(observer); } + // Survives observeJobs replacement. Scenario-wide invariants use this. + void + setAlwaysBeforeJob( + std::function observer) + { + requireSteppingThread("setAlwaysBeforeJob"); + alwaysBeforeJob_ = std::move(observer); + } + void clearJobLag(std::uint32_t nodeId) { @@ -924,6 +935,8 @@ public: if (lag != duration::zero()) clearLaggedPending(nodeId, tier, t, name, lag); recordJob(nodeId, t, name, "run"); + if (alwaysBeforeJob_) + alwaysBeforeJob_(nodeId, t, name); if (beforeJob_) beforeJob_(nodeId, t, name); f(); diff --git a/src/test/jtx/SteppingReplay.h b/src/test/jtx/SteppingReplay.h index ccf6eeeb0f..500d67e622 100644 --- a/src/test/jtx/SteppingReplay.h +++ b/src/test/jtx/SteppingReplay.h @@ -169,6 +169,13 @@ logFirstValDivergence( } // namespace detail +inline std::function& +steppingBusyProbe() +{ + static std::function probe; + return probe; +} + // Run `scenario` K times (K = max(minRuns, --unittest-arg replays=N)) and // assert every run reproduces run 1 exactly — payload and executed-order // fingerprint. The scenario receives a fresh SteppingNetwork with forensics @@ -190,6 +197,13 @@ expectReplays( { SteppingNetwork net(s); net.recordForensics(); + if (steppingBusyProbe()) + { + net.controller().setAlwaysBeforeJob( + [&net](std::uint32_t id, JobType, std::string const&) { + steppingBusyProbe()(net, id); + }); + } auto const payload = scenario(net); if (!s.expect( payload.has_value(), diff --git a/src/xrpld/app/consensus/ConsensusExtensions.cpp b/src/xrpld/app/consensus/ConsensusExtensions.cpp index ce942986b4..a668dbc3a9 100644 --- a/src/xrpld/app/consensus/ConsensusExtensions.cpp +++ b/src/xrpld/app/consensus/ConsensusExtensions.cpp @@ -660,6 +660,7 @@ ConsensusExtensions::admitExportShare( Slice{share.signature.data(), share.signature.size()}); auto outcome = postValidationExportSigCollector_.admitContribution( std::move(*admission.ticket), signatureVerified, validated->info().seq); + publishBusy(); JLOG(j_.trace()) << "ExportShare: collector commit" << " origin=" << share.originTxn << " position=" << unsigned(share.committeePosition) @@ -759,6 +760,7 @@ ConsensusExtensions::onValidatedLedger( } postValidationExportSigCollector_.cleanupStale(validated->info().seq); + publishBusy(); // Replay the retained share view once at each validated cursor. Shares // admitted concurrently are serialized by exportStreamMutex_: they @@ -1954,7 +1956,10 @@ ConsensusExtensions::buildCommitSet(LedgerIndex seq) // Track the active RNG round explicitly. Nodes in observing/switching // mode can have a closed ledger index behind the consensus round while // still building that round's local RNG snapshots. - buildingLedgerSeq_ = seq; + { + std::lock_guard lock(busyMu_); + buildingLedgerSeq_ = seq; + } auto map = std::make_shared(SHAMapType::SIDECAR, app_.getNodeFamily()); @@ -2012,13 +2017,17 @@ ConsensusExtensions::buildCommitSet(LedgerIndex seq) << " entries=" << entryCount << " pendingCommits=" << pendingCommits_.size() << " activeValidators=" << validatorView->size(); + publishBusy(); return hash; } uint256 ConsensusExtensions::buildEntropySet(LedgerIndex seq) { - buildingLedgerSeq_ = seq; + { + std::lock_guard lock(busyMu_); + buildingLedgerSeq_ = seq; + } auto map = std::make_shared(SHAMapType::SIDECAR, app_.getNodeFamily()); @@ -2078,6 +2087,7 @@ ConsensusExtensions::buildEntropySet(LedgerIndex seq) << " entries=" << entryCount << " pendingReveals=" << pendingReveals_.size() << " activeValidators=" << validatorView->size(); + publishBusy(); return hash; } @@ -2087,10 +2097,15 @@ ConsensusExtensions::pendingRoundExports(LedgerIndex candidateSeq) const // This is reconstruction eligibility, not permission to admit or release // a share. A newer validated ledger may already have witnessed an origin // that is still pending in the parent of our in-flight round. - if (!roundParentLedger_ || candidateSeq == 0 || - roundParentLedger_->info().seq != candidateSeq - 1) - return {}; - return pendingExportLatches(*roundParentLedger_, candidateSeq); + std::shared_ptr parent; + { + std::lock_guard lock(busyMu_); + if (!roundParentLedger_ || candidateSeq == 0 || + roundParentLedger_->info().seq != candidateSeq - 1) + return {}; + parent = roundParentLedger_; + } + return pendingExportLatches(*parent, candidateSeq); } uint256 @@ -2161,7 +2176,12 @@ ConsensusExtensions::buildExportSigSet(LedgerIndex seq) bool ConsensusExtensions::hasPendingExportSigs() const { - auto const live = pendingRoundExports(buildingLedgerSeq_.value_or(0)); + LedgerIndex seq; + { + std::lock_guard lock(busyMu_); + seq = buildingLedgerSeq_.value_or(0); + } + auto const live = pendingRoundExports(seq); auto const allSigs = postValidationExportSigCollector_.fullUnionSnapshot(); return std::any_of(allSigs.begin(), allSigs.end(), [&](auto const& entry) { return live.find(entry.first) != live.end(); @@ -2171,7 +2191,12 @@ ConsensusExtensions::hasPendingExportSigs() const bool ConsensusExtensions::hasEligiblePendingExports() const { - return !pendingRoundExports(buildingLedgerSeq_.value_or(0)).empty(); + LedgerIndex seq; + { + std::lock_guard lock(busyMu_); + seq = buildingLedgerSeq_.value_or(0); + } + return !pendingRoundExports(seq).empty(); } void @@ -2353,8 +2378,14 @@ ConsensusExtensions::generateEntropySecret() if (!app_.config().steppingMode) Throw( "Harness entropy requires deterministic stepping mode"); - myEntropySecret_ = - generate(roundPrevLedgerHash_, buildingLedgerSeq_.value_or(0)); + uint256 parentHash; + LedgerIndex seq; + { + std::lock_guard lock(busyMu_); + parentHash = roundPrevLedgerHash_; + seq = buildingLedgerSeq_.value_or(0); + } + myEntropySecret_ = generate(parentHash, seq); } else { @@ -2410,14 +2441,20 @@ ConsensusExtensions::clearRngStatePreservingExport() commitSetMap_.reset(); entropySetMap_.reset(); acceptedEntropySetHash_.reset(); - buildingLedgerSeq_.reset(); - roundPrevLedgerHash_ = uint256{}; - roundParentLedger_.reset(); observedParticipantsHash_.reset(); observedParticipantsCount_ = 0; observedParticipantsBitmapBin_.clear(); likelyParticipants_.clear(); commitProofs_.clear(); + { + // These three are what the published predicate reads. Keep the + // critical section to the stores so accept does not stall the tick. + std::lock_guard lock(busyMu_); + buildingLedgerSeq_.reset(); + roundPrevLedgerHash_ = uint256{}; + roundParentLedger_.reset(); + busyPublished_.store(computeBusyUnlocked(), std::memory_order_relaxed); + } //@@end round-stop-rng-reset // Keep the round-level enable latches intact here. onRoundStart() refreshes // them from the consensus parent before clearing so boundary cleanup, such @@ -2437,9 +2474,12 @@ ConsensusExtensions::clearRngState() exportSigSetMap_.reset(); acceptedExportSigSetHash_.reset(); proposalPublishedExportShares_.clear(); - exportSigGateStarted_ = false; - exportSigGateStart_ = {}; - exportSigConvergenceFailed_ = false; + { + std::lock_guard lock(busyMu_); + exportSigGateStarted_ = false; + exportSigGateStart_ = {}; + exportSigConvergenceFailed_ = false; + } //@@end round-stop-export-reset clearRngStatePreservingExport(); @@ -2781,7 +2821,11 @@ ConsensusExtensions::onPreBuild( if (app_.config().standalone() && hasPendingExportSigs()) buildExportSigSet(seq); - auto const parent = roundParentLedger_; + std::shared_ptr parent; + { + std::lock_guard lock(busyMu_); + parent = roundParentLedger_; + } auto const validated = app_.getLedgerMaster().getValidatedLedger(); // Rebuild only from this round's exact parent and accepted evidence. // Local validation can lag that parent or already have passed it. @@ -2992,11 +3036,8 @@ ConsensusExtensions::onPreBuild( << " buildSeq=" << seq; } - //@@start accept-time-cleanup-success - // Export's ledger-defining signature witnesses are now self-contained - // transactions in the stream; no later apply step reads sidecar memory. - clearRngStatePreservingExport(); - //@@end accept-time-cleanup-success + // onRoundStart already cleared this round's extension state and then + // reassigned the round fields. A second clear here races the heartbeat. } void @@ -3248,20 +3289,27 @@ ConsensusExtensions::onRoundStart( //@@end round-extension-feature-latches clearRngState(); - roundPrevLedgerHash_ = prevLedger.ledger_->info().hash; - roundParentLedger_ = prevLedger.ledger_; - buildingLedgerSeq_ = prevLedger.ledger_->info().seq + 1; + uint256 roundHash; + { + std::lock_guard lock(busyMu_); + roundPrevLedgerHash_ = prevLedger.ledger_->info().hash; + roundParentLedger_ = prevLedger.ledger_; + buildingLedgerSeq_ = prevLedger.ledger_->info().seq + 1; + roundHash = roundPrevLedgerHash_; + busyPublished_.store(computeBusyUnlocked(), std::memory_order_relaxed); + } cacheUNLReport(prevLedger.ledger_); auto const validatorView = activeValidatorView(); if (validatorView->sourceLedgerHash) { XRPL_ASSERT( - *validatorView->sourceLedgerHash == roundPrevLedgerHash_, + *validatorView->sourceLedgerHash == roundHash, "ripple::ConsensusExtensions::onRoundStart : " "active view source matches round parent"); } setExpectedProposers(std::move(lastProposers)); resetSubState(); + publishBusy(); } void @@ -3542,12 +3590,20 @@ ConsensusExtensions::attachExportSignatures( auto const& keys = app_.getValidatorKeys(); auto const validated = app_.getLedgerMaster().getValidatedLedger(); + std::optional buildSeq; + uint256 roundParentHash; + { + std::lock_guard lock(busyMu_); + if (buildingLedgerSeq_) + buildSeq = *buildingLedgerSeq_; + roundParentHash = roundPrevLedgerHash_; + } if (!keys.keys || keys.nodeID == beast::zero || !validated || - !validated->rules().enabled(featureExport) || !buildingLedgerSeq_ || - proposal.prevLedger() != roundPrevLedgerHash_) + !validated->rules().enabled(featureExport) || !buildSeq || + proposal.prevLedger() != roundParentHash) return; - auto const live = pendingExportLatches(*validated, *buildingLedgerSeq_); + auto const live = pendingExportLatches(*validated, *buildSeq); auto const snapshot = postValidationExportSigCollector_.fullUnionSnapshot(); std::size_t attached = 0; for (auto const& [origin, contributions] : snapshot) diff --git a/src/xrpld/app/consensus/ConsensusExtensions.h b/src/xrpld/app/consensus/ConsensusExtensions.h index b884812b0a..8abb7df819 100644 --- a/src/xrpld/app/consensus/ConsensusExtensions.h +++ b/src/xrpld/app/consensus/ConsensusExtensions.h @@ -643,19 +643,46 @@ public: setExportEnabledThisRound(bool v) { exportEnabledThisRound_.store(v, std::memory_order_relaxed); + publishBusy(); } + // Heartbeat reads only this atomic. The predicate lives in computeBusy(); + // every writer of a contributing input calls publishBusy(). bool extensionsBusy() const { - return estState_ != EstablishState::ConvergingTx || - (exportEnabled() && - (exportSigGateStarted_ || hasPendingExportSigs())); + return busyPublished_.load(std::memory_order_relaxed); + } + + void + publishBusy() + { + std::lock_guard lock(busyMu_); + busyPublished_.store(computeBusyUnlocked(), std::memory_order_relaxed); + } + + // Phase changes happen in the tick, which is not a member. The store and + // the publish share busyMu_ so a job-thread publish cannot race them. + void + publishEstState(EstablishState state) + { + std::lock_guard lock(busyMu_); + estState_ = state; + busyPublished_.store(computeBusyUnlocked(), std::memory_order_relaxed); + } + + void + publishExportSigGateStarted() + { + std::lock_guard lock(busyMu_); + exportSigGateStarted_ = true; + busyPublished_.store(computeBusyUnlocked(), std::memory_order_relaxed); } void resetSubState() { + std::lock_guard lock(busyMu_); estState_ = EstablishState::ConvergingTx; revealPhaseStart_ = {}; commitHashConflictStart_ = {}; @@ -664,7 +691,27 @@ public: exportSigGateStarted_ = false; exportSigGateStart_ = {}; exportSigConvergenceFailed_ = false; + busyPublished_.store(computeBusyUnlocked(), std::memory_order_relaxed); } + +private: + bool + computeBusyUnlocked() const + { + return estState_ != EstablishState::ConvergingTx || + (exportEnabled() && + (exportSigGateStarted_ || hasPendingExportSigs())); + } + + bool + computeBusy() const + { + std::lock_guard lock(busyMu_); + return computeBusyUnlocked(); + } + + mutable std::recursive_mutex busyMu_; + std::atomic busyPublished_{false}; }; } // namespace ripple diff --git a/src/xrpld/app/consensus/RCLConsensus.cpp b/src/xrpld/app/consensus/RCLConsensus.cpp index ec2ca70817..92377b6292 100644 --- a/src/xrpld/app/consensus/RCLConsensus.cpp +++ b/src/xrpld/app/consensus/RCLConsensus.cpp @@ -1086,10 +1086,9 @@ RCLConsensus::phase() const bool RCLConsensus::extensionsBusy() const { - // ConsensusExtensions state is mutated by timer, peer-proposal and - // local sidecar snapshot paths under this mutex. Busy polling observes - // the same state, so it must share the same synchronization boundary. - std::lock_guard _{mutex_}; + // The heartbeat reads only the published atomic. It does not take + // mutex_: the accept job and the share jobs do not hold that lock, + // and the atomic is the synchronization boundary for this poll. return consensus_->extensionsBusy(); } diff --git a/src/xrpld/consensus/ConsensusExtensionsTick.h b/src/xrpld/consensus/ConsensusExtensionsTick.h index 6c467d0399..31e21bd1cd 100644 --- a/src/xrpld/consensus/ConsensusExtensionsTick.h +++ b/src/xrpld/consensus/ConsensusExtensionsTick.h @@ -332,7 +332,7 @@ rngTick(Ext& ext, Ctx const& ctx, Propose const& requestProposal) if (ctx.mode == ConsensusMode::proposing) requestProposal(); - ext.estState_ = EstablishState::ConvergingCommit; + ext.publishEstState(EstablishState::ConvergingCommit); ext.commitHashConflictStart_ = {}; JLOG(ext.j_.debug()) << "RNG: transitioned to ConvergingCommit" << " buildSeq=" << buildSeq @@ -382,7 +382,7 @@ rngTick(Ext& ext, Ctx const& ctx, Propose const& requestProposal) ctx.updatePosition(newPos); if (ctx.mode == ConsensusMode::proposing) requestProposal(); - ext.estState_ = EstablishState::ConvergingCommit; + ext.publishEstState(EstablishState::ConvergingCommit); ext.commitHashConflictStart_ = {}; JLOG(ext.j_.debug()) << "RNG: transitioned to ConvergingCommit" @@ -534,7 +534,7 @@ rngTick(Ext& ext, Ctx const& ctx, Propose const& requestProposal) if (ctx.mode == ConsensusMode::proposing) requestProposal(); - ext.estState_ = EstablishState::ConvergingReveal; + ext.publishEstState(EstablishState::ConvergingReveal); //@@end rng-reveal-transition ext.revealPhaseStart_ = ctx.nowSteady; JLOG(ext.j_.debug()) << "RNG: transitioned to ConvergingReveal" @@ -921,7 +921,7 @@ exportTick(Ext& ext, Ctx const& ctx, Propose const& requestProposal) auto startExportSigGate = [&]() -> bool { if (ext.exportSigGateStarted_) return false; - ext.exportSigGateStarted_ = true; + ext.publishExportSigGateStarted(); ext.exportSigGateStart_ = ctx.nowSteady; return true; };