Control memory growth from slow writes

* Don't allow a write batch to grow without bound
* Don't fetch history if write load is high

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

@@ -59,6 +59,9 @@ using namespace std::chrono_literals;
 // Don't acquire history if ledger is too old
 auto constexpr MAX_LEDGER_AGE_ACQUIRE = 1min;
 
+// Don't acquire history if write load is too high
+auto constexpr MAX_WRITE_LOAD_ACQUIRE = 8192;
+
 LedgerMaster::LedgerMaster (Application& app, Stopwatch& stopwatch,
     Stoppable& parent,
     beast::insight::Collector::ptr const& collector, beast::Journal journal)
@@ -1680,7 +1683,8 @@ void LedgerMaster::doAdvance (ScopedLockType& sl)
             if (!standalone_ && !app_.getFeeTrack().isLoadedLocal() &&
                 (app_.getJobQueue().getJobCount(jtPUBOLDLEDGER) < 10) &&
                 (mValidLedgerSeq == mPubLedgerSeq) &&
-                (getValidatedLedgerAge() < MAX_LEDGER_AGE_ACQUIRE))
+                (getValidatedLedgerAge() < MAX_LEDGER_AGE_ACQUIRE) &&
+                (app_.getNodeStore().getWriteLoad() < MAX_WRITE_LOAD_ACQUIRE))
             {
                 // We are in sync, so can acquire
                 InboundLedger::Reason reason = InboundLedger::Reason::HISTORY;

src/ripple/nodestore/Types.h

@@ -32,7 +32,13 @@ enum
     // This is only used to pre-allocate the array for
     // batch objects and does not affect the amount written.
     //
-    batchWritePreallocationSize = 256
+    batchWritePreallocationSize = 256,
+
+    // This sets a limit on the maximum number of writes
+    // in a batch. Actual usage can be twice this since
+    // we have a new batch growing as we write the old.
+    //
+    batchWriteLimitSize = 65536
 };
 
 /** Return codes from Backend operations. */

src/ripple/nodestore/impl/BatchWriter.cpp

@@ -39,7 +39,12 @@ BatchWriter::~BatchWriter ()
 void
 BatchWriter::store (std::shared_ptr<NodeObject> const& object)
 {
-    std::lock_guard<decltype(mWriteMutex)> sl (mWriteMutex);
+    std::unique_lock<decltype(mWriteMutex)> sl (mWriteMutex);
+
+    // If the batch has reached its limit, we wait
+    // until the batch writer is finished
+    while (mWriteSet.size() >= batchWriteLimitSize)
+        mWriteCondition.wait (sl);
 
     mWriteSet.push_back (object);