fix(ledger): harden validated work scheduling

src/test/app/LedgerMaster_test.cpp

@@ -305,14 +305,17 @@ class LedgerMaster_test : public beast::unit_test::suite
         BEAST_EXPECT(queue.size() == 0);
         BEAST_EXPECT(!queue.drainScheduled());
 
-        // A completed or failed-to-post drain can be scheduled again.
+        // A completed or failed-to-post drain can be scheduled again. A
+        // failed post preserves the queued item, so the exact same identity
+        // must be able to re-arm scheduling without inserting a duplicate.
         result = queue.enqueue(a);
         BEAST_EXPECT(result.needsDrain);
         queue.cancelDrain();
         BEAST_EXPECT(!queue.drainScheduled());
-        result = queue.enqueue(b);
-        BEAST_EXPECT(result.inserted);
+        result = queue.enqueue(a);
+        BEAST_EXPECT(!result.inserted);
         BEAST_EXPECT(result.needsDrain);
+        BEAST_EXPECT(queue.size() == 1);
     }
 
     void

src/xrpld/app/ledger/LedgerMaster.h

@@ -336,9 +336,6 @@ private:
     void
     enqueueValidatedLedgerWork(detail::ValidatedLedgerWork work);
 
-    void
-    drainValidatedLedgerWork();
-
     // Returns true if work started.  Always called with m_mutex locked.
     // The passed lock is a reminder to callers.
     bool
@@ -370,7 +367,7 @@ private:
     LedgerHistory mLedgerHistory;
 
     // Local scheduling backpressure, not a protocol or Export capacity.
-    std::unique_ptr<detail::ValidatedLedgerWorkQueue> mValidatedLedgerWorkQueue;
+    std::shared_ptr<detail::ValidatedLedgerWorkQueue> mValidatedLedgerWorkQueue;
 
     CanonicalTXSet mHeldTransactions{uint256()};
 

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

@@ -116,7 +116,7 @@ LedgerMaster::LedgerMaster(
     , m_journal(journal)
     , mLedgerHistory(collector, app)
     , mValidatedLedgerWorkQueue(
-          std::make_unique<detail::ValidatedLedgerWorkQueue>(
+          std::make_shared<detail::ValidatedLedgerWorkQueue>(
               VALIDATED_LEDGER_WORK_QUEUE_CAPACITY))
     , standalone_(app_.config().standalone())
     , fetch_depth_(
@@ -351,21 +351,17 @@ LedgerMaster::enqueueValidatedLedgerWork(detail::ValidatedLedgerWork work)
     if (!result.needsDrain)
         return;
 
-    if (!app_.getJobQueue().addJob(jtADVANCE, "validatedLedgerWork", [this]() {
-            drainValidatedLedgerWork();
+    auto const queue = mValidatedLedgerWorkQueue;
+    if (!app_.getJobQueue().addJob(jtADVANCE, "validatedLedgerWork", [queue]() {
+            // Plateau A only schedules exact validation identities. The
+            // Export consumer attaches at the semantic flip. Capture the
+            // independently owned queue, never LedgerMaster: setup may
+            // fail before the normal JobQueue shutdown path runs.
+            queue->drain([](detail::ValidatedLedgerWork const&) noexcept {});
         }))
         mValidatedLedgerWorkQueue->cancelDrain();
 }
 
-void
-LedgerMaster::drainValidatedLedgerWork()
-{
-    // Plateau A only schedules exact validation identities. The Export
-    // consumer attaches at the semantic flip.
-    mValidatedLedgerWorkQueue->drain(
-        [](detail::ValidatedLedgerWork const&) noexcept {});
-}
-
 void
 LedgerMaster::setPubLedger(std::shared_ptr<Ledger const> const& l)
 {

src/xrpld/app/ledger/detail/ValidatedLedgerWorkQueue.h

@@ -75,7 +75,11 @@ public:
 
         if ((active_ && *active_ == work) ||
             std::find(queue_.begin(), queue_.end(), work) != queue_.end())
-            return {false, false, std::nullopt};
+        {
+            bool const needsDrain = !drainScheduled_;
+            drainScheduled_ = true;
+            return {false, needsDrain, std::nullopt};
+        }
 
         std::optional<ValidatedLedgerWork> evicted;
         if (queue_.size() == capacity_)