fix(consensus): wait for sidecar observation

Require tx-converged peers to advertise sidecar hashes before accepting RNG entropy or export signature success from local quorum alignment.

The RNG reveal fast path now publishes the entropy set and waits for peer observation instead of accepting in the same tick. On timeout, RNG clears the advertised entropy hash and falls back to deterministic zero.

Add unit and CSF regression coverage for asymmetric peer observation.

src/test/consensus/ConsensusExtensions_test.cpp

@@ -113,18 +113,23 @@ struct FakeExtensions
     bool exportSigGateStarted_{false};
     std::chrono::steady_clock::time_point exportSigGateStart_{};
     bool exportSigConvergenceFailed_{false};
+    bool rngOn{false};
     bool localExportSigs{true};
     bool consensusExportTxns{false};
     bool exportOn{true};
+    bool entropyFailed{false};
     std::size_t exportQuorum{4};
     uint256 exportHash{makeHash("local-export-sig-set")};
+    uint256 entropyHash{makeHash("local-entropy-set")};
     std::vector<uint256> fetchedExportSets;
+    std::vector<uint256> fetchedEntropySets;
     int exportBuilds = 0;
+    int entropyBuilds = 0;
 
     bool
     rngEnabled() const
     {
-        return false;
+        return rngOn;
     }
 
     bool
@@ -148,13 +153,13 @@ struct FakeExtensions
     std::size_t
     pendingCommitCount() const
     {
-        return 0;
+        return rngOn ? exportQuorum : 0;
     }
 
     std::size_t
     pendingRevealCount() const
     {
-        return 0;
+        return rngOn ? exportQuorum : 0;
     }
 
     std::size_t
@@ -166,19 +171,19 @@ struct FakeExtensions
     bool
     hasQuorumOfCommits() const
     {
-        return false;
+        return rngOn;
     }
 
     bool
     hasMinimumReveals() const
     {
-        return false;
+        return rngOn;
     }
 
     bool
     hasAnyReveals() const
     {
-        return false;
+        return rngOn;
     }
 
     uint256
@@ -190,7 +195,8 @@ struct FakeExtensions
     uint256
     buildEntropySet(LedgerIndex)
     {
-        return makeHash("entropy-set");
+        ++entropyBuilds;
+        return entropyHash;
     }
 
     uint256
@@ -207,12 +213,15 @@ struct FakeExtensions
     void
     setEntropyFailed()
     {
+        entropyFailed = true;
     }
 
     void
     fetchRngSetIfNeeded(std::optional<uint256> const& hash, SidecarKind kind)
     {
-        if (kind == SidecarKind::exportSig && hash)
+        if (kind == SidecarKind::reveal && hash)
+            fetchedEntropySets.push_back(*hash);
+        else if (kind == SidecarKind::exportSig && hash)
             fetchedExportSets.push_back(*hash);
     }
 
@@ -262,6 +271,7 @@ struct ExportTickHarness
     ConsensusParms parms;
     NetClock::time_point netNow{NetClock::duration{123}};
     std::chrono::steady_clock::time_point start{};
+    std::size_t prevProposers = 4;
     int updates = 0;
     int proposes = 0;
 
@@ -277,6 +287,18 @@ struct ExportTickHarness
             makeNode(id), FakePeerPosition{makeNode(id), peerPosition});
     }
 
+    void
+    addEntropyPeer(
+        std::uint8_t id,
+        std::optional<uint256> entropySetHash,
+        uint256 txSetHash = makeHash("tx-set"))
+    {
+        ExtendedPosition peerPosition{txSetHash};
+        peerPosition.entropySetHash = entropySetHash;
+        peers.emplace(
+            makeNode(id), FakePeerPosition{makeNode(id), peerPosition});
+    }
+
     ExtensionTickResult
     tick(FakeExtensions& ext, std::chrono::milliseconds elapsed = {})
     {
@@ -286,7 +308,7 @@ struct ExportTickHarness
             .nowSteady = start + elapsed,
             .roundTime = elapsed,
             .mode = ConsensusMode::proposing,
-            .prevProposers = 0,
+            .prevProposers = prevProposers,
             .peerPositions = peers,
             .parms = parms,
             .haveCloseTimeConsensus = true,
@@ -418,6 +440,75 @@ class ConsensusExtensions_test : public beast::unit_test::suite
         BEAST_EXPECT(ext.exportSigConvergenceFailed_);
     }
 
