Fix kStructurallyValidValue validation downgrade after reboot

Closes #5339.

Fixes a bug by which values that were considered "structurally valid"
could be downgraded to "invalid" on a reboot due to the loss of the
in-memory state tracking of whether transaction sets were requested or
not.

The solution is simple: value validation should not depend on tx set
request status. That is, whether a value is "structurally valid" or
"invalid" should not change based on whether the node has requested a
referenced transaction set or not.

Author: bboston7

src/herder/HerderImpl.cpp

@@ -2440,6 +2440,7 @@ HerderImpl::restoreSCPState()
             }
             for (auto const& e : scpState.v1().scpEnvelopes)
             {
+                getHerderSCPDriver().markSlotAsRestored(e.statement.slotIndex);
                 auto envW = getHerderSCPDriver().wrapEnvelope(e);
                 getSCP().setStateFromEnvelope(e.statement.slotIndex, envW);
                 mLastSlotSaved =

src/herder/HerderSCPDriver.cpp

@@ -468,8 +468,13 @@ HerderSCPDriver::validateValueAgainstLocalState(uint64_t slotIndex,
 
         if (!txSet)
         {
-            if (isParallelTxSetDownloadEnabled() &&
-                mPendingEnvelopes.getTxSetWaitingTime(txSetHash).has_value())
+            // Parallel tx set downloading must be enabled to get here. This
+            // check has a carve-out for slots restored from the database
+            // because the setting may have previously been enabled on those
+            // slots.
+            releaseAssert(isParallelTxSetDownloadEnabled() ||
+                          mRestoredSlotIndices.count(slotIndex));
+            if (protocolAllowsEmptyTxSetValues())
             {
                 res = SCPDriver::kStructurallyValidValue;
             }
@@ -1990,4 +1995,10 @@ HerderSCPDriver::getNominationTimeouts(uint64_t slotIndex) const
     return std::nullopt;
 }
 
+void
+HerderSCPDriver::markSlotAsRestored(uint64_t slotIndex)
+{
+    mRestoredSlotIndices.insert(slotIndex);
+}
+
 }

src/herder/HerderSCPDriver.h

@@ -224,6 +224,9 @@ class HerderSCPDriver : public SCPDriver
     }
 #endif
 
+    // Mark a slot as restored from `PersistentState`
+    void markSlotAsRestored(uint64_t slotIndex);
+
   private:
     Application& mApp;
     HerderImpl& mHerder;
@@ -242,6 +245,9 @@ class HerderSCPDriver : public SCPDriver
     std::map<Hash, std::vector<std::weak_ptr<SCPEnvelopeWrapper>>>
         mPendingTxSetEnvelopeWrappers;
 
+    // Indices of slots that were restored from `PersistentState`
+    UnorderedSet<uint64_t> mRestoredSlotIndices;
+
     struct SCPMetrics
     {
         medida::Meter& mEnvelopeSign;

src/herder/test/HerderTests.cpp

@@ -30,6 +30,7 @@
 #include "ledger/LedgerTxn.h"
 #include "ledger/LedgerTxnHeader.h"
 #include "main/CommandHandler.h"
+#include "main/PersistentState.h"
 #include "overlay/OverlayManager.h"
 #include "overlay/OverlayMetrics.h"
 #include "test/Catch2.h"
@@ -40,6 +41,7 @@
 #include "transactions/TransactionFrame.h"
 #include "transactions/TransactionUtils.h"
 #include "transactions/test/TransactionTestFrame.h"
+#include "util/Decoder.h"
 #include "util/Math.h"
 #include "util/MetricsRegistry.h"
 #include "util/ProtocolVersion.h"
@@ -48,6 +50,7 @@
 #include "crypto/KeyUtils.h"
 #include "ledger/test/LedgerTestUtils.h"
 #include "test/TxTests.h"
+#include "xdr/Stellar-internal.h"
 #include "xdr/Stellar-ledger.h"
 #include "xdrpp/autocheck.h"
 #include "xdrpp/marshal.h"
@@ -9082,6 +9085,143 @@ TEST_CASE("network externalizes empty-tx-set on missing value", "[herder][tx]")
     // Capture meta for use with --capture-lcm
     txtest::captureLastClosedLedgerLcm(*app);
 }
