Nomination prototype

src/herder/Herder.h

@@ -172,6 +172,11 @@ class Herder
     // sender in the pending or recent tx sets.
     virtual SequenceNumber getMaxSeqInPendingTxs(AccountID const&) = 0;
 
+    // If the node has asked about a tx set with the given hash, return the
+    // highest slot asked about for that tx set, otherwise return 0. Used for
+    // deciding what get tx set requests to accept.
+    virtual uint64 getLastSeenSlotIndexForTxSet(Hash const& hash) = 0;
+
     // Returns sequence number for most recent completed checkpoint that the
     // node knows about, as derived from
     // trackingConsensusLedgerIndex

src/herder/HerderImpl.cpp

@@ -795,6 +795,17 @@ HerderImpl::recvSCPEnvelope(SCPEnvelope const& envelope)
         {
             std::string txt("FETCHING");
             ZoneText(txt.c_str(), txt.size());
+            // If nomination, we should still process this.
+            if (envelope.statement.pledges.type() == SCP_ST_NOMINATE)
+            {
+                auto qSetHash = Slot::getCompanionQuorumSetHashFromStatement(
+                    envelope.statement);
+                auto qSet = mApp.getHerder().getQSet(qSetHash);
+                if (qSet != nullptr)
+                {
+                    processSCPQueue();
+                }
+            }
         }
         else if (status == Herder::ENVELOPE_STATUS_PROCESSED)
         {
@@ -1220,6 +1231,12 @@ HerderImpl::getMaxSeqInPendingTxs(AccountID const& acc)
     return mTransactionQueue.getAccountTransactionQueueInfo(acc).mMaxSeq;
 }
 
+uint64
+HerderImpl::getLastSeenSlotIndexForTxSet(Hash const& hash)
+{
+    return mPendingEnvelopes.getLastSeenSlotIndexForTxSet(hash);
+}
+
 uint32_t
 HerderImpl::getMostRecentCheckpointSeq()
 {

src/herder/HerderImpl.h

@@ -150,6 +150,7 @@ class HerderImpl : public Herder
                                      SCPStatement const& newSt) override;
 
     SequenceNumber getMaxSeqInPendingTxs(AccountID const&) override;
+    uint64 getLastSeenSlotIndexForTxSet(Hash const& hash) override;
 
     uint32_t getMostRecentCheckpointSeq() override;
 

src/herder/HerderSCPDriver.cpp

@@ -137,10 +137,9 @@ class SCPHerderEnvelopeWrapper : public SCPEnvelopeWrapper
             }
             else
             {
-                throw std::runtime_error(fmt::format(
-                    FMT_STRING("SCPHerderEnvelopeWrapper: Wrapping an unknown "
-                               "tx set {} from envelope"),
-                    hexAbbrev(txSetH)));
+                CLOG_TRACE(Herder,
+                           "unknown tx set {}, we might still able to vote",
+                           hexAbbrev(txSetH));
             }
         }
     }
@@ -314,10 +313,10 @@ HerderSCPDriver::validateValueHelper(uint64_t slotIndex, StellarValue const& b,
 
     if (!txSet)
     {
-        CLOG_ERROR(Herder, "validateValue i:{} unknown txSet {}", slotIndex,
+        CLOG_TRACE(Herder, "validateValue i:{} unknown txSet {}", slotIndex,
                    hexAbbrev(txSetHash));
-
-        res = SCPDriver::kInvalidValue;
+        res =
+            nomination ? SCPDriver::kVoteToNominate : SCPDriver::kInvalidValue;
     }
     else if (!checkAndCacheTxSetValid(txSet, closeTimeOffset))
     {
@@ -439,7 +438,7 @@ HerderSCPDriver::extractValidValue(uint64_t slotIndex, Value const& value)
     return res;
 }
 
-// value marshaling
+// value marshalling
 
 std::string
 HerderSCPDriver::toShortString(NodeID const& pk) const
@@ -1178,10 +1177,8 @@ class SCPHerderValueWrapper : public ValueWrapper
         mTxSet = mHerder.getTxSet(sv.txSetHash);
         if (!mTxSet)
         {
-            throw std::runtime_error(fmt::format(
-                FMT_STRING(
-                    "SCPHerderValueWrapper tried to bind an unknown tx set {}"),
-                hexAbbrev(sv.txSetHash)));
+            CLOG_TRACE(Herder, "unknown tx set {}, we might vote to nominate",
+                       hexAbbrev(sv.txSetHash));
         }
     }
 };

