Drop duplicate SCP messages as early as possible

src/main/AppConnector.cpp

@@ -7,6 +7,7 @@
 #include "overlay/BanManager.h"
 #include "overlay/OverlayManager.h"
 #include "overlay/OverlayMetrics.h"
+#include "overlay/Peer.h"
 #include "util/Timer.h"
 
 namespace stellar
@@ -34,7 +35,6 @@ AppConnector::getLedgerManager()
 OverlayManager&
 AppConnector::getOverlayManager()
 {
-    releaseAssert(threadIsMain());
     return mApp.getOverlayManager();
 }
 
@@ -129,4 +129,10 @@ AppConnector::getOverlayMetrics()
     return mApp.getOverlayManager().getOverlayMetrics();
 }
 
+bool
+AppConnector::isScheduled(std::shared_ptr<MsgCapacityTracker> msgTracker)
+{
+    return mApp.getOverlayManager().isScheduled(msgTracker);
+}
+
 }

src/main/AppConnector.h

@@ -14,6 +14,7 @@ struct OverlayMetrics;
 class SorobanNetworkConfig;
 class SorobanMetrics;
 struct LedgerTxnDelta;
+class MsgCapacityTracker;
 
 // Helper class to isolate access to Application; all function helpers must
 // either be called from main or be thread-sade
@@ -51,5 +52,6 @@ class AppConnector
     Config const& getConfig() const;
     bool overlayShuttingDown() const;
     OverlayMetrics& getOverlayMetrics();
+    bool isScheduled(std::shared_ptr<MsgCapacityTracker> msgTracker);
 };
 }

src/overlay/Floodgate.cpp

@@ -56,10 +56,10 @@ Floodgate::clearBelow(uint32_t maxLedger)
 }
 
 bool
-Floodgate::addRecord(StellarMessage const& msg, Peer::pointer peer, Hash& index)
+Floodgate::addRecord(StellarMessage const& msg, Peer::pointer peer,
+                     Hash const& index)
 {
     ZoneScoped;
-    index = xdrBlake2(msg);
     if (mShuttingDown)
     {
         return false;

src/overlay/Floodgate.h

@@ -57,7 +57,7 @@ class Floodgate
     // returns true if this is a new record
     // fills msgID with msg's hash
     bool addRecord(StellarMessage const& msg, Peer::pointer fromPeer,
-                   Hash& msgID);
+                   Hash const& msgID);
 
     // returns true if msg was sent to at least one peer
     // The hash required for transactions

src/overlay/FlowControl.cpp

@@ -240,7 +240,6 @@ SendMoreCapacity
 FlowControl::endMessageProcessing(StellarMessage const& msg)
 {
     ZoneScoped;
-    releaseAssert(threadIsMain());
     std::lock_guard<std::mutex> guard(mFlowControlMutex);
 
     mFloodDataProcessed += mFlowControlCapacity.releaseLocalCapacity(msg);
@@ -560,7 +559,6 @@ bool
 FlowControl::stopThrottling()
 {
     std::lock_guard<std::mutex> guard(mFlowControlMutex);
-    releaseAssert(threadIsMain());
     if (mLastThrottle)
     {
         CLOG_DEBUG(Overlay, "Stop throttling reading from peer {}",

src/overlay/OverlayManager.h

@@ -90,7 +90,7 @@ class OverlayManager
     // Returns true if this is a new message
     // fills msgID with msg's hash
     virtual bool recvFloodedMsgID(StellarMessage const& msg, Peer::pointer peer,
-                                  Hash& msgID) = 0;
+                                  Hash const& msgID) = 0;
 
     bool
     recvFloodedMsg(StellarMessage const& msg, Peer::pointer peer)
@@ -100,8 +100,8 @@ class OverlayManager
     }
 
     // Process incoming transaction, pass it down to the transaction queue
-    virtual void recvTransaction(StellarMessage const& msg,
-                                 Peer::pointer peer) = 0;
+    virtual void recvTransaction(StellarMessage const& msg, Peer::pointer peer,
+                                 Hash const& index) = 0;
 
     // removes msgID from the floodgate's internal state
     // as it's not tracked anymore, calling "broadcast" with a (now forgotten)
@@ -211,5 +211,11 @@ class OverlayManager
     virtual ~OverlayManager()
     {
     }
+
+    // Is message already referenced by the scheduler
+    // This method is always called from one thread, therefore no cache
+    // synchorization is needed
+    virtual bool
+    isScheduled(std::shared_ptr<MsgCapacityTracker> tracker) const = 0;
 };
 }

src/overlay/OverlayManagerImpl.cpp

@@ -321,6 +321,7 @@ OverlayManagerImpl::OverlayManagerImpl(Application& app)
                      mApp.getConfig().TARGET_PEER_CONNECTIONS, mSurveyManager)
     , mResolvingPeersWithBackoff(true)
     , mResolvingPeersRetryCount(0)