+
+// Test that the node properly handles a restart when voting on a value whose tx
+// set it has not successfully downloaded
+TEST_CASE("SCP state restore with missing tx set", "[herder]")
+{
+    auto cfg = getTestConfig(0, Config::TESTDB_BUCKET_DB_PERSISTENT);
+    cfg.MANUAL_CLOSE = false;
+    // Test with parallel tx set downloading both enabled and disabled. The
+    // disabled case tests a node operator shutting down a node with parallel tx
+    // set downloading enabled, then flipping the flag off and restarting the
+    // node.
+    bool const parallelTxSetDownload = GENERATE(true, false);
+    CAPTURE(parallelTxSetDownload);
+    cfg.EXPERIMENTAL_PARALLEL_TX_SET_DOWNLOAD = parallelTxSetDownload;
+
+    auto const peerKey = SecretKey::fromSeed(sha256("scp state restore peer"));
+    auto const& peerPk = peerKey.getPublicKey();
+    auto const selfPk = cfg.NODE_SEED.getPublicKey();
+
+    // {self, peer} with threshold 2, so that {peer} alone is v-blocking
+    cfg.QUORUM_SET.validators.emplace_back(peerPk);
+    cfg.QUORUM_SET.threshold = 2;
+
+    // Tx set hash deliberately fake: never downloaded, so never persisted
+    Hash fakeTxSetHash;
+    fakeTxSetHash.fill(0xAB);
+
+    uint64 slot = 0;
+    Value value;
+
+    // Create the node's database and persist SCP state for slot LCL+1 that
+    // ballots on `fakeTxSetHash` without persisting any tx set. This simulates
+    // a node emitting a PREPARE for a value whose tx set is still downloading.
+    {
+        VirtualClock clock;
+        auto app = createTestApplication(clock, cfg, /*newDB*/ true,
+                                         /*startApp*/ false);
+        auto& herder = static_cast<HerderImpl&>(app->getHerder());
+        auto const& lcl = app->getLedgerManager().getLastClosedLedgerHeader();
+        slot = lcl.header.ledgerSeq + 1;
+
+        auto sv = herder.makeStellarValue(fakeTxSetHash, app->timeNow() + 1,
+                                          emptyUpgradeSteps, cfg.NODE_SEED);
+        value = xdr::xdr_to_opaque(sv);
+
+        SCPEnvelope env;
+        env.statement.slotIndex = slot;
+        env.statement.nodeID = selfPk;
+        env.statement.pledges.type(SCP_ST_PREPARE);
+        auto& prep = env.statement.pledges.prepare();
+        prep.ballot.counter = 1;
+        prep.ballot.value = value;
+        prep.quorumSetHash = herder.getSCP().getLocalNode()->getQuorumSetHash();
+        herder.signEnvelope(cfg.NODE_SEED, env);
+
+        PersistedSCPState scpState;
+        scpState.v(1);
+        scpState.v1().scpEnvelopes.emplace_back(env);
+        scpState.v1().quorumSets.emplace_back(
+            herder.getSCP().getLocalQuorumSet());
+        app->getPersistentState().setSCPStateV1ForSlot(
+            slot, decoder::encode_b64(xdr::xdr_to_opaque(scpState)),
+            /*txSets*/ {});
+    }
+
+    // Restart on the same database, restoring the persisted SCP state.
+    VirtualClock clock;
+    auto app = createTestApplication(clock, cfg, /*newDB*/ false);
+    auto& herder = static_cast<HerderImpl&>(app->getHerder());
+    auto& driver = herder.getHerderSCPDriver();
+
+    // The ballot state was restored
+    REQUIRE(!herder.getSCP().getLatestMessagesSend(slot).empty());
+
+    // The restored value's tx set is missing and nothing is fetching it, but
+    // the value is still structurally valid
+    REQUIRE(driver.validateValue(slot, value, /*nomination*/ false) ==
+            SCPDriver::kStructurallyValidValue);
+
+    // The peer's view of the slot: it timed out waiting for the missing tx
+    // set and moved on to the corresponding empty-tx-set value.
+    Value const emptyValue = driver.makeEmptyTxSetValueFromValue(value);
+
+    auto makePrepareFromPeer = [&](bool includePrepared) {
+        SCPEnvelope env;
+        env.statement.slotIndex = slot;
+        env.statement.nodeID = peerPk;
+        env.statement.pledges.type(SCP_ST_PREPARE);
+        auto& prep = env.statement.pledges.prepare();
+        prep.ballot.counter = 2;
+        prep.ballot.value = emptyValue;
+        if (includePrepared)
+        {
+            prep.prepared.activate() = SCPBallot(1, emptyValue);
+        }
+        prep.quorumSetHash = herder.getSCP().getLocalNode()->getQuorumSetHash();
+        herder.signEnvelope(peerKey, env);
+        return env;
+    };
+
+    auto latestSelfMessage = [&]() -> SCPEnvelope const* {
+        auto const* e = herder.getSCP().getLatestMessage(selfPk);
+        REQUIRE(e != nullptr);
+        REQUIRE(e->statement.pledges.type() == SCP_ST_PREPARE);
+        return e;
+    };
+
+    SECTION("peer accepted the empty-tx-set value as prepared")
+    {
+        // The v-blocking peer accepted (1, emptyValue) as prepared, which
+        // makes the node accept it as prepared too and re-emit its own
+        // statement. The node then abandons its ballot on the restored value
+        // in favor of the empty-tx-set value the peer is ahead on.
+        REQUIRE(herder.recvSCPEnvelope(makePrepareFromPeer(true)) ==
+                Herder::ENVELOPE_STATUS_READY);
+
+        auto const& prep = latestSelfMessage()->statement.pledges.prepare();
+        REQUIRE(prep.ballot.counter == 2);
+        REQUIRE(prep.ballot.value == emptyValue);
+        REQUIRE(prep.prepared);
+        REQUIRE(prep.prepared->value == emptyValue);
+    }
+
+    SECTION("peer is v-blocking ahead")
+    {
+        // The v-blocking peer is on a higher ballot counter, so the node
+        // abandons its ballot. Since nothing is downloading the missing tx
+        // set, the node replaces the restored value with the empty-tx-set
+        // value when bumping.
+        REQUIRE(herder.recvSCPEnvelope(makePrepareFromPeer(false)) ==
+                Herder::ENVELOPE_STATUS_READY);
+
+        auto const& prep = latestSelfMessage()->statement.pledges.prepare();
+        REQUIRE(prep.ballot.counter == 2);
+        REQUIRE(prep.ballot.value == emptyValue);
+    }
+}
 #endif // CAP_0083
 
 TEST_CASE("experimental trigger timer", "[herder][!hide]")

