Fix loadgen seqno issus

Author: SirTyson

src/herder/HerderImpl.cpp

@@ -17,6 +17,7 @@
 #include "herder/RustQuorumCheckerAdaptor.h"
 #include "herder/TxSetFrame.h"
 #include "herder/TxSetUtils.h"
+#include "ledger/ImmutableLedgerView.h"
 #include "ledger/LedgerManager.h"
 #include "ledger/LedgerTxnImpl.h"
 #include "ledger/P23HotArchiveBug.h"
@@ -352,7 +353,8 @@ HerderImpl::processExternalized(uint64 slotIndex, StellarValue const& value,
             }
         }
 #ifdef BUILD_TESTS
-        mApp.getLoadGenerator().cleanupAccounts(txFramesList);
+        mApp.getLoadGenerator().cleanupAccounts(
+            static_cast<uint32_t>(slotIndex), txFramesList);
 #endif
     }
     mApp.getOverlayManager().notifyTxSetExternalized(value.txSetHash, txHashes);
@@ -1671,6 +1673,32 @@ HerderImpl::triggerNextLedger(uint32_t ledgerSeqToTrigger,
             classicTxs.push_back(txFrame);
         }
     }
+    // The mempool is fee-ordered and sequence-number-oblivious, so it can
+    // hand us several transactions from one source account (e.g. a chained
+    // pair). A tx set may only contain one tx per source account, so keep
+    // the lowest sequence number per account and let the others wait for a
+    // later ledger.
+    auto onePerSourceAccount = [](TxFrameList& txs) {
+        std::unordered_map<AccountID, size_t> firstBySource;
+        TxFrameList kept;
+        for (auto const& tx : txs)
+        {
+            auto [it, inserted] =
+                firstBySource.emplace(tx->getSourceID(), kept.size());
+            if (inserted)
+            {
+                kept.push_back(tx);
+            }
+            else if (tx->getSeqNum() < kept[it->second]->getSeqNum())
+            {
+                kept[it->second] = tx;
+            }
+        }
+        txs = std::move(kept);
+    };
+    onePerSourceAccount(classicTxs);
+    onePerSourceAccount(sorobanTxs);
+
     txPhases.emplace_back(std::move(classicTxs));
     if (supportsSoroban)
     {
@@ -1686,6 +1714,44 @@ HerderImpl::triggerNextLedger(uint32_t ledgerSeqToTrigger,
     CLOG_INFO(Herder, "Proposed TX set has {} transactions",
               proposedSet->sizeTxTotal());
 
+    // The mempool does no stateful validation, so it would keep handing us the
+    // transactions that just failed validation (stale sequence number, can't
+    // pay fee, expired, ...) on every nomination, crowding out valid ones.
+    // Drop them, except for transactions with a *future* sequence number:
+    // those are chained behind a pending transaction from the same account
+    // and become valid once it applies.
+    std::vector<Hash> invalidTxHashes;
+    if (!invalidTxPhases.empty())
+    {
+        CheckValidLedgerViewWrapper ledgerView(mApp);
+        for (auto const& phase : invalidTxPhases)
+        {
+            for (auto const& tx : phase)
+            {
+                auto acc = ledgerView.getAccount(tx->getSourceID());
+                if (acc &&
+                    tx->getSeqNum() > acc.current().data.account().seqNum + 1)
+                {
+                    continue;
+                }
+                CLOG_DEBUG(Herder,
+                           "Dropping invalid tx {} from mempool: seq {} "
+                           "(account seq {})",
+                           hexAbbrev(tx->getFullHash()), tx->getSeqNum(),
+                           acc ? acc.current().data.account().seqNum : -1);
+                invalidTxHashes.push_back(tx->getFullHash());
+            }
+        }
+    }
+    if (!invalidTxHashes.empty())
+    {
+        CLOG_DEBUG(Herder,
+                   "Removing {} transactions that failed tx set validation "
+                   "from the mempool",
+                   invalidTxHashes.size());
+        overlayMgr.removeTransactions(invalidTxHashes);
+    }
+
     if (!applicableProposedSet)
     {
         releaseAssert(!mApp.getConfig().FORCE_SCP);

src/herder/test/HerderTests.cpp

@@ -3322,6 +3322,18 @@ TEST_CASE("SCP checkpoint", "[catchup][herder]")
             simulation->getExpectedLedgerCloseTime(),
         false);
 
+    // An out of sync node should buffer every ledger from the checkpoint up to
+    // the main node's latest. The main node applies ledgers asynchronously
+    // after externalizing them and the out of sync node hears about them
+    // asynchronously, so this only holds at moments where the two line up.
+    auto hasBufferedCheckpointToLcl = [&](LedgerApplyManagerImpl const& lam) {
+        auto const& buffered = lam.getBufferedLedgers();
+        return !buffered.empty() &&
+               buffered.begin()->first == firstCheckpoint &&
+               buffered.crbegin()->first ==
+                   mainNode->getLedgerManager().getLastClosedLedgerNum();
+    };
+
     SECTION("GC old checkpoints")
     {
         HerderImpl& herder = static_cast<HerderImpl&>(mainNode->getHerder());
@@ -3362,14 +3374,13 @@ TEST_CASE("SCP checkpoint", "[catchup][herder]")
 
         // Crank until outOfSync node has received checkpoint ledger and started
         // catchup
-        simulation->crankUntil([&]() { return lam.isCatchupInitialized(); },
-                               2 * Herder::SEND_LATEST_CHECKPOINT_DELAY, false);
-
-        auto const& bufferedLedgers = lam.getBufferedLedgers();
-        REQUIRE(!bufferedLedgers.empty());
-        REQUIRE(bufferedLedgers.begin()->first == firstCheckpoint);
-        REQUIRE(bufferedLedgers.crbegin()->first ==
-                mainNode->getLedgerManager().getLastClosedLedgerNum());
+        simulation->crankUntil(
+            [&]() {
+                return lam.isCatchupInitialized() &&
+                       hasBufferedCheckpointToLcl(lam);
+            },
+            2 * Herder::SEND_LATEST_CHECKPOINT_DELAY, false);
+        REQUIRE(hasBufferedCheckpointToLcl(lam));
     }
 
     SECTION("Two out of sync nodes receive checkpoint")
@@ -3391,20 +3402,14 @@ TEST_CASE("SCP checkpoint", "[catchup][herder]")
         // catchup
         simulation->crankUntil(
             [&]() {
-                return cm1.isCatchupInitialized() && cm2.isCatchupInitialized();
+                return cm1.isCatchupInitialized() &&
+                       cm2.isCatchupInitialized() &&
+                       hasBufferedCheckpointToLcl(cm1) &&
+                       hasBufferedCheckpointToLcl(cm2);
             },
             2 * Herder::SEND_LATEST_CHECKPOINT_DELAY, false);
-
-        auto const& bufferedLedgers1 = cm1.getBufferedLedgers();
-        REQUIRE(!bufferedLedgers1.empty());
-        REQUIRE(bufferedLedgers1.begin()->first == firstCheckpoint);
-        REQUIRE(bufferedLedgers1.crbegin()->first ==
-                mainNode->getLedgerManager().getLastClosedLedgerNum());
-        auto const& bufferedLedgers2 = cm2.getBufferedLedgers();
-        REQUIRE(!bufferedLedgers2.empty());
-        REQUIRE(bufferedLedgers2.begin()->first == firstCheckpoint);
-        REQUIRE(bufferedLedgers2.crbegin()->first ==
-                mainNode->getLedgerManager().getLastClosedLedgerNum());
+        REQUIRE(hasBufferedCheckpointToLcl(cm1));
+        REQUIRE(hasBufferedCheckpointToLcl(cm2));
     }
 }
 
