Add HAVE_TX_SET overlay message for targeted tx set fetching

docs/metrics.md

@@ -159,6 +159,11 @@ overlay.inbound.live                      | counter   | number of live inbound c
 overlay.outbound-queue.<X>                | timer     | time <X> traffic sits in flow-controlled queues
 overlay.outbound-queue.drop-<X>           | meter     | number of <X> messages dropped from flow-controlled queues
 overlay.item-fetcher.next-peer            | meter     | ask for item past the first one
+overlay.item-fetcher.claim-ask            | meter     | fetch ask targeted a peer believed to hold the item
+overlay.item-fetcher.claim-dropped        | meter     | HAVE_TX_SET dropped at admission: sending peer exceeded its budget for the current window
+overlay.item-fetcher.claim-grace-wait     | timer     | time a tx set fetch waited from tracker creation to its first ask
+overlay.item-fetcher.claim-grace-satisfied | meter    | tx set fetch's first ask targeted a believed holder
+overlay.item-fetcher.claim-grace-expired  | meter     | tx set fetch's first ask fell back to an SCP message relayer or random peer
 overlay.memory.flood-known                | counter   | number of known flooded entries
 overlay.message.broadcast                 | meter     | message broadcasted
 overlay.message.read                      | meter     | message received

src/herder/Herder.h

@@ -144,6 +144,12 @@ class Herder
     virtual bool recvSCPQuorumSet(Hash const& hash,
                                   SCPQuorumSet const& qset) = 0;
     virtual bool recvTxSet(Hash const& hash, TxSetXDRFrameConstPtr txset) = 0;
+
+    // A peer announced that it has the tx set with `hash`
+    virtual void recvHaveTxSet(Hash const& hash, Peer::pointer peer) = 0;
+
+    // Returns true iff the current protocol allows empty-tx-set values
+    virtual bool protocolAllowsEmptyTxSetValues() const = 0;
     // We are learning about a new transaction.
 #ifdef BUILD_TESTS
     // `isLoadgenTx` is true if the transaction was generated by the load

src/herder/HerderImpl.cpp

@@ -1018,15 +1018,23 @@ HerderImpl::sendSCPStateToPeer(uint32 ledgerSeq, Peer::pointer peer)
     bool log = true;
     auto maxSlots = Herder::LEDGER_VALIDITY_BRACKET;
 
-    auto sendSlot = [weakPeer = std::weak_ptr<Peer>(peer)](SCPEnvelope const& e,
-                                                           bool log) {
+    // Record of which tx sets we've already sent HAVE_TX_SET messages for to
+    // `peer`
+    auto claimedTxSets = std::make_shared<UnorderedSet<Hash>>();
+    auto sendSlot = [this, claimedTxSets, weakPeer = std::weak_ptr<Peer>(peer)](
+                        SCPEnvelope const& e, bool log) {
         // If in the process of shutting down, exit early
         auto peerPtr = weakPeer.lock();
         if (!peerPtr)
         {
             return false;
         }
 
+        // Tell the peer which of the referenced tx sets we hold.  Sent before
+        // the envelope, so the claim is already buffered when the envelope
+        // triggers a fetch.
+        mPendingEnvelopes.sendHaveTxSetClaims(e, peerPtr, *claimedTxSets);
+
         StellarMessage m;
         m.type(SCP_MESSAGE);
         m.envelope() = e;
@@ -1545,6 +1553,19 @@ HerderImpl::recvTxSet(Hash const& hash, TxSetXDRFrameConstPtr txset)
     return mPendingEnvelopes.recvTxSet(hash, txset);
 }
 
+void
+HerderImpl::recvHaveTxSet(Hash const& hash, Peer::pointer peer)
+{
+    ZoneScoped;
+    mPendingEnvelopes.recvHaveTxSet(hash, peer);
+}
+
+bool
+HerderImpl::protocolAllowsEmptyTxSetValues() const
+{
+    return mHerderSCPDriver.protocolAllowsEmptyTxSetValues();
+}
+
 void
 HerderImpl::peerDoesntHave(MessageType type, uint256 const& itemID,
                            Peer::pointer peer)

src/herder/HerderImpl.h

@@ -153,6 +153,8 @@ class HerderImpl : public Herder
 
     bool recvSCPQuorumSet(Hash const& hash, SCPQuorumSet const& qset) override;
     bool recvTxSet(Hash const& hash, TxSetXDRFrameConstPtr txset) override;
+    void recvHaveTxSet(Hash const& hash, Peer::pointer peer) override;
+    bool protocolAllowsEmptyTxSetValues() const override;
     void peerDoesntHave(MessageType type, uint256 const& itemID,
                         Peer::pointer peer) override;
     TxSetResult getTxSet(Hash const& hash) override;

src/herder/PendingEnvelopes.cpp

@@ -30,11 +30,15 @@ PendingEnvelopes::PendingEnvelopes(Application& app, HerderImpl& herder)
     , mHerder(herder)
     , mQsetCache(QSET_CACHE_SIZE)
     , mTxSetFetcher(
-          app, [](Peer::pointer peer, Hash hash) { peer->sendGetTxSet(hash); })
-    , mQuorumSetFetcher(app, [](Peer::pointer peer,
-                                Hash hash) { peer->sendGetQuorumSet(hash); })
+          app, [](Peer::pointer peer, Hash hash) { peer->sendGetTxSet(hash); },
+          ItemFetcherKind::TxSet)
+    , mQuorumSetFetcher(
+          app,
+          [](Peer::pointer peer, Hash hash) { peer->sendGetQuorumSet(hash); },
+          ItemFetcherKind::QuorumSet)
     , mTxSetCache(TXSET_CACHE_SIZE)
     , mValueSizeCache(TXSET_CACHE_SIZE + QSET_CACHE_SIZE)
