Recover old open ledger transactions to the queue:

* Recover to the open ledger once, then to the queue.
* If transaction fails to queue for any reason, drop it.
* New result codes for transactions that can not queue.
* Add minimum queue size
* RIPD-1530
fix XRPLF#2215

src/ripple/app/consensus/RCLConsensus.cpp

@@ -154,6 +154,7 @@ RCLConsensus::relay(RCLCxTx const& tx)
     // If we didn't relay this transaction recently, relay it to all peers
     if (app_.getHashRouter().shouldRelay(tx.id()))
     {
+      JLOG(j_.debug()) << "Relaying disputed tx " << tx.id();
         auto const slice = tx.tx_.slice();
         protocol::TMTransaction msg;
         msg.set_rawtransaction(slice.data(), slice.size());
@@ -163,6 +164,10 @@ RCLConsensus::relay(RCLCxTx const& tx)
         app_.overlay().foreach (send_always(
             std::make_shared<Message>(msg, protocol::mtTRANSACTION)));
     }
+    else
+      {
+        JLOG(j_.debug()) << "Not relaying disputedtx " << tx.id();
+      }
 }
 void
 RCLConsensus::propose(RCLCxPeerPos::Proposal const& proposal)
@@ -303,6 +308,7 @@ RCLConsensus::onClose(
     // Build SHAMap containing all transactions in our open ledger
     for (auto const& tx : initialLedger->txs)
     {
+      JLOG(j_.debug()) << "Adding open ledger TX " << tx.first->getTransactionID();
         Serializer s(2048);
         tx.first->add(s);
         initialSet->addItem(
@@ -474,7 +480,7 @@ RCLConsensus::doAccept(
                 {
                     JLOG(j_.debug())
                         << "Test applying disputed transaction that did"
-                        << " not get in";
+                        << " not get in " << it.second.tx().id();
 
                     SerialIter sit(it.second.tx().tx_.slice());
                     auto txn = std::make_shared<STTx const>(sit);

src/ripple/app/ledger/impl/OpenLedger.cpp

@@ -20,9 +20,13 @@
 #include <BeastConfig.h>
 #include <ripple/app/ledger/OpenLedger.h>
 #include <ripple/app/main/Application.h>
+#include <ripple/app/misc/HashRouter.h>
 #include <ripple/app/misc/TxQ.h>
 #include <ripple/app/tx/apply.h>
 #include <ripple/ledger/CachedView.h>
+#include <ripple/overlay/Message.h>
+#include <ripple/overlay/Overlay.h>
+#include <ripple/overlay/predicates.h>
 #include <ripple/protocol/Feature.h>
 #include <boost/range/adaptor/transformed.hpp>
 
@@ -117,6 +121,29 @@ OpenLedger::accept(Application& app, Rules const& rules,
     for (auto const& item : locals)
         app.getTxQ().apply(app, *next,
             item.second, flags, j_);
+
+    // If we didn't relay this transaction recently, relay it to all peers
+    for (auto const& txpair : next->txs)
+    {
+        auto const& tx = txpair.first;
+        auto const txId = tx->getTransactionID();
+        if (auto const toSkip = app.getHashRouter().shouldRelay(txId))
+        {
+            JLOG(j_.debug()) << "Relaying recovered tx " << txId;
+            protocol::TMTransaction msg;
+            Serializer s;
+
+            tx->add(s);
+            msg.set_rawtransaction(&s.getData().front(), s.getLength());
+            msg.set_status(protocol::tsNEW);
+            msg.set_receivetimestamp(
+                app.timeKeeper().now().time_since_epoch().count());
+            app.overlay().foreach(send_if_not(
+                std::make_shared<Message>(msg, protocol::mtTRANSACTION),
+                peer_in_set(*toSkip)));
+        }
+    }
+
     // Switch to the new open view
     std::lock_guard<
         std::mutex> lock2(current_mutex_);
@@ -143,9 +170,21 @@ OpenLedger::apply_one (Application& app, OpenView& view,
 {
     if (retry)
         flags = flags | tapRETRY;
-    auto const result = ripple::apply(
-        app, view, *tx, flags, j);
-    if (result.second)
+    auto const result = [&]
+    {
+        if (app.getHashRouter().shouldRecover(tx->getTransactionID()))
+            return ripple::apply(
+                app, view, *tx, flags, j);
+        else
+            // If the transaction can't get into the queue for any reason,
+            // drop it. If other nodes / validators have it, it'll be in
+            // their proposed set. If it's stuck on this node, then clean
+            // up the open ledger.
+            return app.getTxQ().apply(
+                app, view, tx, flags | tapPREFER_QUEUE, j);
+    }();
+    if (result.second ||
+            result.first == terQUEUED)
         return Result::success;
     if (isTefFailure (result.first) ||
         isTemMalformed (result.first) ||

src/ripple/app/main/Application.cpp

@@ -469,7 +469,8 @@ class ApplicationImp
         , mFeeTrack (std::make_unique<LoadFeeTrack>(logs_->journal("LoadManager")))
 
         , mHashRouter (std::make_unique<HashRouter>(
-            stopwatch(), HashRouter::getDefaultHoldTime ()))
+            stopwatch(), HashRouter::getDefaultHoldTime (),
+            HashRouter::getDefaultRecoverLimit ()))
 
         , mValidations (make_Validations (*this))
 

src/ripple/app/misc/HashRouter.cpp

@@ -107,4 +107,14 @@ HashRouter::shouldRelay (uint256 const& key)
     return s.releasePeerSet();
 }
 
+bool
+HashRouter::shouldRecover(uint256 const& key)
+{
+    std::lock_guard <std::mutex> lock(mutex_);
+
+    auto& s = emplace(key).first;
+
+    return s.shouldRecover(recoverLimit_);
+}
+
 } // ripple

src/ripple/app/misc/HashRouter.h

@@ -66,7 +66,6 @@ class HashRouter
         static char const* getCountedObjectName () { return "HashRouterEntry"; }
 
         Entry ()
-            : flags_ (0)
         {
         }
 
@@ -107,12 +106,33 @@ class HashRouter
             return true;
         }
 
+        /** Determines if this item should be recovered from the open ledger.
+
+            Counts the number of times the item has been recovered.
+            If it hits the limit, reset the counter and return false.
+            Else, increment the counter and return true.
+
+            @note The limit is signed while the counter is unsigned.
+                A negative limit will retry forever.
+        */
+        bool shouldRecover(std::int32_t limit)
+        {
+            if (recoveries_ == limit)
+            {
+                recoveries_ = 0;
+                return false;
+            }
+            ++recoveries_;
+            return true;
+        }
+
     private:
-        int flags_;
+        int flags_ = 0;
         std::set <PeerShortID> peers_;
         // This could be generalized to a map, if more
         // than one flag needs to expire independently.
         boost::optional<Stopwatch::time_point> relayed_;
+        std::uint16_t recoveries_ = 0;
     };
 
 public:
@@ -123,9 +143,16 @@ class HashRouter
         return 300s;
     }
 
-    HashRouter (Stopwatch& clock, std::chrono::seconds entryHoldTimeInSeconds)
+    static inline std::int32_t getDefaultRecoverLimit()
+    {
+        return 1;
+    }
+
+    HashRouter (Stopwatch& clock, std::chrono::seconds entryHoldTimeInSeconds,
+        std::int32_t recoverLimit)
         : suppressionMap_(clock)
         , holdTime_ (entryHoldTimeInSeconds)
+        , recoverLimit_ (recoverLimit)
     {
     }
 
@@ -164,6 +191,12 @@ class HashRouter
     */
     boost::optional<std::set<PeerShortID>> shouldRelay(uint256 const& key);
 
+    /** Determines whether the hashed item should be recovered
+
+        @return `bool` indicates whether the item should be relayed
+    */
+    bool shouldRecover(uint256 const& key);
+
 private:
     // pair.second indicates whether the entry was created
     std::pair<Entry&, bool> emplace (uint256 const&);
@@ -175,6 +208,8 @@ class HashRouter
         hardened_hash<strong_hash>> suppressionMap_;
 
     std::chrono::seconds const holdTime_;
+
+    std::int32_t const recoverLimit_;
 };
 
 } // ripple