src/herder/PendingEnvelopes.cpp

@@ -234,13 +234,21 @@ PendingEnvelopes::addTxSet(Hash const& hash, uint64 lastSeenSlotIndex,
     mTxSetFetcher.recv(hash, mFetchTxSetTimer);
 }
 
+uint64
+
+PendingEnvelopes::getLastSeenSlotIndexForTxSet(Hash const& hash)
+{
+    auto lastSeenSlotIndex = mTxSetFetcher.getLastSeenSlotIndex(hash);
+    return lastSeenSlotIndex;
+}
+
 bool
 PendingEnvelopes::recvTxSet(Hash const& hash, TxSetFrameConstPtr txset)
 {
     ZoneScoped;
     CLOG_TRACE(Herder, "Got TxSet {}", hexAbbrev(hash));
 
-    auto lastSeenSlotIndex = mTxSetFetcher.getLastSeenSlotIndex(hash);
+    auto lastSeenSlotIndex = getLastSeenSlotIndexForTxSet(hash);
     if (lastSeenSlotIndex == 0)
     {
         return false;
@@ -361,9 +369,26 @@ PendingEnvelopes::recvSCPEnvelope(SCPEnvelope const& envelope)
         }
         else
         {
-            // else just keep waiting for it to come in
-            // and refresh fetchers as needed
-            startFetch(envelope);
+            if (envelope.statement.pledges.type() == SCP_ST_NOMINATE)
+            {
+                auto qSetH = Slot::getCompanionQuorumSetHashFromStatement(
+                    envelope.statement);
+                auto mQSet = mApp.getHerder().getQSet(qSetH);
+                if (mQSet != nullptr)
+                {
+                    // The qset has been fetched.
+                    // Not all tx sets have been fetched,
+                    // but we should process this nomination statement
+                    // in case we can vote for some/all of them.
+                    envelopeReady(envelope);
+                }
+            }
+            else
+            {
+                // else just keep waiting for it to come in
+                // and refresh fetchers as needed
+                startFetch(envelope);
+            }
         }
 
         return Herder::ENVELOPE_STATUS_FETCHING;

src/herder/PendingEnvelopes.h

@@ -164,6 +164,8 @@ class PendingEnvelopes
     TxSetFrameConstPtr putTxSet(Hash const& hash, uint64 slot,
                                 TxSetFrameConstPtr txset);
 
+    uint64 getLastSeenSlotIndexForTxSet(Hash const& hash);
+
     /**
      * Check if @p txset identified by @p hash was requested before from peers.
      * If not, ignores that @p txset. If it was requested, calls

src/herder/TxSetFrame.h

@@ -159,6 +159,7 @@ class TxSetFrame : public NonMovableOrCopyable
     // Test helper that only checks the XDR structure validitiy without
     // validating internal transactions.
     virtual bool checkValidStructure() const;
+    void computeContentsHashForTesting();
     static TxSetFrameConstPtr
     makeFromTransactions(Transactions txs, Application& app,
                          uint64_t lowerBoundCloseTimeOffset,

src/herder/test/HerderTests.cpp

@@ -4346,6 +4346,7 @@ externalize(SecretKey const& sk, LedgerManager& lm, HerderImpl& herder,
     txsPhases.emplace_back(sorobanTxs);
 
     auto txSet = TxSetFrame::makeFromTransactions(txsPhases, app, 0, 0);
+
     herder.getPendingEnvelopes().putTxSet(txSet->getContentsHash(), ledgerSeq,
                                           txSet);
 
@@ -5339,3 +5340,159 @@ TEST_CASE("exclude transactions by operation type", "[herder]")
                 TransactionQueue::AddResult::ADD_STATUS_PENDING);
     }
 }
+
+TEST_CASE("delay sending DONT_HAVE", "[herder]")
+{
+    VirtualClock clock;
+    auto const numNodes = 2;
+    std::vector<std::shared_ptr<Application>> apps;
+    std::chrono::milliseconds const epsilon{1};
+    auto s = SecretKey::pseudoRandomForTesting();
+
+    for (auto i = 0; i < numNodes; i++)
+    {
+        Config cfg = getTestConfig(i);
+        cfg.FLOOD_DEMAND_BACKOFF_DELAY_MS = std::chrono::milliseconds(200);
+        cfg.FLOOD_DEMAND_PERIOD_MS = std::chrono::milliseconds(200);
+        // Using a small tq set size such as 50 may lead to an unexpectedly
+        // small advert/demand size limit.
+        cfg.MANUAL_CLOSE = false;
+        cfg.TESTING_UPGRADE_MAX_TX_SET_SIZE = 1000;
+        cfg.SEND_DONT_HAVE_DELAY = std::chrono::milliseconds{300};
+        // Setting validators
+        cfg.QUORUM_SET.validators.emplace_back(s.getPublicKey());
+        // cfg.LEDGER_PROTOCOL_VERSION
+        apps.push_back(createTestApplication(clock, cfg));
+    }
+
+    auto connection =
+        std::make_shared<LoopbackPeerConnection>(*apps[0], *apps[1]);
+    testutil::crankFor(clock, std::chrono::seconds(5));
+    REQUIRE(connection->getInitiator()->isAuthenticated());
+    REQUIRE(connection->getAcceptor()->isAuthenticated());
+
+    // TODO: maybe replace these lambdas with static definitions?
+    auto createTxn = [](auto n) {
+        StellarMessage txn;
+        txn.type(TRANSACTION);
+        Memo memo(MEMO_TEXT);
+        memo.text() = "tx" + std::to_string(n);
+        txn.transaction().v0().tx.memo = memo;
+
+        return std::make_shared<StellarMessage>(txn);
+    };
+
+    auto createGetTxSetMessage = [](uint256 const& setID) {
+        StellarMessage getTxSet;
+        getTxSet.type(GET_TX_SET);
+        getTxSet.txSetHash() = setID;
+
+        return std::make_shared<StellarMessage>(getTxSet);
+    };
+
+    auto createTxSetMessage = [](TxSetFrameConstPtr txSet) {
+        StellarMessage msg;
+        msg.type(GENERALIZED_TX_SET);
+        txSet->toXDR(msg.generalizedTxSet());
+        return std::make_shared<StellarMessage const>(msg);
+    };
+
+    using TxPair = std::pair<Value, TxSetFrameConstPtr>;
+    using SVUpgrades = decltype(StellarValue::upgrades);
+    auto root = TestAccount::createRoot(*apps[0]);
+
+    auto makeTxUpgradePair = [&](TxSetFrameConstPtr txSet, uint64_t closeTime,
+                                 SVUpgrades const& upgrades) {
+        StellarValue sv = apps[0]->getHerder().makeStellarValue(
+            txSet->getContentsHash(), closeTime, upgrades, root.getSecretKey());
+        auto v = xdr::xdr_to_opaque(sv);
+        return TxPair{v, txSet};
+    };
+
+    auto makeTxPair = [&](TxSetFrameConstPtr txSet, uint64_t closeTime) {
+        return makeTxUpgradePair(txSet, closeTime, emptyUpgradeSteps);
+    };
+
+    auto makeNominationEnvelope = [&s, &apps](TxPair const& p, Hash qSetHash,
+                                              uint64_t slotIndex) {
+        // herder must want the TxSet before receiving it, so we are sending it
+        // a fake envelope
+        auto envelope = SCPEnvelope{};
+        envelope.statement.slotIndex = slotIndex;
+        envelope.statement.pledges.type(SCP_ST_NOMINATE);
+        envelope.statement.pledges.nominate().votes.push_back(p.first);
+        envelope.statement.pledges.nominate().quorumSetHash = qSetHash;
+        envelope.statement.nodeID = s.getPublicKey();
+        static_cast<HerderImpl&>(apps[0]->getHerder())
+            .signEnvelope(s, envelope);
+        return envelope;
+    };
+
+    // Create txn set.
+    auto tx = createTxn(1);
+    std::vector<TransactionFrameBasePtr> txs = {
+        TransactionFrameBase::makeTransactionFromWire(apps[0]->getNetworkID(),
+                                                      tx->transaction()),
+    };
+    auto txnSetFrame = TxSetFrame::makeFromTransactions(txs, *apps[0], 0, 0);
+
+    // Sending get tx set message.
+    auto txSetHash = txnSetFrame->getContentsHash();
+    auto getTxSet = createGetTxSetMessage(txSetHash);
+    connection->getAcceptor()->sendMessage(getTxSet, false);
+
+    REQUIRE(!apps[0]->getHerder().getTxSet(txSetHash));
+    REQUIRE(!apps[1]->getHerder().getTxSet(txSetHash));
+
+    // Should not add to pending getTxSet requests while we wait before
+    // retrying.
+    testutil::crankFor(clock, epsilon);
+
+    auto closedTime = apps[0]
+                          ->getLedgerManager()
+                          .getLastClosedLedgerHeader()
+                          .header.scpValue.closeTime +
+                      1;
+
+    auto p = makeTxPair(txnSetFrame, closedTime);
+
+    auto recvTxPairEnvelope = [&apps, makeNominationEnvelope](TxPair p) {
+        auto envelope = makeNominationEnvelope(
+            p, {}, apps[0]->getHerder().trackingConsensusLedgerIndex() + 1);
+
+        REQUIRE(apps[0]->getHerder().recvSCPEnvelope(envelope) ==
+                Herder::ENVELOPE_STATUS_FETCHING);
+        REQUIRE(apps[1]->getHerder().recvSCPEnvelope(envelope) ==
+                Herder::ENVELOPE_STATUS_FETCHING);
+    };
+
+    SECTION("tx set not received before timeout.")
+    {
+        recvTxPairEnvelope(p);
+        testutil::crankFor(clock, std::chrono::milliseconds{300});
+
+        // Receives tx set.
+        auto txSetMsg = createTxSetMessage(txnSetFrame);
+        connection->getInitiator()->recvMessage(*txSetMsg);
+
+        testutil::crankFor(clock, std::chrono::seconds{1});
+        // Check peer does not have the txn set hash.
+        REQUIRE(apps[0]->getHerder().getTxSet(txSetHash));
+        REQUIRE(!apps[1]->getHerder().getTxSet(txSetHash));
+    }
+
+    SECTION("tx set received before timeout.")
+    {
+        recvTxPairEnvelope(p);
+        // Receives tx set.
+        auto txSetMsg = createTxSetMessage(txnSetFrame);
+        connection->getInitiator()->recvMessage(*txSetMsg);
+
+        // Pending getTxSet requests are empty since node already has the tx
+        // set.
+        testutil::crankFor(clock, std::chrono::seconds{1});
+        // Check peer has the txn set hash.
+        REQUIRE(apps[0]->getHerder().getTxSet(txSetHash));
+        REQUIRE(apps[1]->getHerder().getTxSet(txSetHash));
+    }
+}

src/main/Config.cpp

@@ -156,6 +156,7 @@ Config::Config() : NODE_SEED(SecretKey::random())
     AUTOMATIC_MAINTENANCE_COUNT = 400;
     // automatic self-check happens once every 3 hours
     AUTOMATIC_SELF_CHECK_PERIOD = std::chrono::seconds{3 * 60 * 60};
+    SEND_DONT_HAVE_DELAY = std::chrono::milliseconds{200};
     ARTIFICIALLY_GENERATE_LOAD_FOR_TESTING = false;
     ARTIFICIALLY_ACCELERATE_TIME_FOR_TESTING = false;
     ARTIFICIALLY_SET_CLOSE_TIME_FOR_TESTING = 0;
@@ -224,6 +225,8 @@ Config::Config() : NODE_SEED(SecretKey::random())
     FLOOD_ADVERT_PERIOD_MS = std::chrono::milliseconds(100);
     FLOOD_DEMAND_BACKOFF_DELAY_MS = std::chrono::milliseconds(500);
 
+    TX_SET_BACKOFF_DELAY_MS = std::chrono::milliseconds(1500);
+
     MAX_BATCH_WRITE_COUNT = 1024;
     MAX_BATCH_WRITE_BYTES = 1 * 1024 * 1024;
     PREFERRED_PEERS_ONLY = false;
@@ -1142,6 +1145,11 @@ Config::processConfig(std::shared_ptr<cpptoml::table> t)
                 AUTOMATIC_SELF_CHECK_PERIOD =
                     std::chrono::seconds{readInt<uint32_t>(item)};
             }
+            else if (item.first == "SEND_DONT_HAVE_DELAY")
+            {
+                SEND_DONT_HAVE_DELAY =
+                    std::chrono::seconds{readInt<uint32_t>(item)};
+            }
             else if (item.first == "MANUAL_CLOSE")
             {
                 MANUAL_CLOSE = readBool(item);
@@ -1272,6 +1280,11 @@ Config::processConfig(std::shared_ptr<cpptoml::table> t)
                         "bad value for FLOOD_ARB_TX_DAMPING_FACTOR");
                 }
             }
+            else if (item.first == "TX_SET_BACKOFF_DELAY_MS")
+            {
+                TX_SET_BACKOFF_DELAY_MS =
+                    std::chrono::milliseconds(readInt<int>(item, 1));
+            }
             else if (item.first == "PREFERRED_PEERS")
             {
                 PREFERRED_PEERS = readArray<std::string>(item);