@@ -4023,7 +4028,16 @@ TEST_CASE("SCP message capture from previous ledger", "[herder]")
 {
     // Initialize simulation
     auto networkID = sha256(getTestConfig().NETWORK_PASSPHRASE);
-    auto simulation = std::make_shared<Simulation>(networkID);
+    // A and B keep closing ledgers in real time while C is fed slot 2 by
+    // hand. Use a longer close time so that C's slot-2 EXTERNALIZE reaches
+    // A and B before they close ledger 3.
+    auto confGen = [](int i) {
+        auto cfg = getTestConfig(i);
+        cfg.ARTIFICIALLY_ACCELERATE_TIME_FOR_TESTING = true;
+        cfg.ARTIFICIALLY_SET_CLOSE_TIME_FOR_TESTING = 5;
+        return cfg;
+    };
+    auto simulation = std::make_shared<Simulation>(networkID, confGen);
 
     // Create three validators: A, B, and C
     auto validatorAKey = SecretKey::fromSeed(sha256("validator-A"));

src/ledger/test/LedgerCloseMetaStreamTests.cpp

@@ -82,9 +82,8 @@ TEST_CASE("LedgerCloseMetaStream file descriptor - LIVE_NODE",
         Config cfg3 = getTestConfig(3);
         Config cfg4 = getTestConfig(4);
 
-        // Star topology around node1: addPendingConnection is a no-op with
-        // the Rust overlay, so wire the peers via KNOWN_PEERS instead
-        // (libp2p connections are bidirectional).
+        // Star topology around node1, wired via KNOWN_PEERS (libp2p
+        // connections are bidirectional).
         cfg1.KNOWN_PEERS = {
             fmt::format(FMT_STRING("127.0.0.1:{}"), cfg2.PEER_PORT),
             fmt::format(FMT_STRING("127.0.0.1:{}"), cfg3.PEER_PORT),

src/main/test/ApplicationUtilsTests.cpp

@@ -72,8 +72,7 @@ TEST_CASE("verify checkpoints command - wait condition", "[applicationutils]")
     cfg2.FORCE_SCP = false;
     cfg2.NODE_IS_VALIDATOR = false;
     cfg2.MODE_DOES_CATCHUP = false;
-    // addPendingConnection is a no-op with the Rust overlay: point the
-    // watcher at the validator via KNOWN_PEERS instead.
+    // Point the watcher at the validator via KNOWN_PEERS.
     cfg2.KNOWN_PEERS = {
         fmt::format(FMT_STRING("127.0.0.1:{}"), cfg1.PEER_PORT)};
 

src/overlay/OverlayIPC.cpp

@@ -14,7 +14,9 @@
 #include <filesystem>
 #include <signal.h>
 #include <sstream>
+#include <sys/syscall.h>
 #include <sys/wait.h>
+#include <thread>
 #include <unistd.h>
 #include <vector>
 
@@ -221,9 +223,26 @@ OverlayIPC::shutdown()
         pid_t result = waitpid(mOverlayPid, &status, WNOHANG);
         if (result == 0)
         {
-            // Still running, send SIGTERM
+            // Still running, send SIGTERM and give it a bounded amount of time
+            // to exit before falling back to SIGKILL, so that a wedged overlay
+            // process cannot hang core's shutdown.
             kill(mOverlayPid, SIGTERM);
-            waitpid(mOverlayPid, &status, 0);
+            auto const deadline =
+                std::chrono::steady_clock::now() + std::chrono::seconds(5);
+            while ((result = waitpid(mOverlayPid, &status, WNOHANG)) == 0 &&
+                   std::chrono::steady_clock::now() < deadline)
+            {
+                std::this_thread::sleep_for(std::chrono::milliseconds(20));
+            }
+            if (result == 0)
+            {
+                CLOG_WARNING(Overlay,
+                             "Overlay process {} did not exit after SIGTERM, "
+                             "sending SIGKILL",
+                             mOverlayPid);
+                kill(mOverlayPid, SIGKILL);
+                waitpid(mOverlayPid, &status, 0);
+            }
         }
         mOverlayPid = -1;
     }
@@ -239,6 +258,7 @@ OverlayIPC::spawnOverlay()
         return false;
     }
 
+    long const maxFd = sysconf(_SC_OPEN_MAX);
     pid_t pid = fork();
     if (pid < 0)
     {
@@ -248,7 +268,25 @@ OverlayIPC::spawnOverlay()
 
     if (pid == 0)
     {
-        // Child process - exec overlay binary
+        // Child process. Close every descriptor inherited from core except
+        // stdin/stdout/stderr: otherwise the overlay keeps core's listening
+        // sockets bound after core exits and holds the read ends of sibling
+        // overlays' IPC sockets, which can leave a shutting-down sibling
+        // blocked forever on a write nobody will read. Only async-signal-safe
+        // calls are allowed between fork() and exec().
+        bool closed = false;
+#if defined(__linux__) && defined(SYS_close_range)
+        closed = syscall(SYS_close_range, 3, ~0U, 0) == 0;
+#endif
+        if (!closed)
+        {
+            for (long fd = 3; fd < maxFd; ++fd)
+            {
+                close(static_cast<int>(fd));
+            }
+        }
+
+        // Exec overlay binary
         // Arguments: <binary> --listen <socket-path> --peer-port <port>
         std::string portStr = std::to_string(mPeerPort);
         execl(overlayBinaryPath->c_str(), overlayBinaryPath->c_str(),
@@ -500,6 +538,16 @@ OverlayIPC::notifyTxSetExternalized(Hash const& txSetHash,
     mChannel->send(msg);
 }
 
+void
+OverlayIPC::removeTransactions(std::vector<Hash> const& txHashes)
+{
+    if (txHashes.empty())
+    {
+        return;
+    }
+    notifyTxSetExternalized(Hash{}, txHashes);
+}
+
 std::vector<TransactionEnvelope>
 OverlayIPC::getTopTransactions(size_t count)
 {

src/overlay/OverlayIPC.h

@@ -103,6 +103,14 @@ class OverlayIPC
     void notifyTxSetExternalized(Hash const& txSetHash,
                                  std::vector<Hash> const& txHashes);
 
+    /**
+     * Remove transactions from the Rust mempool by hash, e.g. transactions
+     * that failed validation while building a tx set and would otherwise be
+     * handed back to Core on every nomination. Uses the TX_SET_EXTERNALIZED
+     * message with a zero tx set hash: the overlay only acts on the hashes.
+     */
+    void removeTransactions(std::vector<Hash> const& txHashes);
+
     /**
      * Request top N transactions by fee for nomination.
      *

src/overlay/RustOverlayManager.cpp

@@ -106,10 +106,18 @@ RustOverlayManager::effectiveKnownPeers() const
 void
 RustOverlayManager::addKnownPeerForTesting(std::string const& addr)
 {
-    releaseAssert(!mOverlayIPC || !mOverlayIPC->isConnected());
     releaseAssert(std::find(mExtraKnownPeers.begin(), mExtraKnownPeers.end(),
                             addr) == mExtraKnownPeers.end());
     mExtraKnownPeers.push_back(addr);
+
+    // If the overlay is already running, push the updated peer list so the
+    // Rust side dials the new peer.
+    if (mOverlayIPC && mOverlayIPC->isConnected())
+    {
+        auto const& cfg = mApp.getConfig();
+        mOverlayIPC->setPeerConfig(effectiveKnownPeers(), cfg.PREFERRED_PEERS,
+                                   cfg.PEER_PORT);
+    }
 }
 #endif
 
@@ -193,6 +201,15 @@ RustOverlayManager::notifyTxSetExternalized(Hash const& txSetHash,
     }
 }
 
+void
+RustOverlayManager::removeTransactions(std::vector<Hash> const& txHashes)
+{
+    if (mOverlayIPC && !mShuttingDown)
+    {
+        mOverlayIPC->removeTransactions(txHashes);
+    }
+}
+
 void
 RustOverlayManager::requestTxSet(Hash const& txSetHash, uint32_t slotIndex)
 {

src/overlay/RustOverlayManager.h

@@ -42,7 +42,9 @@ class RustOverlayManager
 #ifdef BUILD_TESTS
     // Advertise an additional peer address ("host:port") to the Rust overlay
     // on top of the config's KNOWN_PEERS. Used by Simulation to wire test
-    // topologies; must be called before start().
+    // topologies. May be called before start() (the peer is included in the
+    // initial peer config) or after it (the updated peer list is pushed to the
+    // Rust overlay immediately).
     void addKnownPeerForTesting(std::string const& addr);
 #endif
 
@@ -59,6 +61,10 @@ class RustOverlayManager
     void notifyTxSetExternalized(Hash const& txSetHash,
                                  std::vector<Hash> const& txHashes);
 
+    // Drop transactions from the Rust mempool (e.g. ones that failed
+    // validation while building a tx set).
+    void removeTransactions(std::vector<Hash> const& txHashes);
+
     // Request TX set from peers (via Rust overlay, async). slotIndex is the
     // slot the set is for, used to stamp the Rust-side cache entry.
     void requestTxSet(Hash const& txSetHash, uint32_t slotIndex);