Allow only 1 job queue slot for each validation ledger check

* refactor filtering of validations to specifically avoid
 concurrent checkAccept() calls for the same validation ledger hash.
* Log when duplicate concurrent validation requests are filtered.
* RAII for containers that track concurrent validation requests.

src/ripple/app/consensus/RCLValidations.cpp

@@ -161,7 +161,9 @@ void
 handleNewValidation(
     Application& app,
     std::shared_ptr<STValidation> const& val,
-    std::string const& source)
+    std::string const& source,
+    BypassAccept const bypassAccept,
+    std::optional<beast::Journal> j)
 {
     auto const& signingKey = val->getSignerPublic();
     auto const& hash = val->getLedgerHash();
@@ -186,7 +188,21 @@ handleNewValidation(
     if (outcome == ValStatus::current)
     {
         if (val->isTrusted())
-            app.getLedgerMaster().checkAccept(hash, seq);
+        {
+            if (bypassAccept == BypassAccept::yes)
+            {
+                assert(j.has_value());
+                if (j.has_value())
+                {
+                    JLOG(j->trace()) << "Bypassing checkAccept for validation "
+                                     << val->getLedgerHash();
+                }
+            }
+            else
+            {
+                app.getLedgerMaster().checkAccept(hash, seq);
+            }
+        }
         return;
     }
 

src/ripple/app/consensus/RCLValidations.h

@@ -25,12 +25,16 @@
 #include <ripple/protocol/Protocol.h>
 #include <ripple/protocol/RippleLedgerHash.h>
 #include <ripple/protocol/STValidation.h>
+#include <optional>
+#include <set>
 #include <vector>
 
 namespace ripple {
 
 class Application;
 
+enum class BypassAccept : bool { no = false, yes };
+
 /** Wrapper over STValidation for generic Validation code
 
     Wraps an STValidation for compatibility with the generic validation code.
@@ -248,7 +252,9 @@ void
 handleNewValidation(
     Application& app,
     std::shared_ptr<STValidation> const& val,
-    std::string const& source);
+    std::string const& source,
+    BypassAccept const bypassAccept = BypassAccept::no,
+    std::optional<beast::Journal> j = std::nullopt);
 
 }  // namespace ripple
 

src/ripple/app/misc/NetworkOPs.cpp

@@ -72,8 +72,10 @@
 #include <boost/asio/steady_timer.hpp>
 
 #include <algorithm>
+#include <exception>
 #include <mutex>
 #include <optional>
+#include <set>
 #include <string>
 #include <tuple>
 #include <unordered_map>
@@ -781,6 +783,9 @@ class NetworkOPsImp final : public NetworkOPs
 
     StateAccounting accounting_{};
 
+    std::set<uint256> pendingValidations_;
+    std::mutex validationsMutex_;
+
 private:
     struct Stats
     {
@@ -2308,7 +2313,35 @@ NetworkOPsImp::recvValidation(
     JLOG(m_journal.trace())
         << "recvValidation " << val->getLedgerHash() << " from " << source;
 
-    handleNewValidation(app_, val, source);
+    std::unique_lock lock(validationsMutex_);
+    BypassAccept bypassAccept = BypassAccept::no;
+    try
+    {
+        if (pendingValidations_.contains(val->getLedgerHash()))
+            bypassAccept = BypassAccept::yes;
+        else
+            pendingValidations_.insert(val->getLedgerHash());
+        lock.unlock();
+        handleNewValidation(app_, val, source, bypassAccept, m_journal);
+    }
+    catch (std::exception const& e)
+    {
+        JLOG(m_journal.warn())
+            << "Exception thrown for handling new validation "
+            << val->getLedgerHash() << ": " << e.what();
+    }
+    catch (...)
+    {
+        JLOG(m_journal.warn())
+            << "Unknown exception thrown for handling new validation "
+            << val->getLedgerHash();
+    }
+    if (bypassAccept == BypassAccept::no)
+    {
+        lock.lock();
+        pendingValidations_.erase(val->getLedgerHash());
+        lock.unlock();
+    }
 
     pubValidation(val);