src/ripple/app/misc/TxQ.h

@@ -55,6 +55,7 @@ class TxQ
     struct Setup
     {
         std::size_t ledgersInQueue = 20;
+        std::size_t queueSizeMin = 2000;
         std::uint32_t retrySequencePercent = 25;
         // TODO: eahennis. Can we remove the multi tx factor?
         std::int32_t multiTxnPercent = -90;

src/ripple/app/misc/impl/TxQ.cpp

@@ -339,7 +339,7 @@ TxQ::canBeHeld(STTx const& tx, OpenView const& view,
             promise to stick around for long enough that it has
             a realistic chance of getting into a ledger.
         */
-        auto lastValid = getLastLedgerSequence(tx);
+        auto const lastValid = getLastLedgerSequence(tx);
         canBeHeld = !lastValid || *lastValid >=
             view.info().seq + setup_.minimumLastLedgerBuffer;
     }
@@ -349,10 +349,15 @@ TxQ::canBeHeld(STTx const& tx, OpenView const& view,
             can queue. Mitigates the lost cost of relaying should
             an early one fail or get dropped.
         */
+        auto const tSeq = tx.getSequence();
         canBeHeld = accountIter == byAccount_.end() ||
             replacementIter ||
                 accountIter->second.getTxnCount() <
-                    setup_.maximumTxnPerAccount;
+                    setup_.maximumTxnPerAccount ||
+                        // Allow the transaction to get in front of the first
+                        // queued transaction. Allows recovery of open ledger
+                        // transactions, and stuck transactions.
+                        tSeq < accountIter->second.transactions.begin()->first;
     }
     return canBeHeld;
 }
