WIP: Enable downloading of transaction sets in parallel with early SCP stages

docs/metrics.md

@@ -187,13 +187,16 @@ scp.pending.discarded                     | counter   | number of discarded enve
 scp.pending.fetching                      | counter   | number of incomplete envelopes
 scp.pending.processed                     | counter   | number of already processed envelopes
 scp.pending.ready                         | counter   | number of envelopes ready to process
+scp.skip.externalized                     | counter   | number of times the local node externalized a skip-ledger value
+scp.skip.value-replaced                   | counter   | number of times the ballot protocol swapped a value for a skip-ledger value
 scp.sync.lost                             | meter     | validator lost sync
 scp.timeout.nominate                      | meter     | timeouts in nomination
 scp.timeout.prepare                       | meter     | timeouts in ballot protocol
 scp.timing.nominated                      | timer     | time spent in nomination
 scp.timing.externalized                   | timer     | time spent in ballot protocol
 scp.timing.first-to-self-externalize-lag  | timer     | delay between first externalize message and local node externalizing
 scp.timing.self-to-others-externalize-lag | timer     | delay between local node externalizing and later externalize messages from other nodes
+scp.timing.ballot-blocked-on-txset        | timer     | time balloting was blocked waiting for a txset download (milliseconds)
 scp.value.invalid                         | meter     | SCP value is invalid
 scp.value.valid                           | meter     | SCP value is valid
 scp.slot.values-referenced                | histogram | number of values referenced per consensus round

docs/stellar-core_example.cfg

@@ -158,6 +158,12 @@ PEER_TIMEOUT=30
 # time when authenticated.
 PEER_STRAGGLER_TIMEOUT=120
 
+# TODO: Update these docs vv
+# TX_SET_DOWNLOAD_TIMEOUT (Integer) default 5000
+# Time in milliseconds before a node gives up waiting on a transaction set and
+# votes to skip the ledger.
+TX_SET_DOWNLOAD_TIMEOUT=5000
+
 # MAX_BATCH_WRITE_COUNT (Integer) default 1024
 # How many messages can this server send at once to a peer
 MAX_BATCH_WRITE_COUNT=1024

src/herder/Herder.cpp

@@ -22,4 +22,6 @@ uint32 const Herder::SCP_EXTRA_LOOKBACK_LEDGERS = 3u;
 std::chrono::minutes const Herder::TX_SET_GC_DELAY(1);
 std::chrono::minutes const Herder::CHECK_FOR_DEAD_NODES_MINUTES(15);
 uint32 const Herder::FLOW_CONTROL_BYTES_EXTRA_BUFFER(2000);
+
+Hash const Herder::SKIP_LEDGER_HASH{};
 }

src/herder/Herder.h

@@ -16,9 +16,15 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <variant>
 
 namespace stellar
 {
+
+// Returned by getTxSet to distinguish "skip" values (no real tx set)
+// from "not yet downloaded" (nullptr).
+struct SkipTxSet {};
+using TxSetResult = std::variant<TxSetXDRFrameConstPtr, SkipTxSet>;
 class Application;
 class XDROutputFileStream;
 
@@ -79,6 +85,9 @@ class Herder
 
     static std::chrono::minutes const TX_SET_GC_DELAY;
 
+    // TODO: Docs
+    static Hash const SKIP_LEDGER_HASH;
+
     enum State
     {
         // Starting up, no state is known
@@ -146,7 +155,7 @@ class Herder
 #endif
     virtual void peerDoesntHave(stellar::MessageType type,
                                 uint256 const& itemID, Peer::pointer peer) = 0;
-    virtual TxSetXDRFrameConstPtr getTxSet(Hash const& hash) = 0;
+    virtual TxSetResult getTxSet(Hash const& hash) = 0;
     virtual SCPQuorumSetPtr getQSet(Hash const& qSetHash) = 0;
 
     // We are learning about a new envelope.

src/herder/HerderImpl.cpp

@@ -305,8 +305,19 @@ HerderImpl::processExternalized(uint64 slotIndex, StellarValue const& value,
                      slotIndex, hexAbbrev(value.txSetHash));
     }
 
-    TxSetXDRFrameConstPtr externalizedSet =
-        mPendingEnvelopes.getTxSet(value.txSetHash);
+    auto result = mPendingEnvelopes.getTxSet(value.txSetHash);
+    TxSetXDRFrameConstPtr externalizedSet;
+    if (std::holds_alternative<SkipTxSet>(result))
+    {
+        auto const& ov = value.ext.originalValue();
+        externalizedSet = TxSetXDRFrame::makeEmpty(
+            ov.previousLedgerHash, ov.previousLedgerVersion);
+    }
+    else
+    {
+        externalizedSet = std::get<TxSetXDRFrameConstPtr>(result);
+    }
+    releaseAssert(externalizedSet != nullptr);
 
     // save the SCP messages in the database
     if (mApp.getConfig().MODE_STORES_HISTORY_MISC)
@@ -866,6 +877,9 @@ HerderImpl::recvSCPEnvelope(SCPEnvelope const& envelope)
         return Herder::ENVELOPE_STATUS_SKIPPED_SELF;
     }
 
