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

@@ -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);

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<uint256> mRecentNodes;
 
+    // Priority nodes to fetch immediately (for partial sync queries)
+    std::set<uint256> priorityHashes_;
+
     SHAMapAddNode mStats;
 
     // Data we have received from peers

src/ripple/app/ledger/InboundLedgers.h

@@ -64,6 +64,15 @@ public:
     virtual std::shared_ptr<Ledger const>
     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

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<uint256, LedgerIndex>
+    getNetworkObservedLedger()
+    {
+        std::lock_guard lock(m_mutex);
+        return mNetworkObservedLedger;
+    }
+
     // Returns the minimum ledger sequence in SQL database, if any.
     std::optional<LedgerIndex>
     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<uint256, LedgerIndex> mLastValidLedger{uint256(), 0};
 
+    // Network-observed ledger from any validations (for partial sync).
+    std::pair<uint256, LedgerIndex> mNetworkObservedLedger{uint256(), 0};
+
     LedgerHistory mLedgerHistory;
 
     CanonicalTXSet mHeldTransactions{uint256()};

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<Peer> 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<Message>(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());
 

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<InboundLedger> 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"

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)

src/ripple/overlay/impl/PeerImp.cpp

@@ -2751,6 +2751,12 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> 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<protocol::TMGetObjectByHash> const& m)
                 {
                     uint256 const hash{obj.hash()};
 
-                    app_.getLedgerMaster().addFetchPack(
-                        hash,
-                        std::make_shared<Blob>(
-                            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<Blob>(
+                                obj.data().begin(), obj.data().end()));
+                    }
                 }
             }
         }

src/ripple/rpc/impl/RPCHelpers.cpp

@@ -38,6 +38,7 @@
 
 #include <ripple/resource/Fees.h>
 #include <regex>
+#include <thread>
 
 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");
         }

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
      *

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

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

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<std::runtime_error>(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<std::mutex> lock(maxSeqMutex_);
     if (maxSeq_ == 0)
     {

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

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<std::mutex> lock(maxSeqMutex_);
     if (maxSeq_ == 0)
     {

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,

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<std::runtime_error>("missing node");
     }