+    , mAnnouncedTxSets(TXSET_CACHE_SIZE)
     , mRebuildQuorum(true)
     , mQuorumTracker(mApp.getConfig().NODE_SEED.getPublicKey())
     , mProcessedCount(
@@ -255,6 +259,82 @@ PendingEnvelopes::addTxSet(Hash const& hash, uint64 lastSeenSlotIndex,
 
     putTxSet(hash, lastSeenSlotIndex, txset);
     mTxSetFetcher.recv(hash, mFetchTxSetTimer);
+
+    if (lastSeenSlotIndex != 0)
+    {
+        // Announce tx set, so long as it was obtained via a live consensus path
+        // (not restored from the database; we would have already announced
+        // those)
+        maybeAnnounceHaveTxSet(hash);
+    }
+}
+
+void
+PendingEnvelopes::maybeAnnounceHaveTxSet(Hash const& hash)
+{
+    ZoneScoped;
+
+    if (mAnnouncedTxSets.exists(hash))
+    {
+        return;
+    }
+    mAnnouncedTxSets.put(hash, true);
+
+    auto msg = std::make_shared<StellarMessage>();
+    msg->type(HAVE_TX_SET);
+    msg->haveTxSet().txSetHash = hash;
+
+    for (auto const& peer : mApp.getOverlayManager().getAuthenticatedPeers())
+    {
+        // Send HAVE_TX_SET only if the peer supports it
+        if (peer.second->getRemoteOverlayVersion() >=
+            Peer::FIRST_OVERLAY_VERSION_SUPPORTING_HAVE_TX_SET)
+        {
+            peer.second->sendMessage(msg);
+        }
+    }
+}
+
+void
+PendingEnvelopes::sendHaveTxSetClaims(SCPEnvelope const& env,
+                                      Peer::pointer const& peer,
+                                      UnorderedSet<Hash>& alreadyClaimed)
+{
+    ZoneScoped;
+
+    // Peers on pre-HAVE_TX_SET overlay versions cannot parse the message
+    if (peer->getRemoteOverlayVersion() <
+        Peer::FIRST_OVERLAY_VERSION_SUPPORTING_HAVE_TX_SET)
+    {
+        return;
+    }
+
+    auto maybeHashes = getTxSetHashes(env);
+    if (!maybeHashes)
+    {
+        return;
+    }
+
+    for (auto const& hash : *maybeHashes)
+    {
+        if (hash == Herder::EMPTY_TX_SET_HASH || !hasTxSet(hash) ||
+            !alreadyClaimed.insert(hash).second)
+        {
+            continue;
+        }
+
+        auto msg = std::make_shared<StellarMessage>();
+        msg->type(HAVE_TX_SET);
+        msg->haveTxSet().txSetHash = hash;
+        peer->sendMessage(msg);
+    }
+}
+
+void
+PendingEnvelopes::recvHaveTxSet(Hash const& hash, Peer::pointer peer)
+{
+    ZoneScoped;
+    mTxSetFetcher.peerClaimsItem(hash, peer);
 }
 
 bool

src/herder/PendingEnvelopes.h

@@ -78,6 +78,9 @@ class PendingEnvelopes
     // keep track of txset/qset hash -> size pairs for quick access
     RandomEvictionCache<Hash, size_t> mValueSizeCache;
 
+    // tx set hashes already announced to peers via HAVE_TX_SET
+    RandomEvictionCache<Hash, bool> mAnnouncedTxSets;
+
     bool mRebuildQuorum;
     QuorumTracker mQuorumTracker;
 
@@ -118,6 +121,10 @@ class PendingEnvelopes
 
     void recordReceivedCost(SCPEnvelope const& env);
 
+    // Announce possession of the tx set with `hash` to all authenticated
+    // peers if not already announced
+    void maybeAnnounceHaveTxSet(Hash const& hash);
+
     UnorderedMap<NodeID, size_t> getCostPerValidator(uint64 slotIndex) const;
 
     // stops all pending downloads for slots outside the range
@@ -185,6 +192,22 @@ class PendingEnvelopes
      */
     bool recvTxSet(Hash const& hash, TxSetXDRFrameConstPtr txset);
 
+    /**
+     * A peer announced that it has the tx set identified by @p hash. Records
+     * the claim with the tx set fetcher so an active fetch can target that
+     * peer. No-op if the tx set is not being fetched.
+     */
+    void recvHaveTxSet(Hash const& hash, Peer::pointer peer);
+
+    /**
+     * Send HAVE_TX_SET to @p peer for every tx set referenced by @p env that
+     * is held in memory.
+     * Adds the hashes of all claims sent to @p peer to @p alreadyClaimed. Does
+     * not send claims for any hashes already present in @p alreadyClaimed.
+     */
+    void sendHaveTxSetClaims(SCPEnvelope const& env, Peer::pointer const& peer,
+                             UnorderedSet<Hash>& alreadyClaimed);
+
     // Returns true if the tx set is available locally (either in cache or
     // is an empty-tx-set hash which doesn't need fetching).
     bool hasTxSet(Hash const& hash) const;

src/overlay/ItemFetcher.cpp

@@ -14,8 +14,16 @@
 namespace stellar
 {
 
-ItemFetcher::ItemFetcher(Application& app, AskPeer askPeer)
-    : mApp(app), mAskPeer(askPeer)
+// Cap on the number of distinct hashes for which we buffer early HAVE_TX_SET
+// claims (claims for items not yet being tracked).
+static size_t const BUFFERED_CLAIMS_CACHE_SIZE = 1000;
+
+ItemFetcher::ItemFetcher(Application& app, AskPeer askPeer,
+                         ItemFetcherKind kind)
+    : mApp(app)
+    , mAskPeer(askPeer)
+    , mKind(kind)
+    , mBufferedClaims(BUFFERED_CLAIMS_CACHE_SIZE)
 {
 }
 
@@ -28,10 +36,27 @@ ItemFetcher::fetch(Hash const& itemHash, SCPEnvelope const& envelope)
     if (entryIt == mTrackers.end())
     { // not being tracked
         TrackerPtr tracker =
-            std::make_shared<Tracker>(mApp, itemHash, mAskPeer);
+            std::make_shared<Tracker>(mApp, itemHash, mAskPeer, mKind);
         mTrackers[itemHash] = tracker;
 
         tracker->listen(envelope);
+
+        // Seed any HAVE_TX_SET claims that arrived before this tracker existed
+        // so the first ask can target a known holder rather than blind-asking.
+        auto* buffered = mBufferedClaims.maybeGet(itemHash);
+        if (buffered)
+        {
+            for (auto const& [nodeID, weak] : *buffered)
+            {
+                if (auto peer = weak.lock())
+                {
+                    tracker->seedClaim(peer);
+                }
+            }
+            // Clear the consumed claims
+            buffered->clear();
+        }
+
         tracker->tryNextPeer();
     }
     else
@@ -156,6 +181,33 @@ ItemFetcher::doesntHave(Hash const& itemHash, Peer::pointer peer)
     }
 }
 
