Enable downloading of transaction sets in parallel with early SCP stages (#5209)

This PR adds support for CAP-0083 and enables downloading of
transactions sets in parallel with SCP up until setting `c` during
balloting.

Author: sisuresh

common.mk

@@ -46,3 +46,7 @@ endif # ENABLE_NEXT_PROTOCOL_VERSION_UNSAFE_FOR_PRODUCTION
 if CAP_0071
 AM_CPPFLAGS += -DCAP_0071
 endif
+
+if CAP_0083
+AM_CPPFLAGS += -DCAP_0083
+endif

configure.ac

@@ -564,6 +564,9 @@ AM_CONDITIONAL(ENABLE_NEXT_PROTOCOL_VERSION_UNSAFE_FOR_PRODUCTION,
 AM_CONDITIONAL(CAP_0071,
     [test x$enable_next_protocol_version_unsafe_for_production = xyes])
 
+AM_CONDITIONAL(CAP_0083,
+    [test x$enable_next_protocol_version_unsafe_for_production = xyes])
+
 AC_PATH_PROG(CARGO, cargo)
 if test x"$CARGO" = x; then
   AC_MSG_ERROR([cannot find cargo, needed for rust code])

docs/metrics.md

@@ -188,13 +188,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.empty-tx-set.externalized             | counter   | number of times the local node externalized an empty-tx-set value
+scp.empty-tx-set.value-replaced           | counter   | number of times the ballot protocol swapped a value for an empty-tx-set 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

@@ -163,6 +163,12 @@ PEER_TIMEOUT=30
 # time when authenticated.
 PEER_STRAGGLER_TIMEOUT=120
 
+# TX_SET_DOWNLOAD_TIMEOUT (Integer) default 5000
+# Time in milliseconds before a validator gives up waiting on a transaction set
+# and votes to drop the tx set from the upcoming ledger. Does nothing without
+# `EXPERIMENTAL_PARALLEL_TX_SET_DOWNLOAD` enabled.
+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
@@ -274,6 +280,10 @@ BACKGROUND_OVERLAY_PROCESSING = true
 # performance on multicore machines. Note that this is not compatible with SQLite.
 EXPERIMENTAL_PARALLEL_LEDGER_APPLY = false
 
+# EXPERIMENTAL_PARALLEL_TX_SET_DOWNLOAD (bool) default false
+# Allow downloading of transaction sets in parallel with SCP
+EXPERIMENTAL_PARALLEL_TX_SET_DOWNLOAD = false
+
 # BACKGROUND_TX_SIG_VERIFICATION (bool) default true
 # Check signatures in the background for transactions received
 # over the network. Does nothing if `BACKGROUND_OVERLAY_PROCESSING` is not

src/Makefile.am

@@ -100,6 +100,9 @@ XDR_FEATURE_FLAGS =
 if CAP_0071
 XDR_FEATURE_FLAGS += -DCAP_0071
 endif
+if CAP_0083
+XDR_FEATURE_FLAGS += -DCAP_0083
+endif
 
 SUFFIXES = .x .h .rs
 .x.h:

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::EMPTY_TX_SET_HASH{};
 }

src/herder/Herder.h

@@ -16,9 +16,17 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <variant>
 
 namespace stellar
 {
+
+// Returned by getTxSet to distinguish "empty tx set" values (no real tx set)
+// from "not yet downloaded" (nullptr).
+struct EmptyTxSet
+{
+};
+using TxSetResult = std::variant<TxSetXDRFrameConstPtr, EmptyTxSet>;
 class Application;
 class XDROutputFileStream;
 
@@ -79,6 +87,9 @@ class Herder
 
     static std::chrono::minutes const TX_SET_GC_DELAY;
 
+    // Hash value indicating a CAP-0083 explicitly empty-tx-set value
+    static Hash const EMPTY_TX_SET_HASH;
+
     enum State
     {
         // Starting up, no state is known
@@ -147,7 +158,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

@@ -328,8 +328,23 @@ 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<EmptyTxSet>(result))
+    {
+#ifdef CAP_0083
+        auto const& ov = value.ext.proposedValue();
+        externalizedSet = TxSetXDRFrame::makeEmpty(ov.previousLedgerHash,
+                                                   ov.previousLedgerVersion);
+#else
+        releaseAssert(false);
+#endif // CAP_0083
+    }
+    else
+    {
+        externalizedSet = std::get<TxSetXDRFrameConstPtr>(result);
+    }
+    releaseAssert(externalizedSet != nullptr);
 
     {
         ZoneNamedN(updateSCPHistoryZone, "update SCP history", true);
@@ -937,7 +952,7 @@ HerderImpl::recvSCPEnvelope(SCPEnvelope const& envelope)
 
 Herder::EnvelopeStatus
 HerderImpl::recvSCPEnvelope(SCPEnvelope const& envelope,
-                            const SCPQuorumSet& qset,
+                            SCPQuorumSet const& qset,
                             TxSetXDRFrameConstPtr txset)
 {
     ZoneScoped;
@@ -1388,7 +1403,7 @@ HerderImpl::peerDoesntHave(MessageType type, uint256 const& itemID,
     mPendingEnvelopes.peerDoesntHave(type, itemID, peer);
 }
 
-TxSetXDRFrameConstPtr
+TxSetResult
 HerderImpl::getTxSet(Hash const& hash)
 {
     return mPendingEnvelopes.getTxSet(hash);
@@ -1639,6 +1654,18 @@ HerderImpl::triggerNextLedger(uint32_t ledgerSeqToTrigger,
         return;
     }
 
+#ifdef BUILD_TESTS
+    if (mApp.getConfig().TESTING_NOMINATE_RANDOM_VALUES &&
+        getHerderSCPDriver().protocolAllowsEmptyTxSetValues())
+    {
+        txSetHash = HashUtils::pseudoRandomForTesting();
+        CLOG_INFO(Herder,
+                  "TESTING_NOMINATE_RANDOM_VALUES: nominating slot {} "
+                  "with random tx-set hash {}",
+                  slotIndex, hexAbbrev(txSetHash));
+    }
+#endif
+
     StellarValue newProposedValue = makeStellarValue(
         txSetHash, nextCloseTime, newUpgrades, mApp.getConfig().NODE_SEED);
     mHerderSCPDriver.nominate(slotIndex, newProposedValue, proposedSet,
@@ -2165,11 +2192,15 @@ 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));
+                }
             }
+            // EmptyTxSet: nothing to persist
         }
         Hash qsHash = Slot::getCompanionQuorumSetHashFromStatement(e.statement);
         SCPQuorumSetPtr qSet = mPendingEnvelopes.getQSet(qsHash);
@@ -2708,11 +2739,34 @@ 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;
+#ifdef CAP_0083
+    case STELLAR_VALUE_EMPTY_TX_SET:
+    {
+        auto const& ov = sv.ext.proposedValue();
+        return PubKeyUtils::verifySig(
+                   ov.lcValueSignature.nodeID, ov.lcValueSignature.signature,
+                   xdr::xdr_to_opaque(mApp.getNetworkID(),
+                                      ENVELOPE_TYPE_SCPVALUE, ov.txSetHash,
+                                      sv.closeTime))
+            .valid;
+    }
+#endif // CAP_0083
+    default:
+        releaseAssert(false);
+    }
 }
 
 StellarValue

src/herder/HerderImpl.h

@@ -155,7 +155,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;
 
     // process ready SCP messages. This may trigger the node to externalze new
@@ -234,7 +234,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;