Improve transaction relay logic

src/test/app/HashRouter_test.cpp

@@ -26,12 +26,21 @@ namespace test {
 
 class HashRouter_test : public beast::unit_test::suite
 {
+    HashRouter::Setup
+    getSetup(std::chrono::seconds hold, std::chrono::seconds relay)
+    {
+        HashRouter::Setup setup;
+        setup.holdTime = hold;
+        setup.relayTime = relay;
+        return setup;
+    }
+
     void
     testNonExpiration()
     {
         using namespace std::chrono_literals;
         TestStopwatch stopwatch;
-        HashRouter router(stopwatch, 2s);
+        HashRouter router(getSetup(2s, 1s), stopwatch);
 
         uint256 const key1(1);
         uint256 const key2(2);
@@ -68,7 +77,7 @@ class HashRouter_test : public beast::unit_test::suite
     {
         using namespace std::chrono_literals;
         TestStopwatch stopwatch;
-        HashRouter router(stopwatch, 2s);
+        HashRouter router(getSetup(2s, 1s), stopwatch);
 
         uint256 const key1(1);
         uint256 const key2(2);
@@ -146,7 +155,7 @@ class HashRouter_test : public beast::unit_test::suite
         // Normal HashRouter
         using namespace std::chrono_literals;
         TestStopwatch stopwatch;
-        HashRouter router(stopwatch, 2s);
+        HashRouter router(getSetup(2s, 1s), stopwatch);
 
         uint256 const key1(1);
         uint256 const key2(2);
@@ -174,7 +183,7 @@ class HashRouter_test : public beast::unit_test::suite
     {
         using namespace std::chrono_literals;
         TestStopwatch stopwatch;
-        HashRouter router(stopwatch, 2s);
+        HashRouter router(getSetup(2s, 1s), stopwatch);
 
         uint256 const key1(1);
         BEAST_EXPECT(router.setFlags(key1, 10));
@@ -187,7 +196,7 @@ class HashRouter_test : public beast::unit_test::suite
     {
         using namespace std::chrono_literals;
         TestStopwatch stopwatch;
-        HashRouter router(stopwatch, 1s);
+        HashRouter router(getSetup(50s, 1s), stopwatch);
 
         uint256 const key1(1);
 
@@ -230,7 +239,7 @@ class HashRouter_test : public beast::unit_test::suite
     {
         using namespace std::chrono_literals;
         TestStopwatch stopwatch;
-        HashRouter router(stopwatch, 5s);
+        HashRouter router(getSetup(5s, 1s), stopwatch);
         uint256 const key(1);
         HashRouter::PeerShortID peer = 1;
         int flags;

src/xrpld/app/ledger/LocalTxs.h

@@ -33,6 +33,11 @@ namespace ripple {
 class LocalTxs
 {
 public:
+    // The number of ledgers to hold a transaction is essentially
+    // arbitrary. It should be sufficient to allow the transaction to
+    // get into a fully-validated ledger.
+    static constexpr int holdLedgers = 5;
+
     virtual ~LocalTxs() = default;
 
     // Add a new local transaction

src/xrpld/app/ledger/detail/LedgerMaster.cpp

@@ -458,25 +458,17 @@ LedgerMaster::storeLedger(std::shared_ptr<Ledger const> ledger)
 void
 LedgerMaster::applyHeldTransactions()
 {
-    std::lock_guard sl(m_mutex);
+    CanonicalTXSet const set = [this]() {
+        std::lock_guard sl(m_mutex);
+        // VFALCO NOTE The hash for an open ledger is undefined so we use
+        // something that is a reasonable substitute.
+        CanonicalTXSet set(app_.openLedger().current()->info().parentHash);
+        std::swap(mHeldTransactions, set);
+        return set;
+    }();
 
-    app_.openLedger().modify([&](OpenView& view, beast::Journal j) {
-        bool any = false;
-        for (auto const& it : mHeldTransactions)
-        {
-            ApplyFlags flags = tapNONE;
-            auto const result =
-                app_.getTxQ().apply(app_, view, it.second, flags, j);
-            any |= result.applied;
-        }
-        return any;
-    });
-
-    // VFALCO TODO recreate the CanonicalTxSet object instead of resetting
-    // it.
-    // VFALCO NOTE The hash for an open ledger is undefined so we use
-    // something that is a reasonable substitute.
-    mHeldTransactions.reset(app_.openLedger().current()->info().parentHash);
+    if (!set.empty())
+        app_.getOPs().processTransactionSet(set);
 }
 
 std::shared_ptr<STTx const>

src/xrpld/app/ledger/detail/LocalTxs.cpp

@@ -53,14 +53,9 @@ namespace ripple {
 class LocalTx
 {
 public:
-    // The number of ledgers to hold a transaction is essentially
-    // arbitrary. It should be sufficient to allow the transaction to
-    // get into a fully-validated ledger.
-    static int const holdLedgers = 5;
-
     LocalTx(LedgerIndex index, std::shared_ptr<STTx const> const& txn)
         : m_txn(txn)
-        , m_expire(index + holdLedgers)
+        , m_expire(index + LocalTxs::holdLedgers)
         , m_id(txn->getTransactionID())
         , m_account(txn->getAccountID(sfAccount))
         , m_seqProxy(txn->getSeqProxy())

src/xrpld/app/main/Application.cpp

@@ -444,8 +444,8 @@ class ApplicationImp : public Application, public BasicApp
               std::make_unique<LoadFeeTrack>(logs_->journal("LoadManager")))
 
         , hashRouter_(std::make_unique<HashRouter>(
-              stopwatch(),
-              HashRouter::getDefaultHoldTime()))
+              setup_HashRouter(*config_),
+              stopwatch()))
 
         , mValidations(
               ValidationParms(),

src/xrpld/app/main/Tuning.h

@@ -25,6 +25,8 @@ namespace ripple {
 constexpr std::size_t fullBelowTargetSize = 524288;
 constexpr std::chrono::seconds fullBelowExpiration = std::chrono::minutes{10};
 
+constexpr std::size_t maxPoppedTransactions = 10;
+
 }  // namespace ripple
 
 #endif

src/xrpld/app/misc/CanonicalTXSet.cpp

@@ -66,18 +66,22 @@ CanonicalTXSet::popAcctTransaction(std::shared_ptr<STTx const> const& tx)
     //  1. Prioritize transactions with Sequences over transactions with
     //     Tickets.
     //
-    //  2. Don't worry about consecutive Sequence numbers.  Creating Tickets
-    //     can introduce a discontinuity in Sequence numbers.
+    //  2. For transactions not using Tickets, look for consecutive Sequence
+    //     numbers. For transactions using Tickets, don't worry about
+    //     consecutive Sequence numbers. Tickets can process out of order.
     //
     //  3. After handling all transactions with Sequences, return Tickets
     //     with the lowest Ticket ID first.
     std::shared_ptr<STTx const> result;
     uint256 const effectiveAccount{accountKey(tx->getAccountID(sfAccount))};
 
-    Key const after(effectiveAccount, tx->getSeqProxy(), beast::zero);
+    auto const seqProxy = tx->getSeqProxy();
+    Key const after(effectiveAccount, seqProxy, beast::zero);
     auto const itrNext{map_.lower_bound(after)};
     if (itrNext != map_.end() &&
-        itrNext->first.getAccount() == effectiveAccount)
+        itrNext->first.getAccount() == effectiveAccount &&
+        (!itrNext->second->getSeqProxy().isSeq() ||
+         itrNext->second->getSeqProxy().value() == seqProxy.value() + 1))
     {
         result = std::move(itrNext->second);
         map_.erase(itrNext);

src/xrpld/app/misc/HashRouter.cpp

@@ -19,6 +19,8 @@
 
 #include <xrpld/app/misc/HashRouter.h>
 
+#include <xrpld/core/Config.h>
+
 namespace ripple {
 
 auto
@@ -33,7 +35,7 @@
     }
 
     // See if any supressions need to be expired
-    expire(suppressionMap_, holdTime_);
+    expire(suppressionMap_, setup_.holdTime);
 
     return std::make_pair(
         std::ref(suppressionMap_.emplace(key, Entry()).first->second), true);
@@ -122,10 +124,45 @@
 
     auto& s = emplace(key).first;
 
-    if (!s.shouldRelay(suppressionMap_.clock().now(), holdTime_))
+    if (!s.shouldRelay(suppressionMap_.clock().now(), setup_.relayTime))
         return {};
 
     return s.releasePeerSet();
 }
 
+HashRouter::Setup
+setup_HashRouter(Config const& config)
+{
+    using namespace std::chrono;
+
+    HashRouter::Setup setup;
+    auto const& section = config.section("hashrouter");
+
+    std::int32_t tmp;
+
+    if (set(tmp, "hold_time", section))
+    {
+        if (tmp < 8)
+            Throw<std::runtime_error>(
+                "HashRouter hold time must be at least 12 seconds (the "
+                "approximate validation time for three ledgers).");
+        setup.holdTime = seconds(tmp);
+    }
+    if (set(tmp, "relay_time", section))
+    {
+        if (tmp < 8)
+            Throw<std::runtime_error>(
+                "HashRouter relay time must be at least 8 seconds (the "
+                "approximate validation time for two ledgers).");
+        setup.relayTime = seconds(tmp);
+    }
+    if (setup.relayTime > setup.holdTime)
+    {
+        Throw<std::runtime_error>(
+            "HashRouter relay time must be less than or equal to hold time");
+    }
+
+    return setup;
+}
+
 }  // namespace ripple

src/xrpld/app/misc/HashRouter.h

@@ -33,6 +33,7 @@ namespace ripple {
 // TODO convert these macros to int constants or an enum
 #define SF_BAD 0x02  // Temporarily bad
 #define SF_SAVED 0x04
+#define SF_HELD 0x08  // Held by LedgerMaster after potential processing failure
 #define SF_TRUSTED 0x10  // comes from trusted source
 
 // Private flags, used internally in apply.cpp.
@@ -44,6 +45,8 @@ namespace ripple {
 #define SF_PRIVATE5 0x1000
 #define SF_PRIVATE6 0x2000
 
+class Config;
+
 /** Routing table for objects identified by hash.
 
     This table keeps track of which hashes have been received by which peers.
@@ -56,6 +59,27 @@ class HashRouter
     // The type here *MUST* match the type of Peer::id_t
     using PeerShortID = std::uint32_t;
 
+    /** Structure used to customize @ref HashRouter behavior.
+     *
+     * Even though these items are configurable, don't change them unless there
+     * is a good reason, and network-wide coordination to do it.
+     */
+    struct Setup
+    {
+        /// Default constructor
+        explicit Setup() = default;
+
+        using seconds = std::chrono::seconds;
+
+        /** Expiration time for a hash entry
+         */
+        seconds holdTime{300};
+
+        /** Amount of time required before a relayed item will be relayed again.
+         */
+        seconds relayTime{30};
+    };
+
 private:
     /** An entry in the routing table.
      */
@@ -135,16 +159,8 @@ class HashRouter
     };
 
 public:
-    static inline std::chrono::seconds
-    getDefaultHoldTime()
-    {
-        using namespace std::chrono;
-
-        return 300s;
-    }
-
-    HashRouter(Stopwatch& clock, std::chrono::seconds entryHoldTimeInSeconds)
-        : suppressionMap_(clock), holdTime_(entryHoldTimeInSeconds)
+    HashRouter(Setup const& setup, Stopwatch& clock)
+        : setup_(setup), suppressionMap_(clock)
     {
     }
 
@@ -195,11 +211,11 @@ class HashRouter
         Effects:
 
             If the item should be relayed, this function will not
-            return `true` again until the hold time has expired.
+            return a seated optional again until the relay time has expired.
             The internal set of peers will also be reset.
 
         @return A `std::optional` set of peers which do not need to be
-            relayed to. If the result is uninitialized, the item should
+            relayed to. If the result is unseated, the item should
             _not_ be relayed.
     */
     std::optional<std::set<PeerShortID>>
@@ -212,17 +228,21 @@ class HashRouter
 
     std::mutex mutable mutex_;
 
+    // Configurable parameters
+    Setup const setup_;
+
     // Stores all suppressed hashes and their expiration time
     beast::aged_unordered_map<
         uint256,
         Entry,
         Stopwatch::clock_type,
         hardened_hash<strong_hash>>
         suppressionMap_;
-
-    std::chrono::seconds const holdTime_;
 };
 
+HashRouter::Setup
+setup_HashRouter(Config const&);
+
 }  // namespace ripple
 
 #endif