+    void
+    testRngEntropyGateRequiresFullObservation()
+    {
+        testcase("RNG entropy gate requires full sidecar observation");
+
+        FakeExtensions ext;
+        ext.rngOn = true;
+        ext.exportOn = false;
+        ext.estState_ = EstablishState::ConvergingReveal;
+
+        ExportTickHarness harness;
+        auto const localHash = ext.entropyHash;
+
+        harness.addEntropyPeer(1, localHash);
+        harness.addEntropyPeer(2, localHash);
+        harness.addEntropyPeer(3, localHash);
+        harness.addEntropyPeer(4, std::nullopt);
+
+        auto result = harness.tick(ext);
+        BEAST_EXPECT(!result.readyForAccept);
+        BEAST_EXPECT(harness.position.entropySetHash == localHash);
+        BEAST_EXPECT(ext.entropySetPublished_);
+
+        // Quorum alignment is not safe if a tx-converged peer has not
+        // advertised any entropySetHash. Otherwise local observation order
+        // can split non-zero entropy from deterministic zero fallback.
+        result = harness.tick(ext, std::chrono::milliseconds{100});
+        BEAST_EXPECT(!result.readyForAccept);
+        BEAST_EXPECT(!ext.entropyFailed);
+        BEAST_EXPECT(harness.position.entropySetHash == localHash);
+
+        result = harness.tick(
+            ext,
+            harness.parms.rngREVEAL_TIMEOUT * 2 + std::chrono::milliseconds{1});
+        BEAST_EXPECT(result.readyForAccept);
+        BEAST_EXPECT(ext.entropyFailed);
+        BEAST_EXPECT(!harness.position.entropySetHash);
+    }
+
+    void
+    testRngFastPathWaitsAfterEntropyPublish()
+    {
+        testcase("RNG fast path waits after entropy publish");
+
+        FakeExtensions ext;
+        ext.rngOn = true;
+        ext.exportOn = false;
+        ext.estState_ = EstablishState::ConvergingCommit;
+
+        ExportTickHarness harness;
+        auto const localHash = ext.entropyHash;
+
+        harness.addEntropyPeer(1, localHash);
+        harness.addEntropyPeer(2, localHash);
+        harness.addEntropyPeer(3, localHash);
+        harness.addEntropyPeer(4, localHash);
+
+        auto result = harness.tick(ext);
+        BEAST_EXPECT(!result.readyForAccept);
+        BEAST_EXPECT(ext.estState_ == EstablishState::ConvergingReveal);
+        BEAST_EXPECT(ext.entropySetPublished_);
+        BEAST_EXPECT(harness.position.entropySetHash == localHash);
+
+        result = harness.tick(ext, std::chrono::milliseconds{100});
+        BEAST_EXPECT(result.readyForAccept);
+        BEAST_EXPECT(!ext.entropyFailed);
+        BEAST_EXPECT(harness.position.entropySetHash == localHash);
+    }
+
     void
     testExportSigGateAllowsAlignedQuorumDespiteMinorityConflict()
     {
@@ -443,6 +534,38 @@ class ConsensusExtensions_test : public beast::unit_test::suite
         BEAST_EXPECT(ext.fetchedExportSets.front() == conflictHash);
     }
 