@@ -512,8 +517,7 @@ TxQ::tryClearAccountQueue(Application& app, OpenView& view,
                 non-blockers?
             Yes: Remove the queued transaction. Continue to next
                 step.
-            No: Reject `txn` with `telINSUF_FEE_P` or
-                `telCAN_NOT_QUEUE`. Stop.
+            No: Reject `txn` with `telCAN_NOT_QUEUE_FEE`. Stop.
         No: Continue to next step.
     3. Does this tx have the expected sequence number for the
             account?
@@ -527,11 +531,11 @@ TxQ::tryClearAccountQueue(Application& app, OpenView& view,
                     than the previous tx?
                 No: Reject with `telINSUF_FEE_P`. Stop.
                 Yes: Are any of the prior sequence txs blockers?
-                    Yes: Reject with `telCAN_NOT_QUEUE`. Stop.
+                    Yes: Reject with `telCAN_NOT_QUEUE_BLOCKED`. Stop.
                     No: Are the fees in-flight of the other
                             queued txs >= than the account
                             balance or minimum account reserve?
-                        Yes: Reject with `telCAN_NOT_QUEUE`. Stop.
+                        Yes: Reject with `telCAN_NOT_QUEUE_BALANCE`. Stop.
                         No: Create a throwaway sandbox `View`. Modify
                             the account's sequence number to match
                             the tx (avoid `terPRE_SEQ`), and decrease
@@ -550,8 +554,7 @@ TxQ::tryClearAccountQueue(Application& app, OpenView& view,
                 it to `doApply()` and return that result.
             No: Continue to the next step.
     6. Can the tx be held in the queue? (See TxQ::canBeHeld).
-            No: Reject `txn` with `telINSUF_FEE_P` if this tx
-                has the current sequence, or `telCAN_NOT_QUEUE`
+            No: Reject `txn` with `telCAN_NOT_QUEUE_FULL`
                 if not. Stop.
             Yes: Continue to the next step.
     7. Is the queue full?
@@ -613,8 +616,15 @@ TxQ::apply(Application& app, OpenView& view,
     auto const baseFee = calculateBaseFee(app, view, *tx, j);
     auto const feeLevelPaid = getFeeLevelPaid(*tx,
         baseLevel, baseFee, setup_);
-    auto const requiredFeeLevel = FeeMetrics::scaleFeeLevel(
-        metricsSnapshot, view);
+    auto const requiredFeeLevel = [&]()
+    {
+        auto feeLevel = FeeMetrics::scaleFeeLevel(metricsSnapshot, view);
+        if ((flags & tapPREFER_QUEUE) && byFee_.size())
+        {
+            return std::max(feeLevel, byFee_.begin()->feeLevel);
+        }
+        return feeLevel;
+    }();
 
     auto accountIter = byAccount_.find(account);
     bool const accountExists = accountIter != byAccount_.end();
@@ -679,8 +689,7 @@ TxQ::apply(Application& app, OpenView& view,
                                 transactionID <<
                                 " in favor of normal queued " <<
                                 existingIter->second.txID;
-                            return{existingIter == txQAcct.transactions.begin() ?
-                                telINSUF_FEE_P : telCAN_NOT_QUEUE, false };
+                            return {telCAN_NOT_QUEUE_BLOCKS, false };
                         }
                     }
                 }
@@ -710,7 +719,7 @@ TxQ::apply(Application& app, OpenView& view,
                     transactionID <<
                     " in favor of queued " <<
                     existingIter->second.txID;
-                return{ telINSUF_FEE_P, false };
+                return{ telCAN_NOT_QUEUE_FEE, false };
             }
         }
     }
