Continue to capture SCP messages for previous ledger in database

Closes #3769

Upon closing a ledger, nodes record SCP messages received during the
ledger period in the `scphistory` database table. However, any messages
about that ledger heard *after* close were dropped. This change captures
more of those messages by modifying the `processExternalized` function
(which is responsible for recording SCP messages on ledger close) to
also record any new or more recent messages heard about the previous
ledger.

Author: bboston7

src/herder/HerderImpl.cpp

@@ -307,10 +307,28 @@ HerderImpl::processExternalized(uint64 slotIndex, StellarValue const& value,
     // save the SCP messages in the database
     if (mApp.getConfig().MODE_STORES_HISTORY_MISC)
     {
+        ZoneNamedN(updateSCPHistoryZone, "update SCP history", true);
+
+        // NOTE: Consolidating the two saveSCPHistory calls into one transaction
+        // provides modest performance increases for sqlite and insignificant
+        // performance increases for postgres.
+        soci::transaction txScope(mApp.getDatabase().getSession());
+        if (slotIndex != 0)
+        {
+            // Save any new SCP messages received about the previous ledger.
+            // NOTE: This call uses an empty `QuorumTracker::QuorumMap` because
+            // there is no new quorum map for the previous ledger.
+            mApp.getHerderPersistence().saveSCPHistory(
+                static_cast<uint32>(slotIndex - 1),
+                getSCP().getExternalizingState(slotIndex - 1),
+                QuorumTracker::QuorumMap());
+        }
+        // Store SCP messages received about the current ledger being closed.
         mApp.getHerderPersistence().saveSCPHistory(
             static_cast<uint32>(slotIndex),
             getSCP().getExternalizingState(slotIndex),
             mPendingEnvelopes.getCurrentlyTrackedQuorum());
+        txScope.commit();
     }
 
     // reflect upgrades with the ones included in this SCP round

src/herder/HerderPersistenceImpl.cpp

@@ -62,6 +62,11 @@ HerderPersistenceImpl::saveSCPHistory(uint32_t seq,
             st.execute(true);
         }
     }
+
+    // Prepare multi-row insert into scphistory
+    std::vector<std::string> nodeIDs;
+    std::vector<uint32_t> seqs;
+    std::vector<std::string> envelopes;
     for (auto const& e : envs)
     {
         auto const& qHash =
@@ -76,21 +81,28 @@ HerderPersistenceImpl::saveSCPHistory(uint32_t seq,
         std::string envelopeEncoded;
         envelopeEncoded = decoder::encode_b64(envelopeBytes);
 
+        nodeIDs.push_back(nodeIDStrKey);
+        seqs.push_back(seq);
+        envelopes.push_back(envelopeEncoded);
+    }
+
+    if (!envs.empty())
+    {
+        // Perform multi-row insert into scphistory
         auto prepEnv =
             db.getPreparedStatement("INSERT INTO scphistory "
                                     "(nodeid, ledgerseq, envelope) VALUES "
                                     "(:n, :l, :e)");
-
         auto& st = prepEnv.statement();
-        st.exchange(soci::use(nodeIDStrKey));
-        st.exchange(soci::use(seq));
-        st.exchange(soci::use(envelopeEncoded));
+        st.exchange(soci::use(nodeIDs, "n"));
+        st.exchange(soci::use(seqs, "l"));
+        st.exchange(soci::use(envelopes, "e"));
         st.define_and_bind();
         {
             ZoneNamedN(insertSCPHistoryZone, "insert scphistory", true);
             st.execute(true);
         }
-        if (st.get_affected_rows() != 1)
+        if (st.get_affected_rows() != envs.size())
         {
             throw std::runtime_error("Could not update data in SQL");
         }

src/herder/test/HerderTests.cpp

@@ -5803,3 +5803,126 @@ TEST_CASE("exclude transactions by operation type", "[herder]")
                 TransactionQueue::AddResult::ADD_STATUS_PENDING);
     }
 }