+    void
+    testExportSigGateRequiresFullObservation()
+    {
+        testcase("Export sig gate requires full sidecar observation");
+
+        FakeExtensions ext;
+        ExportTickHarness harness;
+        auto const localHash = ext.exportHash;
+
+        harness.addPeer(1, localHash);
+        harness.addPeer(2, localHash);
+        harness.addPeer(3, localHash);
+        harness.addPeer(4, std::nullopt);
+
+        auto result = harness.tick(ext);
+        BEAST_EXPECT(!result.readyForAccept);
+        BEAST_EXPECT(harness.position.exportSigSetHash == localHash);
+        BEAST_EXPECT(ext.exportSigGateStarted_);
+
+        // Local quorum alignment is not enough if a tx-converged peer has
+        // not advertised any exportSigSetHash yet.
+        result = harness.tick(ext, std::chrono::milliseconds{100});
+        BEAST_EXPECT(!result.readyForAccept);
+        BEAST_EXPECT(!ext.exportSigConvergenceFailed_);
+
+        result = harness.tick(
+            ext,
+            harness.parms.rngREVEAL_TIMEOUT * 2 + std::chrono::milliseconds{1});
+        BEAST_EXPECT(result.readyForAccept);
+        BEAST_EXPECT(ext.exportSigConvergenceFailed_);
+    }
+
     void
     testExportSigGateFetchesAdvertisedPeerSets()
     {
@@ -591,7 +714,10 @@ public:
     {
         testActiveValidatorViewAppliesNegativeUNL();
         testExportSigGateRequiresQuorumAlignment();
+        testRngEntropyGateRequiresFullObservation();
+        testRngFastPathWaitsAfterEntropyPublish();
         testExportSigGateAllowsAlignedQuorumDespiteMinorityConflict();
+        testExportSigGateRequiresFullObservation();
         testExportSigGateFetchesAdvertisedPeerSets();
         testExportSigGateBoundsCandidateObservationWindow();
         testExportSigGateSkipsWhenExportDisabled();

src/test/consensus/ConsensusRng_test.cpp

@@ -781,6 +781,57 @@ public:
         }
     }
 
