From 5d85d2df4bf436e5606cdc80d82a994111a6dbda Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Sun, 30 Nov 2025 11:27:57 +0700 Subject: [PATCH] feat: add priority node fetching and network-observed ledger tracking Partial sync mode improvements for faster RPC queries during sync: - Track network-observed ledger from any validation (not just trusted) to allow queries before trusted validators are configured - Add priority node fetching: queries can request specific nodes be fetched immediately via addPriorityNode/addPriorityHash - Store state/transaction nodes directly to node store (not fetch pack) so partial sync queries find them immediately - Add poll-wait loops in RPCHelpers for ledger header acquisition - Replace postAndYield with sleep_for in SHAMap finishFetch - Implement linear backoff for re-requests (50ms increments, max 2s) --- src/ripple/app/consensus/RCLValidations.cpp | 18 ++++++ src/ripple/app/ledger/InboundLedger.h | 9 +++ src/ripple/app/ledger/InboundLedgers.h | 9 +++ src/ripple/app/ledger/LedgerMaster.h | 28 +++++++++ src/ripple/app/ledger/impl/InboundLedger.cpp | 46 +++++++++++++++ src/ripple/app/ledger/impl/InboundLedgers.cpp | 31 ++++++++++ src/ripple/app/ledger/impl/LedgerMaster.cpp | 14 ++++- src/ripple/overlay/impl/PeerImp.cpp | 37 ++++++++++-- src/ripple/rpc/impl/RPCHelpers.cpp | 59 ++++++++++++++++++- src/ripple/shamap/Family.h | 7 ++- src/ripple/shamap/NodeFamily.h | 5 +- src/ripple/shamap/ShardFamily.h | 6 +- src/ripple/shamap/impl/NodeFamily.cpp | 12 +++- src/ripple/shamap/impl/SHAMap.cpp | 26 ++++++-- src/ripple/shamap/impl/ShardFamily.cpp | 10 +++- src/test/app/LedgerReplay_test.cpp | 5 ++ src/test/shamap/common.h | 6 +- 17 files changed, 305 insertions(+), 23 deletions(-) diff --git a/src/ripple/app/consensus/RCLValidations.cpp b/src/ripple/app/consensus/RCLValidations.cpp index d5512bae1..5096621f7 100644 --- a/src/ripple/app/consensus/RCLValidations.cpp +++ b/src/ripple/app/consensus/RCLValidations.cpp @@ -178,8 +178,26 @@ handleNewValidation( auto const outcome = validations.add(calcNodeID(masterKey.value_or(signingKey)), val); + if (j.has_value()) + { + JLOG(j->warn()) << "handleNewValidation: seq=" << seq + << " hash=" << hash << " trusted=" << val->isTrusted() + << " outcome=" + << (outcome == ValStatus::current + ? "current" + : outcome == ValStatus::stale + ? "stale" + : outcome == ValStatus::badSeq ? "badSeq" + : "other"); + } + if (outcome == ValStatus::current) { + // For partial sync: track the network-observed ledger from ANY + // validation (not just trusted). This allows queries before + // trusted validators are fully configured. + app.getLedgerMaster().setNetworkObservedLedger(hash, seq); + if (val->isTrusted()) { // Was: app.getLedgerMaster().checkAccept(hash, seq); diff --git a/src/ripple/app/ledger/InboundLedger.h b/src/ripple/app/ledger/InboundLedger.h index fc5ee8f69..0364788d5 100644 --- a/src/ripple/app/ledger/InboundLedger.h +++ b/src/ripple/app/ledger/InboundLedger.h @@ -114,6 +114,12 @@ public: void runData(); + /** Add a node hash to the priority queue for immediate fetching. + Used by partial sync mode to prioritize nodes needed by queries. + */ + void + addPriorityHash(uint256 const& hash); + void touch() { @@ -195,6 +201,9 @@ private: std::set mRecentNodes; + // Priority nodes to fetch immediately (for partial sync queries) + std::set priorityHashes_; + SHAMapAddNode mStats; // Data we have received from peers diff --git a/src/ripple/app/ledger/InboundLedgers.h b/src/ripple/app/ledger/InboundLedgers.h index a4b7fc290..bb8b9c38f 100644 --- a/src/ripple/app/ledger/InboundLedgers.h +++ b/src/ripple/app/ledger/InboundLedgers.h @@ -64,6 +64,15 @@ public: virtual std::shared_ptr getPartialLedger(uint256 const& hash) = 0; + /** Add a priority node hash for immediate fetching. + Used by partial sync mode to prioritize specific nodes + needed by queries. + @param ledgerSeq The ledger sequence being acquired + @param nodeHash The specific node hash to prioritize + */ + virtual void + addPriorityNode(std::uint32_t ledgerSeq, uint256 const& nodeHash) = 0; + // VFALCO TODO Remove the dependency on the Peer object. // virtual bool diff --git a/src/ripple/app/ledger/LedgerMaster.h b/src/ripple/app/ledger/LedgerMaster.h index 8f3f057dd..2c7e6d40b 100644 --- a/src/ripple/app/ledger/LedgerMaster.h +++ b/src/ripple/app/ledger/LedgerMaster.h @@ -308,6 +308,30 @@ public: return mLastValidLedger; } + //! For partial sync: set the network-observed ledger from any validation. + //! This allows queries before trusted validators are fully configured. + void + setNetworkObservedLedger(uint256 const& hash, LedgerIndex seq) + { + std::lock_guard lock(m_mutex); + if (seq > mNetworkObservedLedger.second) + { + JLOG(jPartialSync_.warn()) + << "network-observed ledger updated to seq=" << seq + << " hash=" << hash; + mNetworkObservedLedger = std::make_pair(hash, seq); + } + } + + //! Get the network-observed ledger (from any validations, not just + //! trusted). + std::pair + getNetworkObservedLedger() + { + std::lock_guard lock(m_mutex); + return mNetworkObservedLedger; + } + // Returns the minimum ledger sequence in SQL database, if any. std::optional minSqlSeq(); @@ -357,6 +381,7 @@ private: Application& app_; beast::Journal m_journal; + beast::Journal jPartialSync_; std::recursive_mutex mutable m_mutex; @@ -381,6 +406,9 @@ private: // Fully validated ledger, whether or not we have the ledger resident. std::pair mLastValidLedger{uint256(), 0}; + // Network-observed ledger from any validations (for partial sync). + std::pair mNetworkObservedLedger{uint256(), 0}; + LedgerHistory mLedgerHistory; CanonicalTXSet mHeldTransactions{uint256()}; diff --git a/src/ripple/app/ledger/impl/InboundLedger.cpp b/src/ripple/app/ledger/impl/InboundLedger.cpp index c08331efe..5d2c367ee 100644 --- a/src/ripple/app/ledger/impl/InboundLedger.cpp +++ b/src/ripple/app/ledger/impl/InboundLedger.cpp @@ -190,6 +190,15 @@ InboundLedger::update(std::uint32_t seq) touch(); } +void +InboundLedger::addPriorityHash(uint256 const& hash) +{ + ScopedLockType sl(mtx_); + priorityHashes_.insert(hash); + JLOG(journal_.debug()) << "Added priority hash " << hash << " for ledger " + << hash_; +} + bool InboundLedger::checkLocal() { @@ -588,6 +597,43 @@ InboundLedger::trigger(std::shared_ptr const& peer, TriggerReason reason) } } + // Handle priority hashes immediately (for partial sync queries) + if (mHaveHeader && !priorityHashes_.empty()) + { + JLOG(journal_.warn()) << "PRIORITY: trigger() sending " + << priorityHashes_.size() << " priority requests"; + + protocol::TMGetObjectByHash tmBH; + tmBH.set_query(true); + tmBH.set_type(protocol::TMGetObjectByHash::otSTATE_NODE); + tmBH.set_ledgerhash(hash_.begin(), hash_.size()); + + for (auto const& h : priorityHashes_) + { + JLOG(journal_.warn()) << "PRIORITY: requesting node " << h; + protocol::TMIndexedObject* io = tmBH.add_objects(); + io->set_hash(h.begin(), h.size()); + if (mSeq != 0) + io->set_ledgerseq(mSeq); + } + + // Send to all peers in our peer set + auto packet = std::make_shared(tmBH, protocol::mtGET_OBJECTS); + auto const& peerIds = mPeerSet->getPeerIds(); + std::size_t sentCount = 0; + for (auto id : peerIds) + { + if (auto p = app_.overlay().findPeerByShortID(id)) + { + p->send(packet); + ++sentCount; + } + } + JLOG(journal_.warn()) << "PRIORITY: sent to " << sentCount << " peers"; + + priorityHashes_.clear(); + } + protocol::TMGetLedger tmGL; tmGL.set_ledgerhash(hash_.begin(), hash_.size()); diff --git a/src/ripple/app/ledger/impl/InboundLedgers.cpp b/src/ripple/app/ledger/impl/InboundLedgers.cpp index ad9dc9d0f..fb1a0b676 100644 --- a/src/ripple/app/ledger/impl/InboundLedgers.cpp +++ b/src/ripple/app/ledger/impl/InboundLedgers.cpp @@ -202,6 +202,37 @@ public: return nullptr; } + void + addPriorityNode(std::uint32_t ledgerSeq, uint256 const& nodeHash) override + { + std::shared_ptr inbound; + { + ScopedLockType sl(mLock); + // Find inbound ledger by sequence (need to iterate) + for (auto const& [hash, ledger] : mLedgers) + { + if (ledger->getSeq() == ledgerSeq && !ledger->isFailed() && + !ledger->isComplete()) + { + inbound = ledger; + break; + } + } + } + + if (inbound) + { + inbound->addPriorityHash(nodeHash); + JLOG(j_.warn()) << "PRIORITY: added node " << nodeHash + << " for ledger seq " << ledgerSeq; + } + else + { + JLOG(j_.warn()) << "PRIORITY: no inbound ledger for seq " + << ledgerSeq << " (node " << nodeHash << ")"; + } + } + /* This gets called when "We got some data from an inbound ledger" diff --git a/src/ripple/app/ledger/impl/LedgerMaster.cpp b/src/ripple/app/ledger/impl/LedgerMaster.cpp index 35f56add3..28a9a759f 100644 --- a/src/ripple/app/ledger/impl/LedgerMaster.cpp +++ b/src/ripple/app/ledger/impl/LedgerMaster.cpp @@ -187,6 +187,7 @@ LedgerMaster::LedgerMaster( beast::Journal journal) : app_(app) , m_journal(journal) + , jPartialSync_(app.journal("PartialSync")) , mLedgerHistory(collector, app) , standalone_(app_.config().standalone()) , fetch_depth_( @@ -1009,11 +1010,22 @@ LedgerMaster::checkAccept(uint256 const& hash, std::uint32_t seq) auto validations = app_.validators().negativeUNLFilter( app_.getValidations().getTrustedForLedger(hash, seq)); valCount = validations.size(); - if (valCount >= app_.validators().quorum()) + auto const quorum = app_.validators().quorum(); + + JLOG(m_journal.warn()) + << "checkAccept: hash=" << hash << " seq=" << seq + << " valCount=" << valCount << " quorum=" << quorum + << " mLastValidLedger.seq=" << mLastValidLedger.second; + + if (valCount >= quorum) { std::lock_guard ml(m_mutex); if (seq > mLastValidLedger.second) + { + JLOG(m_journal.warn()) + << "checkAccept: setting mLastValidLedger to seq=" << seq; mLastValidLedger = std::make_pair(hash, seq); + } } if (seq == mValidLedgerSeq) diff --git a/src/ripple/overlay/impl/PeerImp.cpp b/src/ripple/overlay/impl/PeerImp.cpp index adb05ddf6..62c8d8c65 100644 --- a/src/ripple/overlay/impl/PeerImp.cpp +++ b/src/ripple/overlay/impl/PeerImp.cpp @@ -2751,6 +2751,12 @@ PeerImp::onMessage(std::shared_ptr const& m) bool pLDo = true; bool progress = false; + // For state/transaction node requests, store directly to db + // (not fetch pack) so partial sync queries can find them immediately + bool const directStore = + packet.type() == protocol::TMGetObjectByHash::otSTATE_NODE || + packet.type() == protocol::TMGetObjectByHash::otTRANSACTION_NODE; + for (int i = 0; i < packet.objects_size(); ++i) { const protocol::TMIndexedObject& obj = packet.objects(i); @@ -2783,10 +2789,33 @@ PeerImp::onMessage(std::shared_ptr const& m) { uint256 const hash{obj.hash()}; - app_.getLedgerMaster().addFetchPack( - hash, - std::make_shared( - obj.data().begin(), obj.data().end())); + if (directStore) + { + // Store directly to node store for immediate + // availability + auto const hotType = + (packet.type() == + protocol::TMGetObjectByHash::otSTATE_NODE) + ? hotACCOUNT_NODE + : hotTRANSACTION_NODE; + + JLOG(p_journal_.warn()) + << "PRIORITY: received node " << hash << " for seq " + << pLSeq << " storing to db"; + + app_.getNodeStore().store( + hotType, + Blob(obj.data().begin(), obj.data().end()), + hash, + pLSeq); + } + else + { + app_.getLedgerMaster().addFetchPack( + hash, + std::make_shared( + obj.data().begin(), obj.data().end())); + } } } } diff --git a/src/ripple/rpc/impl/RPCHelpers.cpp b/src/ripple/rpc/impl/RPCHelpers.cpp index 5b2e31774..8722ed03e 100644 --- a/src/ripple/rpc/impl/RPCHelpers.cpp +++ b/src/ripple/rpc/impl/RPCHelpers.cpp @@ -38,6 +38,7 @@ #include #include +#include namespace ripple { namespace RPC { @@ -663,10 +664,64 @@ getLedger(T& ledger, LedgerShortcut shortcut, Context& context) { auto [hash, seq] = context.ledgerMaster.getLastValidatedLedger(); JLOG(context.j.warn()) - << "Partial sync: getValidatedLedger null, trying hash=" << hash - << " seq=" << seq; + << "Partial sync: getValidatedLedger null, trying trusted hash=" + << hash << " seq=" << seq; + + // If no trusted validations yet, try network-observed ledger + if (hash.isZero()) + { + std::tie(hash, seq) = + context.ledgerMaster.getNetworkObservedLedger(); + JLOG(context.j.warn()) + << "Partial sync: trying network-observed hash=" << hash + << " seq=" << seq; + + // Poll-wait for validations to arrive (up to ~10 seconds) + if (hash.isZero()) + { + for (int i = 0; i < 100 && hash.isZero(); ++i) + { + std::this_thread::sleep_for( + std::chrono::milliseconds(100)); + std::tie(hash, seq) = + context.ledgerMaster.getNetworkObservedLedger(); + } + if (hash.isNonZero()) + { + JLOG(context.j.warn()) + << "Partial sync: got network-observed hash=" + << hash << " seq=" << seq; + } + } + } + if (hash.isNonZero()) + { ledger = context.app.getInboundLedgers().getPartialLedger(hash); + // If no InboundLedger exists yet, trigger acquisition and wait + if (!ledger) + { + JLOG(context.j.warn()) + << "Partial sync: acquiring ledger " << hash; + context.app.getInboundLedgers().acquire( + hash, seq, InboundLedger::Reason::CONSENSUS); + + // Poll-wait for the ledger header (up to ~10 seconds) + int i = 0; + for (; i < 100 && !ledger; ++i) + { + std::this_thread::sleep_for( + std::chrono::milliseconds(100)); + ledger = + context.app.getInboundLedgers().getPartialLedger( + hash); + } + JLOG(context.j.warn()) + << "Partial sync: poll-wait completed after " << i + << " iterations, ledger=" + << (ledger ? "found" : "null"); + } + } JLOG(context.j.warn()) << "Partial sync: getPartialLedger returned " << (ledger ? "ledger" : "null"); } diff --git a/src/ripple/shamap/Family.h b/src/ripple/shamap/Family.h index fea5545d3..c0c93b9ac 100644 --- a/src/ripple/shamap/Family.h +++ b/src/ripple/shamap/Family.h @@ -81,9 +81,14 @@ public: * * @param refNum Sequence of ledger to acquire. * @param nodeHash Hash of missing node to report in throw. + * @param prioritize If true, prioritize fetching this specific node + * (used by partial sync mode for RPC queries). */ virtual void - missingNodeAcquireBySeq(std::uint32_t refNum, uint256 const& nodeHash) = 0; + missingNodeAcquireBySeq( + std::uint32_t refNum, + uint256 const& nodeHash, + bool prioritize = false) = 0; /** Acquire ledger that has a missing node by ledger hash * diff --git a/src/ripple/shamap/NodeFamily.h b/src/ripple/shamap/NodeFamily.h index f20abccce..f12f2765d 100644 --- a/src/ripple/shamap/NodeFamily.h +++ b/src/ripple/shamap/NodeFamily.h @@ -83,7 +83,10 @@ public: reset() override; void - missingNodeAcquireBySeq(std::uint32_t seq, uint256 const& hash) override; + missingNodeAcquireBySeq( + std::uint32_t seq, + uint256 const& hash, + bool prioritize = false) override; void missingNodeAcquireByHash(uint256 const& hash, std::uint32_t seq) override diff --git a/src/ripple/shamap/ShardFamily.h b/src/ripple/shamap/ShardFamily.h index de809cf58..dfd7cb243 100644 --- a/src/ripple/shamap/ShardFamily.h +++ b/src/ripple/shamap/ShardFamily.h @@ -89,8 +89,10 @@ public: reset() override; void - missingNodeAcquireBySeq(std::uint32_t seq, uint256 const& nodeHash) - override; + missingNodeAcquireBySeq( + std::uint32_t seq, + uint256 const& nodeHash, + bool prioritize = false) override; void missingNodeAcquireByHash(uint256 const& hash, std::uint32_t seq) override diff --git a/src/ripple/shamap/impl/NodeFamily.cpp b/src/ripple/shamap/impl/NodeFamily.cpp index 1752db06a..e7604d5c5 100644 --- a/src/ripple/shamap/impl/NodeFamily.cpp +++ b/src/ripple/shamap/impl/NodeFamily.cpp @@ -66,9 +66,13 @@ NodeFamily::reset() } void -NodeFamily::missingNodeAcquireBySeq(std::uint32_t seq, uint256 const& nodeHash) +NodeFamily::missingNodeAcquireBySeq( + std::uint32_t seq, + uint256 const& nodeHash, + bool prioritize) { - JLOG(j_.error()) << "Missing node in " << seq; + JLOG(j_.error()) << "Missing node in " << seq << " hash=" << nodeHash + << (prioritize ? " [PRIORITY]" : ""); if (app_.config().reporting()) { std::stringstream ss; @@ -77,6 +81,10 @@ NodeFamily::missingNodeAcquireBySeq(std::uint32_t seq, uint256 const& nodeHash) Throw(ss.str()); } + // Add priority for the specific node hash needed by the query + if (prioritize && nodeHash.isNonZero()) + app_.getInboundLedgers().addPriorityNode(seq, nodeHash); + std::unique_lock lock(maxSeqMutex_); if (maxSeq_ == 0) { diff --git a/src/ripple/shamap/impl/SHAMap.cpp b/src/ripple/shamap/impl/SHAMap.cpp index 6b721b4cc..928ebb48e 100644 --- a/src/ripple/shamap/impl/SHAMap.cpp +++ b/src/ripple/shamap/impl/SHAMap.cpp @@ -197,15 +197,20 @@ SHAMap::finishFetch( constexpr auto timeout = 30s; auto const deadline = steady_clock::now() + timeout; + // Linear backoff for re-requests: 50ms, 100ms, 150ms... up to + // 2s + auto nextRequestDelay = 50ms; + constexpr auto maxRequestDelay = 2000ms; + constexpr auto backoffStep = 50ms; + auto nextRequestTime = steady_clock::now() + nextRequestDelay; + JLOG(journal_.debug()) << "finishFetch: waiting for node " << hash; while (steady_clock::now() < deadline) { - // Yield and reschedule - allows other work to happen - // and gives time for the node to arrive - if (!coro->postAndYield()) - break; // Job queue is stopping + // Sleep for the poll interval + std::this_thread::sleep_for(pollInterval); // Try to fetch from cache/db again if (auto obj = f_.db().fetchNodeObject( @@ -220,8 +225,17 @@ SHAMap::finishFetch( return node; } - // Re-request in case it got lost - f_.missingNodeAcquireBySeq(ledgerSeq_, hash.as_uint256()); + // Re-request with priority using linear backoff + auto now = steady_clock::now(); + if (now >= nextRequestTime) + { + f_.missingNodeAcquireBySeq( + ledgerSeq_, hash.as_uint256(), true /*prioritize*/); + // Increase delay for next request (linear backoff) + if (nextRequestDelay < maxRequestDelay) + nextRequestDelay += backoffStep; + nextRequestTime = now + nextRequestDelay; + } } JLOG(journal_.warn()) diff --git a/src/ripple/shamap/impl/ShardFamily.cpp b/src/ripple/shamap/impl/ShardFamily.cpp index f22d4152e..0fcb91fbb 100644 --- a/src/ripple/shamap/impl/ShardFamily.cpp +++ b/src/ripple/shamap/impl/ShardFamily.cpp @@ -153,11 +153,17 @@ ShardFamily::reset() } void -ShardFamily::missingNodeAcquireBySeq(std::uint32_t seq, uint256 const& nodeHash) +ShardFamily::missingNodeAcquireBySeq( + std::uint32_t seq, + uint256 const& nodeHash, + bool prioritize) { - std::ignore = nodeHash; JLOG(j_.error()) << "Missing node in ledger sequence " << seq; + // Add priority for the specific node hash needed by the query + if (prioritize && nodeHash.isNonZero()) + app_.getInboundLedgers().addPriorityNode(seq, nodeHash); + std::unique_lock lock(maxSeqMutex_); if (maxSeq_ == 0) { diff --git a/src/test/app/LedgerReplay_test.cpp b/src/test/app/LedgerReplay_test.cpp index ce0deb621..0968f1439 100644 --- a/src/test/app/LedgerReplay_test.cpp +++ b/src/test/app/LedgerReplay_test.cpp @@ -132,6 +132,11 @@ public: return {}; } + virtual void + addPriorityNode(std::uint32_t ledgerSeq, uint256 const& nodeHash) override + { + } + virtual bool gotLedgerData( LedgerHash const& ledgerHash, diff --git a/src/test/shamap/common.h b/src/test/shamap/common.h index 591e95803..64e40edb5 100644 --- a/src/test/shamap/common.h +++ b/src/test/shamap/common.h @@ -105,8 +105,10 @@ public: } void - missingNodeAcquireBySeq(std::uint32_t refNum, uint256 const& nodeHash) - override + missingNodeAcquireBySeq( + std::uint32_t refNum, + uint256 const& nodeHash, + bool prioritize = false) override { Throw("missing node"); }