+    , mScheduledMessages(100000, true)
 {
     mPeerSources[PeerType::INBOUND] = std::make_unique<RandomPeerSource>(
         mPeerManager, RandomPeerSource::nextAttemptCutoff(PeerType::INBOUND));
@@ -1140,15 +1141,37 @@ OverlayManagerImpl::shufflePeerList(std::vector<Peer::pointer>& peerList)
 
 bool
 OverlayManagerImpl::recvFloodedMsgID(StellarMessage const& msg,
-                                     Peer::pointer peer, Hash& msgID)
+                                     Peer::pointer peer, Hash const& msgID)
 {
     ZoneScoped;
     return mFloodGate.addRecord(msg, peer, msgID);
 }
 
+bool
+OverlayManagerImpl::isScheduled(
+    std::shared_ptr<MsgCapacityTracker> tracker) const
+{
+    releaseAssert(!threadIsMain() ||
+                  !mApp.getConfig().EXPERIMENTAL_BACKGROUND_OVERLAY_PROCESSING);
+    if (!tracker->maybeGetHash())
+    {
+        return false;
+    }
+    auto index = tracker->maybeGetHash().value();
+    if (mScheduledMessages.exists(index))
+    {
+        if (mScheduledMessages.get(index).lock())
+        {
+            return true;
+        }
+    }
+    mScheduledMessages.put(index, std::weak_ptr<MsgCapacityTracker>(tracker));
+    return false;
+}
+
 void
 OverlayManagerImpl::recvTransaction(StellarMessage const& msg,
-                                    Peer::pointer peer)
+                                    Peer::pointer peer, Hash const& index)
 {
     ZoneScoped;
     auto transaction = TransactionFrameBase::makeTransactionFromWire(
@@ -1157,8 +1180,7 @@ OverlayManagerImpl::recvTransaction(StellarMessage const& msg,
     {
         // record that this peer sent us this transaction
         // add it to the floodmap so that this peer gets credit for it
-        Hash msgID;
-        recvFloodedMsgID(msg, peer, msgID);
+        recvFloodedMsgID(msg, peer, index);
 
         mTxDemandsManager.recordTxPullLatency(transaction->getFullHash(), peer);
 
@@ -1171,7 +1193,7 @@ OverlayManagerImpl::recvTransaction(StellarMessage const& msg,
               addResult.code ==
                   TransactionQueue::AddResultCode::ADD_STATUS_DUPLICATE))
         {
-            forgetFloodedMsg(msgID);
+            forgetFloodedMsg(index);
             CLOG_DEBUG(Overlay,
                        "Peer::recvTransaction Discarded transaction {} from {}",
                        hexAbbrev(transaction->getFullHash()), peer->toString());

src/overlay/OverlayManagerImpl.h

@@ -113,9 +113,9 @@ class OverlayManagerImpl : public OverlayManager
 
     void clearLedgersBelow(uint32_t ledgerSeq, uint32_t lclSeq) override;
     bool recvFloodedMsgID(StellarMessage const& msg, Peer::pointer peer,
-                          Hash& msgID) override;
-    void recvTransaction(StellarMessage const& msg,
-                         Peer::pointer peer) override;
+                          Hash const& msgID) override;
+    void recvTransaction(StellarMessage const& msg, Peer::pointer peer,
+                         Hash const& index) override;
     void forgetFloodedMsg(Hash const& msgID) override;
     void recvTxDemand(FloodDemand const& dmd, Peer::pointer peer) override;
     bool broadcastMessage(std::shared_ptr<StellarMessage const> msg,
@@ -181,6 +181,8 @@ class OverlayManagerImpl : public OverlayManager
     std::future<ResolvedPeers> mResolvedPeers;
     bool mResolvingPeersWithBackoff;
     int mResolvingPeersRetryCount;
+    mutable RandomEvictionCache<Hash, std::weak_ptr<MsgCapacityTracker>>
+        mScheduledMessages;
 
     void triggerPeerResolution();
     std::pair<std::vector<PeerBareAddress>, bool>
@@ -214,5 +216,8 @@ class OverlayManagerImpl : public OverlayManager
     // Returns `true` iff the overlay can accept the outbound peer at `address`.
     // Logs whenever a peer cannot be accepted.
     bool canAcceptOutboundPeer(PeerBareAddress const& address) const;
+
+    bool
+    isScheduled(std::shared_ptr<MsgCapacityTracker> tracker) const override;
 };
 }