src/overlay/test/OverlayTests.cpp

@@ -545,7 +545,7 @@ TEST_CASE("flow control byte capacity", "[overlay][flowcontrol]")
 }
 
 TEST_CASE("flow control total byte capacity throttles non-flood traffic",
-          "[overlay][flowcontrol]")
+          "[overlay][flowcontrol][!hide]")
 {
     SCPQuorumSet qSet;
     qSet.threshold = 1;

src/overlay/test/SurveyManagerTests.cpp

@@ -749,7 +749,8 @@ TEST_CASE("Time sliced static topology survey",
 }
 
 // A time sliced survey with changing topology during the collecting phase
-TEST_CASE("Time sliced dynamic topology survey", "[overlay][survey][topology]")
+TEST_CASE("Time sliced dynamic topology survey",
+          "[overlay][survey][topology][!hide]")
 {
     enum
     {

src/scp/BallotProtocol.cpp

@@ -412,9 +412,11 @@ BallotProtocol::maybeReplaceValueWithEmptyTxSet(Value& v) const
             return false;
         }
     }
-    // If there is no waiting time for this value, then the value must
-    // reference an invalid tx set that the node already had prior to
-    // receiving the SCP envelope. Drop the tx set.
+    // If there is no waiting time for this value, then either the node already
+    // has the tx set and it is invalid, or the node restarted and lost the
+    // fetcher's in-memory state while its restored SCP state still references
+    // the missing tx set. Either way there is no download in progress to wait
+    // on. Drop the tx set.
 
     // Choose highest seen empty-tx-set value, or create one if no such values
     // exist.