+void
+ItemFetcher::peerClaimsItem(Hash const& itemHash, Peer::pointer peer)
+{
+    ZoneScoped;
+    auto const& iter = mTrackers.find(itemHash);
+    if (iter != mTrackers.end())
+    {
+        iter->second->peerClaims(peer);
+    }
+    else
+    {
+        // Not yet tracking this item. Buffer the claim so the tracker can
+        // be seeded when it is created
+        auto* buffered = mBufferedClaims.maybeGet(itemHash);
+        if (buffered)
+        {
+            (*buffered)[peer->getPeerID()] = peer;
+        }
+        else
+        {
+            mBufferedClaims.put(itemHash,
+                                UnorderedMap<NodeID, std::weak_ptr<Peer>>{
+                                    {peer->getPeerID(), peer}});
+        }
+    }
+}
+
 void
 ItemFetcher::recv(Hash itemHash, medida::Timer& timer)
 {
@@ -197,5 +249,12 @@ ItemFetcher::getTracker(Hash const& h)
     }
     return it->second;
 }
+
+size_t
+ItemFetcher::getNumBufferedClaims(Hash const& itemHash)
+{
+    auto* buffered = mBufferedClaims.maybeGet(itemHash);
+    return buffered ? buffered->size() : 0;
+}
 #endif
 }