+
+// Test that Herder updates the scphistory table with additional messages from
+// ledger `n-1` when closing ledger `n`
+TEST_CASE("SCP message capture from previous ledger", "[herder]")
+{
+    constexpr uint32_t version =
+        static_cast<uint32_t>(SOROBAN_PROTOCOL_VERSION);
+
+    // Initialize simulation
+    auto networkID = sha256(getTestConfig().NETWORK_PASSPHRASE);
+    auto simulation = std::make_shared<Simulation>(
+        Simulation::OVER_LOOPBACK, networkID, [](int i) {
+            auto cfg = getTestConfig(i, Config::TESTDB_ON_DISK_SQLITE);
+            cfg.TESTING_UPGRADE_LEDGER_PROTOCOL_VERSION = version;
+            return cfg;
+        });
+
+    // Create three validators: A, B, and C
+    auto validatorAKey = SecretKey::fromSeed(sha256("validator-A"));
+    auto validatorBKey = SecretKey::fromSeed(sha256("validator-B"));
+    auto validatorCKey = SecretKey::fromSeed(sha256("validator-C"));
+
+    // Put all validators in a quorum set of threshold 2
+    SCPQuorumSet qset;
+    qset.threshold = 2;
+    qset.validators.push_back(validatorAKey.getPublicKey());
+    qset.validators.push_back(validatorBKey.getPublicKey());
+    qset.validators.push_back(validatorCKey.getPublicKey());
+
+    // Connect validators A and B, but leave C disconnected
+    auto A = simulation->addNode(validatorAKey, qset);
+    auto B = simulation->addNode(validatorBKey, qset);
+    auto C = simulation->addNode(validatorCKey, qset);
+    simulation->addPendingConnection(validatorAKey.getPublicKey(),
+                                     validatorBKey.getPublicKey());
+    simulation->startAllNodes();
+
+    // Crank A and B until they're on ledger 2
+    simulation->crankUntil(
+        [&]() {
+            return A->getLedgerManager().getLastClosedLedgerNum() == 2 &&
+                   B->getLedgerManager().getLastClosedLedgerNum() == 2;
+        },
+        2 * Herder::EXP_LEDGER_TIMESPAN_SECONDS, false);
+
+    // Return the number of entries in a node's scphistory table for the given
+    // ledger number
+    auto getNumSCPHistoryEntries = [&](Application::pointer node,
+                                       uint32_t ledgerNum) {
+        auto& db = node->getDatabase();
+        auto prep = db.getPreparedStatement(
+            "SELECT COUNT(*) FROM scphistory WHERE ledgerseq = :l");
+        auto& st = prep.statement();
+        st.exchange(soci::use(ledgerNum));
+        uint32_t count;
+        st.exchange(soci::into(count));
+        st.define_and_bind();
+        st.execute(true);
+        return count;
+    };
+
+    // A and B should have 2 entries in their scphistory table for ledger 2. C
+    // should have 0.
+    REQUIRE(getNumSCPHistoryEntries(A, 2) == 2);
+    REQUIRE(getNumSCPHistoryEntries(B, 2) == 2);
+    REQUIRE(getNumSCPHistoryEntries(C, 2) == 0);
+
+    // Get messages from A and B
+    HerderImpl& herderA = dynamic_cast<HerderImpl&>(A->getHerder());
+    HerderImpl& herderB = dynamic_cast<HerderImpl&>(B->getHerder());
+    std::vector<SCPEnvelope> AEnvs = herderA.getSCP().getLatestMessagesSend(2);
+    std::vector<SCPEnvelope> BEnvs = herderB.getSCP().getLatestMessagesSend(2);
+
+    // Pass A and B's messages to C
+    for (auto const& env : AEnvs)
+    {
+        C->getHerder().recvSCPEnvelope(env);
+    }
+    for (auto const& env : BEnvs)
+    {
+        C->getHerder().recvSCPEnvelope(env);
+    }
+
+    // Crank C until it is on ledger 2
+    simulation->crankUntil(
+        [&]() { return C->getLedgerManager().getLastClosedLedgerNum() == 2; },
+        2 * Herder::EXP_LEDGER_TIMESPAN_SECONDS, false);
+
+    // Get messages from C
+    HerderImpl& herderC = dynamic_cast<HerderImpl&>(C->getHerder());
+    std::vector<SCPEnvelope> CEnvs = herderC.getSCP().getLatestMessagesSend(2);
+
+    // Pass C's messages to A and B
+    for (auto const& env : CEnvs)
+    {
+        A->getHerder().recvSCPEnvelope(env);
+        B->getHerder().recvSCPEnvelope(env);
+    }
+
+    // Crank A and B until they're on ledger 3
+    simulation->crankUntil(
+        [&]() {
+            return A->getLedgerManager().getLastClosedLedgerNum() == 3 &&
+                   B->getLedgerManager().getLastClosedLedgerNum() == 3;
+        },
+        2 * Herder::EXP_LEDGER_TIMESPAN_SECONDS, false);
+
+    // A and B should now have 3 entries in their scphistory table for ledger 2.
+    REQUIRE(getNumSCPHistoryEntries(A, 2) == 3);
+    REQUIRE(getNumSCPHistoryEntries(B, 2) == 3);
+
+    // Connect C to B and crank C to catch up with A and B
+    simulation->addConnection(validatorCKey.getPublicKey(),
+                              validatorBKey.getPublicKey());
+    simulation->crankUntil(
+        [&]() { return C->getLedgerManager().getLastClosedLedgerNum() == 3; },
+        2 * Herder::EXP_LEDGER_TIMESPAN_SECONDS, false);
+
+    // C should have 3 entries in its scphistory table for ledger 2. This check
+    // ensures that C does not double count messages from ledger 2 when closing
+    // ledger 3.
+    REQUIRE(getNumSCPHistoryEntries(C, 2) == 3);
+}