+    void
+    testRngFastPathDoesNotOutrunPeerObservation()
+    {
+        using namespace csf;
+        using namespace std::chrono;
+
+        testcase("RNG fast path does not outrun peer observation");
+
+        // Peer 0 can build the full reveal set immediately. The other peers
+        // miss peer 0's proposal-carried reveal, so peer 0 is the one most
+        // likely to hit the reveal fast path early. It must still publish and
+        // wait for sidecar observation instead of closing a non-zero ledger
+        // before the rest of the round can merge or safely zero.
+
+        ConsensusParms const parms{};
+        Sim sim;
+
+        PeerGroup peers = sim.createGroup(5);
+        for (Peer* peer : peers)
+            peer->ce().enableRngConsensus_ = true;
+
+        peers.trustAndConnect(
+            peers, round<milliseconds>(0.2 * parms.ledgerGRANULARITY));
+
+        sim.run(1);
+        BEAST_EXPECT(sim.synchronized(peers));
+
+        for (std::size_t i = 1; i < peers.size(); ++i)
+            peers[i]->ce().dropRevealFrom_.insert(peers[0]->id);
+
+        sim.run(1);
+
+        BEAST_EXPECT(sim.branches(peers) == 1);
+
+        for (Peer const* lhs : peers)
+        {
+            for (Peer const* rhs : peers)
+            {
+                if (lhs->lastClosedLedger.id() != rhs->lastClosedLedger.id())
+                    continue;
+
+                BEAST_EXPECT(
+                    lhs->ce().lastEntropyDigest_ ==
+                    rhs->ce().lastEntropyDigest_);
+                BEAST_EXPECT(
+                    lhs->ce().lastEntropyWasFallback_ ==
+                    rhs->ce().lastEntropyWasFallback_);
+            }
+        }
+    }
+
     void
     testRngNoEntropyWithoutPeerAlignment()
     {
@@ -929,6 +980,7 @@ public:
         RUN(testRngEntropyFallbackOnMajorRevealLoss);
         RUN(testRngSingleByzantineCannotDenyEntropy);
         RUN(testRngEntropyHashConflictWithoutQuorumFallsBackToZero);
+        RUN(testRngFastPathDoesNotOutrunPeerObservation);
         RUN(testRngNoEntropyWithoutPeerAlignment);
         RUN(testRngAlignmentRequiredForNonZeroEntropy);
 

src/xrpld/consensus/ConsensusExtensionsTick.h

@@ -447,8 +447,13 @@ extensionsTick(Ext& ext, Ctx const& ctx)
             if (ext.hasMinimumReveals())
             {
                 publishEntropySet();
+                ext.entropySetPublished_ = true;
+                ext.entropyPublishStart_ = ctx.nowSteady;
                 JLOG(ext.j_.debug())
-                    << "RNG: fast-path published entropySet in same tick";
+                    << "RNG: fast-path published entropySet, waiting for "
+                       "peer observation";
+                logRngDiag("rng-reveal-fast-path-entropy-published-wait");
+                return {};
             }
             else
             {
@@ -544,6 +549,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                         bool conflict = false;
                         std::size_t aligned = 0;
                         std::size_t peersSeen = 0;
+                        std::size_t txConverged = 0;
                     };
 
                     // Phase 2: check peer agreement.  Extension hashes do not
@@ -560,6 +566,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                             auto const& pp = peerPos.proposal().position();
                             if (!(pp == pos))
                                 continue;  // not tx-converged
+                            ++state.txConverged;
                             if (!pp.entropySetHash)
                                 continue;  // peer hasn't published yet
                             ++state.peersSeen;
@@ -583,6 +590,24 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                     auto quorumAligned = [&] {
                         return entropyState.aligned + 1 >= entropyQuorum;
                     };
+                    auto fullObservation = [&] {
+                        // Local quorum alignment is not enough if some
+                        // tx-converged peers have not advertised their
+                        // entropy sidecar hash yet. Otherwise one node can
+                        // accept non-zero from an asymmetric local view while
+                        // the rest of the network times out to zero.
+                        return entropyState.peersSeen ==
+                            entropyState.txConverged;
+                    };
+                    auto clearEntropyHash = [&] {
+                        auto failedPos = ctx.getPosition();
+                        if (!failedPos.entropySetHash)
+                            return;
+                        failedPos.entropySetHash.reset();
+                        ctx.updatePosition(failedPos);
+                        if (ctx.mode == ConsensusMode::proposing)
+                            ctx.propose();
+                    };
 
                     if (entropyState.conflict && !quorumAligned())
                     {
@@ -610,14 +635,17 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                             inspectEntropyPeers(ctx.getPosition(), true);
                     }
 
-                    if (entropyState.conflict && quorumAligned())
+                    if (entropyState.conflict && quorumAligned() &&
+                        fullObservation())
                     {
                         JLOG(ext.j_.debug())
                             << "RNG: entropySetHash conflict ignored after "
                                "quorum alignment"
                             << " alignedParticipants="
                             << (entropyState.aligned + 1)
-                            << " quorum=" << entropyQuorum;
+                            << " quorum=" << entropyQuorum
+                            << " peersSeen=" << entropyState.peersSeen
+                            << " txConverged=" << entropyState.txConverged;
                     }
                     else if (entropyState.conflict)
                     {
@@ -647,6 +675,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
 
                         // Deadline exceeded — fall back to zero.
                         ext.setEntropyFailed();
+                        clearEntropyHash();
                         JLOG(ext.j_.warn())
                             << "RNG: entropySetHash conflict persisted "
                                "past deadline, falling back to zero "
@@ -656,11 +685,14 @@ extensionsTick(Ext& ext, Ctx const& ctx)
 
                     // Positive alignment check: require at least one
                     // tx-converged quorum with a matching entropySetHash
-                    // before accepting non-zero entropy.  Without this,
-                    // a small minority of early publishers could accept
-                    // non-zero entropy before the round has enough shared
-                    // sidecar agreement.
-                    if (!entropyState.conflict && !quorumAligned())
+                    // before accepting non-zero entropy, and require every
+                    // tx-converged peer we are counting to have advertised
+                    // some entropySetHash.  Without the full-observation
+                    // part, asymmetric proposal delivery lets a node accept
+                    // non-zero while peers that are still missing sidecar
+                    // hashes hit the deadline and deterministically zero.
+                    if (!entropyState.conflict &&
+                        (!quorumAligned() || !fullObservation()))
                     {
                         auto const entropyElapsed =
                             ctx.nowSteady - ext.entropyPublishStart_;
@@ -674,18 +706,21 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                                 << " alignedParticipants="
                                 << (entropyState.aligned + 1)
                                 << " quorum=" << entropyQuorum
-                                << " peersSeen=" << entropyState.peersSeen;
+                                << " peersSeen=" << entropyState.peersSeen
+                                << " txConverged=" << entropyState.txConverged;
                             logRngDiag("rng-entropy-hash-quorum-wait");
                             return {};
                         }
                         ext.setEntropyFailed();
+                        clearEntropyHash();
                         JLOG(ext.j_.warn())
                             << "RNG: entropySetHash quorum alignment missing "
                                "within deadline, falling back to zero"
                             << " alignedParticipants="
                             << (entropyState.aligned + 1)
                             << " quorum=" << entropyQuorum
-                            << " peersSeen=" << entropyState.peersSeen;
+                            << " peersSeen=" << entropyState.peersSeen
+                            << " txConverged=" << entropyState.txConverged;
                         logRngDiag("rng-entropy-hash-quorum-timeout");
                     }
 
@@ -695,6 +730,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                         << " alignedParticipants=" << (entropyState.aligned + 1)
                         << " quorum=" << entropyQuorum
                         << " peersSeen=" << entropyState.peersSeen
+                        << " txConverged=" << entropyState.txConverged
                         << " conflict="
                         << (entropyState.conflict ? "yes" : "no");
                 }