src/overlay/ItemFetcher.h

@@ -5,11 +5,15 @@
 #pragma once
 
 #include "overlay/Peer.h"
+#include "overlay/Tracker.h"
 #include "util/NonCopyable.h"
+#include "util/RandomEvictionCache.h"
 #include "util/Timer.h"
+#include "util/UnorderedMap.h"
 #include <functional>
 #include <map>
 #include <optional>
+#include <vector>
 
 namespace medida
 {
@@ -20,7 +24,6 @@ class Timer;
 namespace stellar
 {
 
-class Tracker;
 class TxSetXDRFrame;
 struct SCPQuorumSet;
 using SCPQuorumSetPtr = std::shared_ptr<SCPQuorumSet>;
@@ -41,9 +44,11 @@ class ItemFetcher : private NonMovableOrCopyable
     using TrackerPtr = std::shared_ptr<Tracker>;
 
     /**
-     * Create ItemFetcher that fetches data using @p askPeer delegate.
+     * Create ItemFetcher that fetches data using @p askPeer delegate. @p kind
+     * labels which fetcher this is.
      */
-    explicit ItemFetcher(Application& app, AskPeer askPeer);
+    explicit ItemFetcher(Application& app, AskPeer askPeer,
+                         ItemFetcherKind kind);
 
     /**
      * Fetch data identified by @p hash and needed by @p envelope. Multiple
@@ -92,6 +97,11 @@ class ItemFetcher : private NonMovableOrCopyable
      */
     void doesntHave(Hash const& itemHash, Peer::pointer peer);
 
+    /**
+     * Record that @p peer claims to have the data identified by @p itemHash.
+     */
+    void peerClaimsItem(Hash const& itemHash, Peer::pointer peer);
+
     /**
      * Called when data with given @p itemHash was received. All envelopes
      * added before with @see fetch and the same @p itemHash will be resent
@@ -101,6 +111,7 @@ class ItemFetcher : private NonMovableOrCopyable
 
 #ifdef BUILD_TESTS
     std::shared_ptr<Tracker> getTracker(Hash const& h);
+    size_t getNumBufferedClaims(Hash const& itemHash);
 #endif
 
   protected:
@@ -113,5 +124,9 @@ class ItemFetcher : private NonMovableOrCopyable
 
   private:
     AskPeer mAskPeer;
+    ItemFetcherKind mKind;
+    // HAVE_TX_SET claims received for hashes not yet being tracked.
+    RandomEvictionCache<Hash, UnorderedMap<NodeID, std::weak_ptr<Peer>>>
+        mBufferedClaims;
 };
 }