From 0536e2e9eb2fccaeca5734e040a84fc6fb2b105a Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Mon, 13 Jul 2026 18:39:44 +0700 Subject: [PATCH] refactor(ledger): queue validated-ledger work --- src/test/app/LedgerMaster_test.cpp | 63 +++++++ src/xrpld/app/ledger/LedgerMaster.h | 19 +- src/xrpld/app/ledger/detail/LedgerMaster.cpp | 42 +++++ .../ledger/detail/ValidatedLedgerWorkQueue.h | 164 ++++++++++++++++++ 4 files changed, 285 insertions(+), 3 deletions(-) create mode 100644 src/xrpld/app/ledger/detail/ValidatedLedgerWorkQueue.h diff --git a/src/test/app/LedgerMaster_test.cpp b/src/test/app/LedgerMaster_test.cpp index eaa2100c0..918bf1b79 100644 --- a/src/test/app/LedgerMaster_test.cpp +++ b/src/test/app/LedgerMaster_test.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include @@ -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 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(); diff --git a/src/xrpld/app/ledger/LedgerMaster.h b/src/xrpld/app/ledger/LedgerMaster.h index c4ac1e510..1c9ab0e9f 100644 --- a/src/xrpld/app/ledger/LedgerMaster.h +++ b/src/xrpld/app/ledger/LedgerMaster.h @@ -37,14 +37,18 @@ #include #include #include -#include - +#include #include +#include 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 mValidatedLedgerWorkQueue; + CanonicalTXSet mHeldTransactions{uint256()}; // A set of transactions to replay during the next close diff --git a/src/xrpld/app/ledger/detail/LedgerMaster.cpp b/src/xrpld/app/ledger/detail/LedgerMaster.cpp index 5b3275c04..950c397da 100644 --- a/src/xrpld/app/ledger/detail/LedgerMaster.cpp +++ b/src/xrpld/app/ledger/detail/LedgerMaster.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -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( + 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 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 diff --git a/src/xrpld/app/ledger/detail/ValidatedLedgerWorkQueue.h b/src/xrpld/app/ledger/detail/ValidatedLedgerWorkQueue.h new file mode 100644 index 000000000..35a8c1621 --- /dev/null +++ b/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 +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +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 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 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 + std::size_t + drain(Consumer&& consumer) + { + static_assert( + std::is_nothrow_invocable_v); + + std::size_t drained = 0; + for (;;) + { + std::optional 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 queue_; + std::optional active_; + bool drainScheduled_{false}; +}; + +} // namespace ripple::detail + +#endif