@@ -1009,6 +1045,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                     bool conflict = false;
                     std::size_t aligned = 0;
                     std::size_t peersSeen = 0;
+                    std::size_t txConverged = 0;
                 };
 
                 auto inspectExportPeers =
@@ -1023,6 +1060,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                         auto const& pp = peerPos.proposal().position();
                         if (!(pp == pos))
                             continue;  // not tx-converged
+                        ++state.txConverged;
                         if (!pp.exportSigSetHash)
                             continue;  // peer hasn't published yet
                         ++state.peersSeen;
@@ -1046,6 +1084,12 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                 auto quorumAligned = [&] {
                     return exportState.aligned + 1 >= exportQuorum;
                 };
+                auto fullObservation = [&] {
+                    // Export success changes ledger effects too. Require a
+                    // full view of tx-converged peers before treating a local
+                    // quorum as safe enough to succeed in this ledger.
+                    return exportState.peersSeen == exportState.txConverged;
+                };
 
                 if (exportState.conflict && !quorumAligned())
                 {
@@ -1068,15 +1112,20 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                     exportState = inspectExportPeers(ctx.getPosition(), true);
                 }
 
-                if (exportState.conflict && quorumAligned())
+                if (exportState.conflict && quorumAligned() &&
+                    fullObservation())
                 {
                     JLOG(ext.j_.info())
                         << "Export: exportSigSetHash conflict ignored after "
                            "quorum alignment"
                         << " alignedParticipants=" << (exportState.aligned + 1)
-                        << " quorum=" << exportQuorum;
+                        << " quorum=" << exportQuorum
+                        << " peersSeen=" << exportState.peersSeen
+                        << " txConverged=" << exportState.txConverged;
                 }
-                else if (exportState.conflict || !quorumAligned())
+                else if (
+                    exportState.conflict || !quorumAligned() ||
+                    !fullObservation())
                 {
                     auto const elapsed =
                         ctx.nowSteady - ext.exportSigGateStart_;
@@ -1090,6 +1139,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                             << (exportState.aligned + 1)
                             << " quorum=" << exportQuorum
                             << " peersSeen=" << exportState.peersSeen
+                            << " txConverged=" << exportState.txConverged
                             << " conflict="
                             << (exportState.conflict ? "yes" : "no");
                         return {};
@@ -1102,6 +1152,7 @@ extensionsTick(Ext& ext, Ctx const& ctx)
                         << " alignedParticipants=" << (exportState.aligned + 1)
                         << " quorum=" << exportQuorum
                         << " peersSeen=" << exportState.peersSeen
+                        << " txConverged=" << exportState.txConverged
                         << " conflict="
                         << (exportState.conflict ? "yes" : "no");
                 }