+    // This call fetches everything. Will only return ENVELOPE_STATUS_READY once
+    // everything is fetched though! Will need a new status to allow it to
+    // proceed to nomination at least, I think.
     auto status = mPendingEnvelopes.recvSCPEnvelope(envelope);
     if (status == Herder::ENVELOPE_STATUS_READY)
     {
@@ -880,10 +894,26 @@ HerderImpl::recvSCPEnvelope(SCPEnvelope const& envelope)
     }
     else
     {
-        if (status == Herder::ENVELOPE_STATUS_FETCHING)
+        SCPStatementType type = envelope.statement.pledges.type();
+        // Allow parallel tx set downloading if the node is in sync and this is
+        // a NOMINATE or PREPARE message. Technically both of these criteria
+        // should be properly handled downstream, but this provides some
+        // additional assurance.
+        if (mApp.getState() == Application::State::APP_SYNCED_STATE &&
+            status == Herder::ENVELOPE_STATUS_FETCHING &&
+            (type == SCP_ST_NOMINATE || type == SCP_ST_PREPARE))
         {
             std::string txt("FETCHING");
             ZoneText(txt.c_str(), txt.size());
+
+            // If we have the quorum set, then proceed without the tx set.
+            auto qSetHash = Slot::getCompanionQuorumSetHashFromStatement(
+                envelope.statement);
+            auto maybeQSet = mApp.getHerder().getQSet(qSetHash);
+            if (maybeQSet)
+            {
+                processSCPQueue();
+            }
         }
         else if (status == Herder::ENVELOPE_STATUS_PROCESSED)
         {
@@ -1079,6 +1109,7 @@ void
 HerderImpl::processSCPQueueUpToIndex(uint64 slotIndex)
 {
     ZoneScoped;
+    CLOG_TRACE(Proto, "Processing SCP queue up to index {}", slotIndex);
     while (true)
     {
         SCPEnvelopeWrapperPtr envW = mPendingEnvelopes.pop(slotIndex);
@@ -1316,7 +1347,7 @@ HerderImpl::peerDoesntHave(MessageType type, uint256 const& itemID,
     mPendingEnvelopes.peerDoesntHave(type, itemID, peer);
 }
 
-TxSetXDRFrameConstPtr
+TxSetResult
 HerderImpl::getTxSet(Hash const& hash)
 {
     return mPendingEnvelopes.getTxSet(hash);
@@ -2082,11 +2113,16 @@ HerderImpl::persistSCPState(uint64 slot)
         // saves transaction sets referred by the statement
         for (auto const& h : getValidatedTxSetHashes(e))
         {
-            auto txSet = mPendingEnvelopes.getTxSet(h);
-            if (txSet && !mApp.getPersistentState().hasTxSet(h))
+            auto result = mPendingEnvelopes.getTxSet(h);
+            if (auto* txSetPtr =
+                    std::get_if<TxSetXDRFrameConstPtr>(&result))
             {
-                txSets.insert(std::make_pair(h, txSet));
+                if (*txSetPtr && !mApp.getPersistentState().hasTxSet(h))
+                {
+                    txSets.insert(std::make_pair(h, *txSetPtr));
+                }
             }
+            // SkipTxSet: nothing to persist
         }
         Hash qsHash = Slot::getCompanionQuorumSetHashFromStatement(e.statement);
         SCPQuorumSetPtr qSet = mPendingEnvelopes.getQSet(qsHash);
@@ -2604,11 +2640,32 @@ bool
 HerderImpl::verifyStellarValueSignature(StellarValue const& sv)
 {
     ZoneScoped;
-    auto [b, _] = PubKeyUtils::verifySig(
-        sv.ext.lcValueSignature().nodeID, sv.ext.lcValueSignature().signature,
-        xdr::xdr_to_opaque(mApp.getNetworkID(), ENVELOPE_TYPE_SCPVALUE,
-                           sv.txSetHash, sv.closeTime));
-    return b;
+    switch (sv.ext.v())
+    {
+    case STELLAR_VALUE_BASIC:
+        // This function should never be called with an unsigned value
+        releaseAssert(false);
+    case STELLAR_VALUE_SIGNED:
+        return PubKeyUtils::verifySig(sv.ext.lcValueSignature().nodeID,
+                                      sv.ext.lcValueSignature().signature,
+                                      xdr::xdr_to_opaque(mApp.getNetworkID(),
+                                                         ENVELOPE_TYPE_SCPVALUE,
+                                                         sv.txSetHash,
+                                                         sv.closeTime))
+            .valid;
+    case STELLAR_VALUE_SKIP:
+    {
+        auto const& ov = sv.ext.originalValue();
+        return PubKeyUtils::verifySig(
+                   ov.lcValueSignature.nodeID, ov.lcValueSignature.signature,
+                   xdr::xdr_to_opaque(mApp.getNetworkID(),
+                                      ENVELOPE_TYPE_SCPVALUE, ov.txSetHash,
+                                      sv.closeTime))
+            .valid;
+    }
+    default:
+        releaseAssert(false);
+    }
 }
 
 StellarValue

src/herder/HerderImpl.h

@@ -154,7 +154,7 @@ class HerderImpl : public Herder
     bool recvTxSet(Hash const& hash, TxSetXDRFrameConstPtr txset) override;
     void peerDoesntHave(MessageType type, uint256 const& itemID,
                         Peer::pointer peer) override;
-    TxSetXDRFrameConstPtr getTxSet(Hash const& hash) override;
+    TxSetResult getTxSet(Hash const& hash) override;
     SCPQuorumSetPtr getQSet(Hash const& qSetHash) override;
 
     void processSCPQueue();
@@ -228,7 +228,7 @@ class HerderImpl : public Herder
     // helper function to sign envelopes
     void signEnvelope(SecretKey const& s, SCPEnvelope& envelope);
 
-    // helper function to verify SCPValues are signed
+    // helper function to verify SCPValues signatures
     bool verifyStellarValueSignature(StellarValue const& sv);
 
     size_t getMaxQueueSizeOps() const override;