refactor(ledger): queue validated-ledger work

src/test/app/LedgerMaster_test.cpp

@@ -21,6 +21,7 @@
 #include <test/jtx/Env.h>
 #include <xrpld/app/ledger/LedgerMaster.h>
 #include <xrpld/app/ledger/detail/PublishGap.h>
+#include <xrpld/app/ledger/detail/ValidatedLedgerWorkQueue.h>
 #include <xrpl/basics/RangeSet.h>
 #include <xrpl/protocol/jss.h>
 
@@ -253,6 +254,67 @@ class LedgerMaster_test : public beast::unit_test::suite
         BEAST_EXPECT(!lm.isPinned(seq));
     }
 
+    void
+    testValidatedLedgerWorkQueue()
+    {
+        testcase("validated ledger work queue");
+
+        using detail::ValidatedLedgerWork;
+        using detail::ValidatedLedgerWorkQueue;
+
+        ValidatedLedgerWorkQueue queue{2};
+        ValidatedLedgerWork const a{10, uint256{1}};
+        ValidatedLedgerWork const fork{10, uint256{2}};
+        ValidatedLedgerWork const b{11, uint256{3}};
+
+        auto result = queue.enqueue(a);
+        BEAST_EXPECT(result.inserted);
+        BEAST_EXPECT(result.needsDrain);
+        BEAST_EXPECT(!result.evicted);
+
+        result = queue.enqueue(a);
+        BEAST_EXPECT(!result.inserted);
+        BEAST_EXPECT(!result.needsDrain);
+        BEAST_EXPECT(queue.size() == 1);
+
+        // Sequence alone is not identity: a different hash remains distinct.
+        result = queue.enqueue(fork);
+        BEAST_EXPECT(result.inserted);
+        BEAST_EXPECT(!result.needsDrain);
+        BEAST_EXPECT(queue.size() == 2);
+
+        // Saturation drops the oldest pending event and retains newer work.
+        result = queue.enqueue(b);
+        BEAST_EXPECT(result.inserted);
+        BEAST_EXPECT(!result.needsDrain);
+        BEAST_EXPECT(result.evicted && *result.evicted == a);
+
+        std::vector<ValidatedLedgerWork> drained;
+        auto const count =
+            queue.drain([&](ValidatedLedgerWork const& work) noexcept {
+                // An exact duplicate cannot be queued while it is active.
+                auto const duplicate = queue.enqueue(work);
+                BEAST_EXPECT(!duplicate.inserted);
+                drained.push_back(work);
+            });
+
+        BEAST_EXPECT(count == 2);
+        BEAST_EXPECT(drained.size() == 2);
+        BEAST_EXPECT(drained[0] == fork);
+        BEAST_EXPECT(drained[1] == b);
+        BEAST_EXPECT(queue.size() == 0);
+        BEAST_EXPECT(!queue.drainScheduled());
+
+        // A completed or failed-to-post drain can be scheduled again.
+        result = queue.enqueue(a);
+        BEAST_EXPECT(result.needsDrain);
+        queue.cancelDrain();
+        BEAST_EXPECT(!queue.drainScheduled());
+        result = queue.enqueue(b);
+        BEAST_EXPECT(result.inserted);
+        BEAST_EXPECT(result.needsDrain);
+    }
+
     void
     testSetPinnedRangesImmediateMerge()
     {
@@ -334,6 +396,7 @@ public:
         using namespace test::jtx;
         FeatureBitset const all{supported_amendments() - featureXahauGenesis};
         testCanSkipPinnedGap();
+        testValidatedLedgerWorkQueue();
         testWithFeats(all);
         testPinUnpinSymmetry();
         testSetPinnedRangesImmediateMerge();

src/xrpld/app/ledger/LedgerMaster.h

@@ -37,14 +37,18 @@
 #include <xrpl/protocol/RippleLedgerHash.h>
 #include <xrpl/protocol/STValidation.h>
 #include <xrpl/protocol/messages.h>
-#include <optional>
-
+#include <memory>
 #include <mutex>
+#include <optional>
 
 namespace ripple {
 
 class Peer;
 class Transaction;
+namespace detail {
+struct ValidatedLedgerWork;
+class ValidatedLedgerWorkQueue;
+}  // namespace detail
 
 // Tracks the current ledger and any ledgers in the process of closing
 // Tracks ledger history
@@ -58,7 +62,7 @@ public:
         beast::insight::Collector::ptr const& collector,
         beast::Journal journal);
 
-    virtual ~LedgerMaster() = default;
+    virtual ~LedgerMaster();
 
     LedgerIndex
     getCurrentLedgerIndex();
@@ -329,6 +333,12 @@ private:
     void
     updatePaths();
 
+    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
@@ -359,6 +369,9 @@ private:
 
     LedgerHistory mLedgerHistory;
 
+    // Local scheduling backpressure, not a protocol or Export capacity.
+    std::unique_ptr<detail::ValidatedLedgerWorkQueue> mValidatedLedgerWorkQueue;
+
     CanonicalTXSet mHeldTransactions{uint256()};
 
     // A set of transactions to replay during the next close

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

@@ -25,6 +25,7 @@
 #include <xrpld/app/ledger/OrderBookDB.h>
 #include <xrpld/app/ledger/PendingSaves.h>
 #include <xrpld/app/ledger/detail/PublishGap.h>
+#include <xrpld/app/ledger/detail/ValidatedLedgerWorkQueue.h>
 #include <xrpld/app/main/Application.h>
 #include <xrpld/app/misc/AmendmentTable.h>
 #include <xrpld/app/misc/HashRouter.h>
@@ -72,6 +73,11 @@ static constexpr std::chrono::minutes MAX_LEDGER_AGE_ACQUIRE{1};
 // Don't acquire history if write load is too high
 static constexpr int MAX_WRITE_LOAD_ACQUIRE{8192};
 
+// The newest validation event is retained when this local scheduling queue is
+// saturated. A future consumer can rescan the durable pending index from that
+// point, so this is not a protocol limit.
+static constexpr std::size_t VALIDATED_LEDGER_WORK_QUEUE_CAPACITY{256};
+
 // Helper function for LedgerMaster::doAdvance()
 // Return true if candidateLedger should be fetched from the network.
 static bool
@@ -109,6 +115,9 @@ LedgerMaster::LedgerMaster(
     : app_(app)
     , m_journal(journal)
     , mLedgerHistory(collector, app)
+    , mValidatedLedgerWorkQueue(
+          std::make_unique<detail::ValidatedLedgerWorkQueue>(
+              VALIDATED_LEDGER_WORK_QUEUE_CAPACITY))
     , standalone_(app_.config().standalone())
     , fetch_depth_(
           app_.getSHAMapStore().clampFetchDepth(app_.config().FETCH_DEPTH))
@@ -124,6 +133,8 @@ LedgerMaster::LedgerMaster(
 {
 }
 
+LedgerMaster::~LedgerMaster() = default;
+
 LedgerIndex
 LedgerMaster::getCurrentLedgerIndex()
 {
@@ -322,6 +333,37 @@ LedgerMaster::setValidLedger(std::shared_ptr<Ledger const> const& l)
                 app_.getOPs().clearAmendmentWarned();
         }
     }
+
+    enqueueValidatedLedgerWork({l->info().seq, l->info().hash});
+}
+
+void
+LedgerMaster::enqueueValidatedLedgerWork(detail::ValidatedLedgerWork work)
+{
+    auto const result = mValidatedLedgerWorkQueue->enqueue(std::move(work));
+    if (result.evicted)
+    {
+        JLOG(m_journal.warn())
+            << "Validated-ledger work queue evicted seq=" << result.evicted->seq
+            << " hash=" << result.evicted->hash;
+    }
+
+    if (!result.needsDrain)
+        return;
+
+    if (!app_.getJobQueue().addJob(jtADVANCE, "validatedLedgerWork", [this]() {
+            drainValidatedLedgerWork();
+        }))
+        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

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

@@ -0,0 +1,164 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2026 XRPLF
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#ifndef RIPPLE_APP_LEDGER_DETAIL_VALIDATEDLEDGERWORKQUEUE_H_INCLUDED
+#define RIPPLE_APP_LEDGER_DETAIL_VALIDATEDLEDGERWORKQUEUE_H_INCLUDED
+
+#include <xrpl/basics/contract.h>
+#include <xrpl/protocol/Protocol.h>
+#include <xrpl/protocol/RippleLedgerHash.h>
+
+#include <algorithm>
+#include <cstddef>
+#include <deque>
+#include <functional>
+#include <mutex>
+#include <optional>
+#include <type_traits>
+#include <utility>
+
+namespace ripple::detail {
+
+struct ValidatedLedgerWork
+{
+    LedgerIndex seq;
+    LedgerHash hash;
+
+    friend bool
+    operator==(ValidatedLedgerWork const&, ValidatedLedgerWork const&) =
+        default;
+};
+
+/** A bounded, exact-key queue for work unlocked by ledger validation.
+
+    The queue owns only scheduling identity. Consumers are responsible for
+    looking up any ledger or feature-specific state after the item is drained.
+*/
+class ValidatedLedgerWorkQueue
+{
+public:
+    struct EnqueueResult
+    {
+        bool inserted;
+        bool needsDrain;
+        std::optional<ValidatedLedgerWork> evicted;
+    };
+
+    explicit ValidatedLedgerWorkQueue(std::size_t capacity)
+        : capacity_(capacity)
+    {
+        XRPL_ASSERT(
+            capacity_ > 0,
+            "ripple::detail::ValidatedLedgerWorkQueue : positive capacity");
+    }
+
+    EnqueueResult
+    enqueue(ValidatedLedgerWork work)
+    {
+        std::lock_guard lock(mutex_);
+
+        if ((active_ && *active_ == work) ||
+            std::find(queue_.begin(), queue_.end(), work) != queue_.end())
+            return {false, false, std::nullopt};
+
+        std::optional<ValidatedLedgerWork> evicted;
+        if (queue_.size() == capacity_)
+        {
+            evicted = std::move(queue_.front());
+            queue_.pop_front();
+        }
+
+        queue_.push_back(std::move(work));
+        bool const needsDrain = !drainScheduled_;
+        drainScheduled_ = true;
+        return {true, needsDrain, std::move(evicted)};
+    }
+
+    /** Drain queued items in FIFO order.
+
+        The consumer runs without the queue mutex. It must not throw; future
+        consumers should catch feature-specific failures within the callback.
+    */
+    template <class Consumer>
+    std::size_t
+    drain(Consumer&& consumer)
+    {
+        static_assert(
+            std::is_nothrow_invocable_v<Consumer&, ValidatedLedgerWork const&>);
+
+        std::size_t drained = 0;
+        for (;;)
+        {
+            std::optional<ValidatedLedgerWork> work;
+            {
+                std::lock_guard lock(mutex_);
+                active_.reset();
+                if (queue_.empty())
+                {
+                    drainScheduled_ = false;
+                    return drained;
+                }
+
+                work = std::move(queue_.front());
+                queue_.pop_front();
+                active_ = *work;
+            }
+
+            std::invoke(consumer, *work);
+            ++drained;
+        }
+    }
+
+    /** Re-arm scheduling if posting the drain job failed. */
+    void
+    cancelDrain()
+    {
+        std::lock_guard lock(mutex_);
+        XRPL_ASSERT(
+            !active_,
+            "ripple::detail::ValidatedLedgerWorkQueue::cancelDrain : no "
+            "active consumer");
+        drainScheduled_ = false;
+    }
+
+    std::size_t
+    size() const
+    {
+        std::lock_guard lock(mutex_);
+        return queue_.size();
+    }
+
+    bool
+    drainScheduled() const
+    {
+        std::lock_guard lock(mutex_);
+        return drainScheduled_;
+    }
+
+private:
+    std::size_t const capacity_;
+    mutable std::mutex mutex_;
+    std::deque<ValidatedLedgerWork> queue_;
+    std::optional<ValidatedLedgerWork> active_;
+    bool drainScheduled_{false};
+};
+
+}  // namespace ripple::detail
+
+#endif