@@ -806,7 +815,7 @@ TxQ::apply(Application& app, OpenView& view,
                             transactionID <<
                             ". A blocker-type transaction " <<
                             "is in the queue.";
-                        return{ telCAN_NOT_QUEUE, false };
+                        return{ telCAN_NOT_QUEUE_BLOCKED, false };
                     }
                     multiTxn->fee +=
                         workingIter->second.consequences->fee;
@@ -824,7 +833,7 @@ TxQ::apply(Application& app, OpenView& view,
                     than the account's current balance, or the
                     minimum reserve. If it is, then there's a risk
                     that the fees won't get paid, so drop this
-                    transaction with a telCAN_NOT_QUEUE result.
+                    transaction with a telCAN_NOT_QUEUE_BALANCE result.
                     TODO: Decide whether to count the current txn fee
                         in this limit if it's the last transaction for
                         this account. Currently, it will not count,
@@ -866,7 +875,7 @@ TxQ::apply(Application& app, OpenView& view,
                         "Ignoring transaction " <<
                         transactionID <<
                         ". Total fees in flight too high.";
-                    return{ telCAN_NOT_QUEUE, false };
+                    return{ telCAN_NOT_QUEUE_BALANCE, false };
                 }
 
                 // Create the test view from the current view
@@ -970,8 +979,7 @@ TxQ::apply(Application& app, OpenView& view,
         JLOG(j_.trace()) << "Transaction " <<
             transactionID <<
             " can not be held";
-        return { feeLevelPaid >= requiredFeeLevel ?
-            telCAN_NOT_QUEUE : telINSUF_FEE_P, false };
+        return { telCAN_NOT_QUEUE, false };
     }
 
     // If the queue is full, decide whether to drop the current
@@ -986,7 +994,7 @@ TxQ::apply(Application& app, OpenView& view,
                 transactionID <<
                 " would kick a transaction from the same account (" <<
                 account << ") out of the queue.";
-            return { telCAN_NOT_QUEUE, false };
+            return { telCAN_NOT_QUEUE_FULL, false };
         }
         auto const& endAccount = byAccount_.at(lastRIter->account);
         auto endEffectiveFeeLevel = [&]()
@@ -1038,7 +1046,7 @@ TxQ::apply(Application& app, OpenView& view,
             JLOG(j_.warn()) << "Queue is full, and transaction " <<
                 transactionID <<
                 " fee is lower than end item's account average fee";
-            return { telINSUF_FEE_P, false };
+            return { telCAN_NOT_QUEUE_FULL, false };
         }
     }
 
@@ -1105,7 +1113,8 @@ TxQ::processClosedLedger(Application& app,
     auto ledgerSeq = view.info().seq;
 
     if (!timeLeap)
-        maxSize_ = snapshot.txnsExpected * setup_.ledgersInQueue;
+        maxSize_ = std::max (snapshot.txnsExpected * setup_.ledgersInQueue,
+             setup_.queueSizeMin);
 
     // Remove any queued candidates whose LastLedgerSequence has gone by.
     for(auto candidateIter = byFee_.begin(); candidateIter != byFee_.end(); )

src/ripple/ledger/ApplyView.h

@@ -36,6 +36,11 @@ enum ApplyFlags
     // Transaction can be retried, soft failures allowed
     tapRETRY            = 0x20,
 
+    // Transaction must pay more than both the open ledger
+    // fee and all transactions in the queue to get into the
+    // open ledger
+    tapPREFER_QUEUE     = 0x40,
+
     // Transaction came from a privileged source
     tapUNLIMITED        = 0x400,
 };