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.

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<Ledger>(
+            *parent, env.app().timeKeeper().closeTime());
+        auto const deadline = validated->info().seq + 1;
+        auto const origin = makeHash("busy-flag-origin");
+        auto latch =
+            std::make_shared<SLE>(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<Ledger>(
+            *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();

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() = {};
     }
 };
 

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
         {

src/test/jtx/SteppingController.h

@@ -157,6 +157,8 @@ private:
     std::map<std::pair<std::uint32_t, int>, duration> jobLags_;
     std::map<std::tuple<std::uint32_t, JobType, std::string>, duration> namedJobLags_;
     std::function<void(std::uint32_t, JobType, std::string const&)> beforeJob_;
+    // Survives observeJobs replacement. Scenario-wide invariants use this.
+    std::function<void(std::uint32_t, JobType, std::string const&)> 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<void(std::uint32_t, JobType, std::string const&)> 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();

src/test/jtx/SteppingReplay.h

@@ -169,6 +169,13 @@ logFirstValDivergence(
 
 }  // namespace detail
 
+inline std::function<void(SteppingNetwork&, std::uint32_t)>&
+steppingBusyProbe()
+{
+    static std::function<void(SteppingNetwork&, std::uint32_t)> 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(),

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<SHAMap>(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<SHAMap>(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<Ledger const> 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<std::logic_error>(
                 "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<Ledger const> 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<LedgerIndex> 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)

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<bool> busyPublished_{false};
 };
 
 }  // namespace ripple

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();
 }
 

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;
         };