Compare commits

..

2 Commits

Author SHA1 Message Date
tequ
9f97d4643d Merge fixAMMClawbackRounding amendment into featureAMMClawback amendment 2026-02-25 15:27:37 +09:00
yinyiqian1
d9666cee77 fixAMMClawbackRounding: adjust last holder's LPToken balance (#5513)
Due to rounding, the LPTokenBalance of the last LP might not match the LP's trustline balance. This was fixed for `AMMWithdraw` in `fixAMMv1_1` by adjusting the LPTokenBalance to be the same as the trustline balance. Since `AMMClawback` is also performing a withdrawal, we need to adjust LPTokenBalance as well in `AMMClawback.`

This change includes:
1. Refactored `verifyAndAdjustLPTokenBalance` function in `AMMUtils`, which both`AMMWithdraw` and `AMMClawback` call to adjust LPTokenBalance.
2. Added the unit test `testLastHolderLPTokenBalance` to test the scenario.
3. Modify the existing unit tests for `fixAMMClawbackRounding`.
2026-02-25 15:27:33 +09:00
22 changed files with 161 additions and 1142 deletions

View File

@@ -68,17 +68,6 @@ target_link_libraries(xrpl.imports.main
$<$<BOOL:${voidstar}>:antithesis-sdk-cpp>
)
# date-tz for enhanced logging (always linked, code is #ifdef guarded)
if(TARGET date::date-tz)
target_link_libraries(xrpl.imports.main INTERFACE date::date-tz)
endif()
# BEAST_ENHANCED_LOGGING: enable for Debug builds OR when explicitly requested
# Uses generator expression so it works with multi-config generators (Xcode, VS, Ninja Multi-Config)
target_compile_definitions(xrpl.imports.main INTERFACE
$<$<OR:$<CONFIG:Debug>,$<BOOL:${BEAST_ENHANCED_LOGGING}>>:BEAST_ENHANCED_LOGGING=1>
)
include(add_module)
include(target_link_modules)

View File

@@ -1,106 +0,0 @@
#pragma once
#include <shared_mutex>
// On Linux (glibc), std::shared_mutex wraps pthread_rwlock_t initialised
// with PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP. This means a
// pending exclusive lock() blocks new shared (reader) acquisitions,
// causing reader starvation when writers contend frequently.
//
// On macOS / ARM (libc++), std::shared_mutex is already reader-preferring,
// so the same code behaves differently across platforms.
//
// This header provides reader_preferring_shared_mutex:
// - On Linux it wraps pthread_rwlock_t initialised with
// PTHREAD_RWLOCK_PREFER_READER_NP, matching macOS semantics.
// - On all other platforms it is a type alias for std::shared_mutex.
//
// The interface is identical to std::shared_mutex, so it works with
// std::shared_lock and std::unique_lock.
#if defined(__linux__)
#include <cerrno>
#include <pthread.h>
#include <stdexcept>
namespace ripple {
class reader_preferring_shared_mutex
{
pthread_rwlock_t rwlock_;
public:
reader_preferring_shared_mutex()
{
pthread_rwlockattr_t attr;
pthread_rwlockattr_init(&attr);
pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_READER_NP);
int rc = pthread_rwlock_init(&rwlock_, &attr);
pthread_rwlockattr_destroy(&attr);
if (rc != 0)
throw std::system_error(
rc, std::system_category(), "pthread_rwlock_init");
}
~reader_preferring_shared_mutex()
{
pthread_rwlock_destroy(&rwlock_);
}
reader_preferring_shared_mutex(reader_preferring_shared_mutex const&) =
delete;
reader_preferring_shared_mutex&
operator=(reader_preferring_shared_mutex const&) = delete;
// Exclusive (writer) locking
void
lock()
{
pthread_rwlock_wrlock(&rwlock_);
}
bool
try_lock()
{
return pthread_rwlock_trywrlock(&rwlock_) == 0;
}
void
unlock()
{
pthread_rwlock_unlock(&rwlock_);
}
// Shared (reader) locking
void
lock_shared()
{
pthread_rwlock_rdlock(&rwlock_);
}
bool
try_lock_shared()
{
return pthread_rwlock_tryrdlock(&rwlock_) == 0;
}
void
unlock_shared()
{
pthread_rwlock_unlock(&rwlock_);
}
};
} // namespace ripple
#else // !__linux__
namespace ripple {
// macOS, Windows, etc. — std::shared_mutex is already reader-preferring.
using reader_preferring_shared_mutex = std::shared_mutex;
} // namespace ripple
#endif

View File

@@ -511,7 +511,6 @@ public:
// End CachedSLEs functions.
private:
//@@start tagged-cache-fetch-promote
std::shared_ptr<T>
initialFetch(key_type const& key, std::lock_guard<mutex_type> const& l)
{
@@ -538,7 +537,6 @@ private:
m_cache.erase(cit);
return {};
}
//@@end tagged-cache-fetch-promote
void
collect_metrics()
@@ -601,7 +599,6 @@ private:
class ValueEntry
{
public:
//@@start tagged-cache-dual-tier
std::shared_ptr<mapped_type> ptr;
std::weak_ptr<mapped_type> weak_ptr;
clock_type::time_point last_access;
@@ -612,7 +609,6 @@ private:
: ptr(ptr_), weak_ptr(ptr_), last_access(last_access_)
{
}
//@@end tagged-cache-dual-tier
bool
isWeak() const
@@ -672,7 +668,6 @@ private:
stuffToSweep.first.reserve(partition.size());
stuffToSweep.second.reserve(partition.size());
{
//@@start tagged-cache-sweep-demote
auto cit = partition.begin();
while (cit != partition.end())
{
@@ -715,7 +710,6 @@ private:
++cit;
}
}
//@@end tagged-cache-sweep-demote
}
if (mapRemovals || cacheRemovals)

View File

@@ -20,13 +20,8 @@
#include <xrpl/basics/Log.h>
#include <xrpl/basics/contract.h>
#include <xrpl/beast/utility/instrumentation.h>
#ifndef BOOST_STACKTRACE_GNU_SOURCE_NOT_REQUIRED
#define BOOST_STACKTRACE_GNU_SOURCE_NOT_REQUIRED
#endif
#include <boost/stacktrace.hpp>
#include <cstdlib>
#include <iostream>
#include <sstream>
namespace ripple {
@@ -46,12 +41,7 @@ accessViolation() noexcept
void
LogThrow(std::string const& title)
{
std::ostringstream oss;
oss << title << '\n' << boost::stacktrace::stacktrace();
JLOG(debugLog().warn()) << oss.str();
// Also mirror to stderr so uncaught exceptions leave a trace even when
// log output is buffered/lost before terminate().
std::cerr << oss.str() << std::endl;
JLOG(debugLog().warn()) << title;
}
[[noreturn]] void

View File

@@ -169,7 +169,7 @@ public:
}
virtual void
onLedgerFetched(std::shared_ptr<InboundLedger> const&) override
onLedgerFetched() override
{
}

View File

@@ -83,9 +83,9 @@ public:
virtual std::size_t
fetchRate() = 0;
/** Called when a complete history ledger is obtained. */
/** Called when a complete ledger is obtained. */
virtual void
onLedgerFetched(std::shared_ptr<InboundLedger> const& inbound) = 0;
onLedgerFetched() = 0;
virtual void
gotFetchPack() = 0;

View File

@@ -50,8 +50,6 @@
#include <xrpl/protocol/digest.h>
#include <xrpl/protocol/jss.h>
#include <boost/optional.hpp>
#include <cstdlib>
#include <string_view>
#include <utility>
#include <vector>
@@ -61,33 +59,6 @@ namespace ripple {
create_genesis_t const create_genesis{};
namespace {
bool
isRWDBNullMode()
{
static bool const v = [] {
char const* e = std::getenv("XAHAU_RWDB_NULL");
return e && *e && std::string_view{e} != "0";
}();
return v;
}
template <class Map>
std::size_t
wireCompleteSHAMap(Map const& map)
{
std::size_t leaves = 0;
for (auto const& item : map)
{
(void)item;
++leaves;
}
return leaves;
}
} // namespace
uint256
calculateLedgerHash(LedgerInfo const& info)
{
@@ -278,7 +249,6 @@ Ledger::Ledger(
stateMap_.flushDirty(hotACCOUNT_NODE);
setImmutable();
setFullyWired();
}
Ledger::Ledger(
@@ -343,7 +313,6 @@ Ledger::Ledger(
// Create a new ledger that follows this one
Ledger::Ledger(Ledger const& prevLedger, NetClock::time_point closeTime)
: mImmutable(false)
, fullyWired_(prevLedger.isFullyWired())
, txMap_(SHAMapType::TRANSACTION, prevLedger.txMap_.family())
, stateMap_(prevLedger.stateMap_, true)
, fees_(prevLedger.fees_)
@@ -421,30 +390,6 @@ Ledger::setImmutable(bool rehash)
setup();
}
bool
Ledger::fullWireForUse(beast::Journal journal, char const* context) const
{
if (!isRWDBNullMode() || isFullyWired())
return true;
try
{
auto const stateLeaves = wireCompleteSHAMap(stateMap_);
auto const txLeaves = wireCompleteSHAMap(txMap_);
setFullyWired();
JLOG(journal.info())
<< context << ": fully wired ledger " << info_.seq << " ("
<< stateLeaves << " state leaves, " << txLeaves << " tx leaves)";
return true;
}
catch (SHAMapMissingNode const& e)
{
JLOG(journal.warn()) << context << ": incomplete ledger " << info_.seq
<< ": " << e.what();
return false;
}
}
// raw setters for catalogue
void
Ledger::setCloseFlags(int closeFlags)
@@ -1185,17 +1130,14 @@ loadLedgerHelper(LedgerInfo const& info, Application& app, bool acquire)
}
static void
finishLoadByIndexOrHash(std::shared_ptr<Ledger>& ledger, beast::Journal j)
finishLoadByIndexOrHash(
std::shared_ptr<Ledger> const& ledger,
Config const& config,
beast::Journal j)
{
if (!ledger)
return;
if (!ledger->fullWireForUse(j, "finishLoadByIndexOrHash"))
{
ledger.reset();
return;
}
XRPL_ASSERT(
ledger->read(keylet::fees()),
"ripple::finishLoadByIndexOrHash : valid ledger fees");
@@ -1213,13 +1155,7 @@ getLatestLedger(Application& app)
app.getRelationalDatabase().getNewestLedgerInfo();
if (!info)
return {std::shared_ptr<Ledger>(), {}, {}};
auto ledger = loadLedgerHelper(*info, app, true);
if (ledger &&
!ledger->fullWireForUse(app.journal("Ledger"), "getLatestLedger"))
{
ledger.reset();
}
return {ledger, info->seq, info->hash};
return {loadLedgerHelper(*info, app, true), info->seq, info->hash};
}
std::shared_ptr<Ledger>
@@ -1229,7 +1165,7 @@ loadByIndex(std::uint32_t ledgerIndex, Application& app, bool acquire)
app.getRelationalDatabase().getLedgerInfoByIndex(ledgerIndex))
{
std::shared_ptr<Ledger> ledger = loadLedgerHelper(*info, app, acquire);
finishLoadByIndexOrHash(ledger, app.journal("Ledger"));
finishLoadByIndexOrHash(ledger, app.config(), app.journal("Ledger"));
return ledger;
}
return {};
@@ -1242,7 +1178,7 @@ loadByHash(uint256 const& ledgerHash, Application& app, bool acquire)
app.getRelationalDatabase().getLedgerInfoByHash(ledgerHash))
{
std::shared_ptr<Ledger> ledger = loadLedgerHelper(*info, app, acquire);
finishLoadByIndexOrHash(ledger, app.journal("Ledger"));
finishLoadByIndexOrHash(ledger, app.config(), app.journal("Ledger"));
XRPL_ASSERT(
!ledger || ledger->info().hash == ledgerHash,
"ripple::loadByHash : ledger hash match if loaded");

View File

@@ -31,7 +31,6 @@
#include <xrpl/protocol/STLedgerEntry.h>
#include <xrpl/protocol/Serializer.h>
#include <xrpl/protocol/TxMeta.h>
#include <atomic>
#include <mutex>
namespace ripple {
@@ -295,21 +294,6 @@ public:
return mImmutable;
}
bool
isFullyWired() const
{
return fullyWired_.load(std::memory_order_acquire);
}
void
setFullyWired() const
{
fullyWired_.store(true, std::memory_order_release);
}
bool
fullWireForUse(beast::Journal journal, char const* context) const;
/* Mark this ledger as "should be full".
"Full" is metadata property of the ledger, it indicates
@@ -433,7 +417,6 @@ private:
defaultFees(Config const& config);
bool mImmutable;
mutable std::atomic<bool> fullyWired_{false};
// A SHAMap containing the transactions associated with this ledger.
SHAMap mutable txMap_;

View File

@@ -37,7 +37,6 @@
#include <xrpl/protocol/RippleLedgerHash.h>
#include <xrpl/protocol/STValidation.h>
#include <xrpl/protocol/messages.h>
#include <deque>
#include <optional>
#include <mutex>
@@ -348,12 +347,6 @@ private:
// The last ledger we handled fetching history
std::shared_ptr<Ledger const> mHistLedger;
// Sliding window of recently validated ledgers pinned in memory so their
// SHAMap state trees remain reachable via shared_ptr. Required when the
// node store does not persist state nodes (e.g. RWDB with
// XAHAU_RWDB_DISCARD_HOT_ACCOUNT_NODE). Guarded by m_mutex.
std::deque<std::shared_ptr<Ledger const>> mRetainedLedgers;
// Fully validated ledger, whether or not we have the ledger resident.
std::pair<uint256, LedgerIndex> mLastValidLedger{uint256(), 0};

View File

@@ -35,29 +35,12 @@
#include <boost/iterator/function_output_iterator.hpp>
#include <algorithm>
#include <cstdlib>
#include <limits>
#include <random>
#include <string_view>
namespace ripple {
using namespace std::chrono_literals;
namespace {
bool
isRWDBNullMode()
{
static bool const v = [] {
char const* e = std::getenv("XAHAU_RWDB_NULL");
return e && *e && std::string_view{e} != "0";
}();
return v;
}
} // namespace
enum {
// Number of peers to start with
peerCountStart = 5
@@ -137,24 +120,13 @@ InboundLedger::init(ScopedLockType& collectionLock)
JLOG(journal_.debug()) << "Acquiring ledger we already have in "
<< " local store. " << hash_;
// tryDB's getMissingNodes(1, filter) call already descended through
// every non-fullbelow branch and hooked children via canonicalizeChild.
// With the FullBelowCache liveness check in SHAMapSync, short-circuits
// only fire when the canonical subtree is proven alive via TreeNodeCache,
// so read-time lazy fetches are guaranteed to resolve. No upfront walk
// needed.
if (isRWDBNullMode() && !mLedger->isFullyWired())
mLedger->setFullyWired();
XRPL_ASSERT(
mLedger->read(keylet::fees()),
"ripple::InboundLedger::init : valid ledger fees");
mLedger->setImmutable();
if (mReason == Reason::HISTORY)
{
app_.getInboundLedgers().onLedgerFetched(shared_from_this());
return;
}
app_.getLedgerMaster().storeLedger(mLedger);
@@ -379,6 +351,10 @@ InboundLedger::tryDB(NodeStore::Database& srcDB)
{
JLOG(journal_.debug()) << "Had everything locally";
complete_ = true;
XRPL_ASSERT(
mLedger->read(keylet::fees()),
"ripple::InboundLedger::tryDB : valid ledger fees");
mLedger->setImmutable();
}
}
@@ -477,25 +453,14 @@ InboundLedger::done()
if (complete_ && !failed_ && mLedger)
{
// Sync's addKnownNode calls have canonicalized every arriving node
// into TreeNodeCache and hooked each into its parent via
// canonicalizeChild. With the FullBelowCache liveness check in
// SHAMapSync, any FBC short-circuit during sync's getMissingNodes
// walk is only taken when the canonical subtree is alive, so
// read-time lazy fetches are guaranteed to resolve via
// TreeNodeCache. No post-sync walk needed.
if (isRWDBNullMode() && !mLedger->isFullyWired())
mLedger->setFullyWired();
XRPL_ASSERT(
mLedger->read(keylet::fees()),
"ripple::InboundLedger::done : valid ledger fees");
mLedger->setImmutable();
switch (mReason)
{
case Reason::HISTORY:
app_.getInboundLedgers().onLedgerFetched(shared_from_this());
app_.getInboundLedgers().onLedgerFetched();
break;
default:
app_.getLedgerMaster().storeLedger(mLedger);
@@ -508,42 +473,6 @@ InboundLedger::done()
jtLEDGER_DATA, "AcquisitionDone", [self = shared_from_this()]() {
if (self->complete_ && !self->failed_)
{
if (!isRWDBNullMode() && self->mReason != Reason::HISTORY)
{
// Prime the state tree BEFORE checkAccept so consensus
// never sees a lazy tree. Runs off any inbound lock —
// this job is dispatched without mtx_ held.
// visitDifferences against prior validated walks only
// the delta; canonicalization means shared subtrees are
// the same inner objects (already wired). Gated on
// non-HISTORY to avoid paying on historical backfills.
auto const prior =
self->app_.getLedgerMaster().getValidatedLedger();
SHAMap const* have = prior ? &prior->stateMap() : nullptr;
try
{
std::size_t walked = 0;
self->mLedger->stateMap().visitDifferences(
have, [&walked](SHAMapTreeNode const&) {
++walked;
return true;
});
JLOG(self->journal_.info())
<< "Inbound prime: ledger "
<< self->mLedger->info().seq << " wired " << walked
<< (have ? " delta nodes vs prior validated"
: " nodes (first full walk)");
}
catch (SHAMapMissingNode const& e)
{
JLOG(self->journal_.warn())
<< "Inbound prime: incomplete state tree for "
<< "ledger " << self->mLedger->info().seq << ": "
<< e.what();
}
}
self->app_.getLedgerMaster().checkAccept(self->getLedger());
self->app_.getLedgerMaster().tryAdvance();
}
@@ -970,7 +899,6 @@ InboundLedger::receiveNode(protocol::TMLedgerData& packet, SHAMapAddNode& san)
{
auto const f = filter.get();
//@@start receive-node-link-loop
for (auto const& node : packet.nodes())
{
auto const nodeID = deserializeSHAMapNodeID(node.nodeid());
@@ -993,7 +921,6 @@ InboundLedger::receiveNode(protocol::TMLedgerData& packet, SHAMapAddNode& san)
return;
}
}
//@@end receive-node-link-loop
}
catch (std::exception const& e)
{

View File

@@ -22,7 +22,6 @@
#include <xrpld/app/main/Application.h>
#include <xrpld/app/misc/NetworkOPs.h>
#include <xrpld/core/JobQueue.h>
#include <xrpld/ledger/View.h>
#include <xrpld/perflog/PerfLog.h>
#include <xrpl/basics/DecayingSample.h>
#include <xrpl/basics/Log.h>
@@ -31,9 +30,7 @@
#include <xrpl/beast/core/LexicalCast.h>
#include <xrpl/protocol/jss.h>
#include <deque>
#include <exception>
#include <limits>
#include <memory>
#include <mutex>
#include <vector>
@@ -309,27 +306,11 @@ public:
return 60 * fetchRate_.value(m_clock.now());
}
// Should only be called with a complete inbound ledger that has
// a reason of history.
// Should only be called with an inboundledger that has
// a reason of history
void
onLedgerFetched(std::shared_ptr<InboundLedger> const& inbound) override
onLedgerFetched() override
{
if (!inbound)
return;
auto const ledger = inbound->getLedger();
if (!ledger || !ledger->isFullyWired())
return;
{
ScopedLockType sl(mLock);
if (auto const it = mLedgers.find(ledger->info().hash);
it != mLedgers.end() && it->second.get() == inbound.get())
{
mLedgers.erase(it);
}
}
std::lock_guard lock(fetchRateMutex_);
fetchRate_.add(1, m_clock.now());
}

View File

@@ -523,8 +523,6 @@ LedgerMaster::clearLedger(std::uint32_t seq)
}
mCompleteLedgers.erase(seq);
JLOG(m_journal.info()) << "mCompleteLedgers[clearLedger]: erase(" << seq
<< ") -> " << to_string(mCompleteLedgers);
}
bool
@@ -690,9 +688,6 @@ LedgerMaster::tryFill(std::shared_ptr<Ledger const> ledger)
{
std::lock_guard ml(mCompleteLock);
mCompleteLedgers.insert(range(minHas, maxHas));
JLOG(m_journal.info())
<< "mCompleteLedgers[tryFill/inner]: insert(" << minHas
<< "-" << maxHas << ") -> " << to_string(mCompleteLedgers);
}
maxHas = minHas;
ledgerHashes = app_.getRelationalDatabase().getHashesByIndex(
@@ -702,12 +697,11 @@ LedgerMaster::tryFill(std::shared_ptr<Ledger const> ledger)
if (it == ledgerHashes.end())
break;
auto const& firstHash = ledgerHashes.begin()->second.ledgerHash;
if (!nodeStore.fetchNodeObject(
firstHash, ledgerHashes.begin()->first) &&
!getLedgerByHash(firstHash))
ledgerHashes.begin()->second.ledgerHash,
ledgerHashes.begin()->first))
{
// Not in node store and not in memory — genuinely missing
// The ledger is not backed by the node store
JLOG(m_journal.warn()) << "SQL DB ledger sequence " << seq
<< " mismatches node store";
break;
@@ -723,9 +717,6 @@ LedgerMaster::tryFill(std::shared_ptr<Ledger const> ledger)
{
std::lock_guard ml(mCompleteLock);
mCompleteLedgers.insert(range(minHas, maxHas));
JLOG(m_journal.info())
<< "mCompleteLedgers[tryFill/final]: insert(" << minHas << "-"
<< maxHas << ") -> " << to_string(mCompleteLedgers);
}
{
std::lock_guard ml(m_mutex);
@@ -869,130 +860,9 @@ LedgerMaster::setFullLedger(
pendSaveValidated(app_, ledger, isSynchronous, isCurrent);
// Pin a sliding window of recently validated current ledgers so their
// SHAMap state trees stay resident via shared_ptr. This tracks the
// server's active online band rather than retaining arbitrary historical
// backfill ledgers.
std::vector<std::shared_ptr<Ledger const>> retiredLedgers;
if (isCurrent && ledger_history_ > 0)
{
std::lock_guard ml(m_mutex);
bool const isFirst = mRetainedLedgers.empty();
mRetainedLedgers.push_back(ledger);
while (mRetainedLedgers.size() > ledger_history_)
{
retiredLedgers.push_back(std::move(mRetainedLedgers.front()));
mRetainedLedgers.pop_front();
}
// Legacy bootstrap for lazy trees. In null mode the ledger has
// already been fully wired before it reaches retention, so there is
// nothing left to do here.
if (isFirst && !ledger->isFullyWired())
{
try
{
std::size_t leafCount = 0;
for (auto const& item : ledger->stateMap())
{
(void)item;
++leafCount;
}
JLOG(m_journal.info())
<< "Retention: primed state tree for ledger "
<< ledger->info().seq << " (" << leafCount << " leaves)";
}
catch (SHAMapMissingNode const& e)
{
JLOG(m_journal.warn())
<< "Retention: incomplete state tree for ledger "
<< ledger->info().seq << ": " << e.what();
}
}
}
// In memory-resident mode we retire every time a Ledger falls off
// mRetainedLedgers. No OperatingMode gate: mRetainedLedgers already
// caps at ledger_history_ on every publish regardless of
// DISCONNECTED/SYNCING/TRACKING/FULL, so mCompleteLedgers should
// track it step-for-step. The earlier sticky-TRACKING gate was
// defensive against fetchForHistory re-inserting historical seqs
// and fighting the prune, but fetchForHistory is now
// !memoryResidentMode-gated in doAdvance, so there's nothing left
// to fight.
bool const shouldRetire = app_.getSHAMapStore().memoryResidentMode();
// The mCompleteLedgers insert of the new seq AND the bulk-prefix prune
// of retired seqs both run under one mCompleteLock acquisition. This
// closes the transient insert-before-prune window where observers
// would see ledger_history + 1 entries briefly. Peers get a
// complete_ledgers range that stays tight at exactly ledger_history.
LedgerIndex maxRetiredSeq = 0;
if (shouldRetire)
{
for (auto const& r : retiredLedgers)
{
if (r && r->info().seq > maxRetiredSeq)
maxRetiredSeq = r->info().seq;
}
}
{
std::lock_guard ml(mCompleteLock);
mCompleteLedgers.insert(ledger->info().seq);
// Inline bulk-prefix prune under the same lock. This is the body
// of clearPriorLedgers without its own lock acquisition. Pinning
// is preserved.
if (maxRetiredSeq > 0)
{
auto pinnedCopy = mPinnedLedgers;
RangeSet<std::uint32_t> toClear;
toClear.insert(range(0u, maxRetiredSeq));
for (auto const& interval : toClear)
mCompleteLedgers.erase(interval);
for (auto const& interval : pinnedCopy)
mCompleteLedgers.insert(interval);
JLOG(m_journal.info())
<< "mCompleteLedgers[setFullLedger/insert+prune]: insert("
<< ledger->info().seq << ") + clearPrior(" << maxRetiredSeq + 1
<< ") -> " << to_string(mCompleteLedgers);
}
else
{
JLOG(m_journal.info())
<< "mCompleteLedgers[setFullLedger]: insert("
<< ledger->info().seq << ") -> " << to_string(mCompleteLedgers);
}
}
// Heavy work goes async (LedgerHistory cache eviction, relational
// deletes, and the shared_ptr destruction cascade through the retired
// Ledgers' SHAMap spines). The retired Ledgers stay alive in the
// captured vector until the job runs; destruction happens on the
// worker thread, off doAdvance's critical path.
//
// Dispatch unconditionally whenever we have retired Ledgers — even
// pre-TRACKING, where shouldRetire is false and we skip the
// mCompleteLedgers / relational / LedgerHistory pruning. The job
// still owns the shared_ptrs, so their destruction cascade runs on
// the worker, not on the advance thread. Without this, retired
// Ledgers fall out of scope synchronously in setFullLedger and the
// advance thread blocks on a million-leaf destruction per publish,
// producing the sync-stall-then-flurry pattern during catch-up.
if (!retiredLedgers.empty())
{
app_.getJobQueue().addJob(
jtLEDGER_DATA,
"retireLedgers",
[&app = app_, shouldRetire, retired = std::move(retiredLedgers)]() {
if (shouldRetire)
app.getSHAMapStore().retireLedgers(retired);
// Otherwise `retired` just destructs here on this
// worker thread as the lambda exits — bookkeeping
// side effects skipped, destruction cascade kept off
// the advance thread either way.
});
}
{
@@ -1793,12 +1663,6 @@ LedgerMaster::getCloseTimeByHash(
LedgerHash const& ledgerHash,
std::uint32_t index)
{
// Prefer an in-memory Ledger (retained / history cache) over the node
// store so this works in RWDB-only configs where headers may not be
// persisted long-term.
if (auto ledger = getLedgerByHash(ledgerHash))
return ledger->info().closeTime;
auto nodeObject = app_.getNodeStore().fetchNodeObject(ledgerHash, index);
if (nodeObject && (nodeObject->getData().size() >= 120))
{
@@ -1951,9 +1815,6 @@ LedgerMaster::setLedgerRangePresent(
{
std::lock_guard sl(mCompleteLock);
mCompleteLedgers.insert(range(minV, maxV));
JLOG(m_journal.info()) << "mCompleteLedgers[setLedgerRangePresent]: insert("
<< minV << "-" << maxV << ") -> "
<< to_string(mCompleteLedgers);
if (pin)
{
@@ -1997,8 +1858,6 @@ LedgerMaster::clearPriorLedgers(LedgerIndex seq)
for (auto const& interval : pinnedCopy)
mCompleteLedgers.insert(interval);
JLOG(m_journal.info()) << "mCompleteLedgers[clearPriorLedgers]: clearPrior("
<< seq << ") -> " << to_string(mCompleteLedgers);
JLOG(m_journal.debug()) << "clearPriorLedgers: after restoration, pinned="
<< to_string(mPinnedLedgers);
}
@@ -2071,16 +1930,7 @@ LedgerMaster::fetchForHistory(
mHistLedger = ledger;
fillInProgress = mFillInProgress;
}
// tryFill walks back the ledger's parent-hash chain and marks
// every seq it finds in mCompleteLedgers, so peers know we
// have the whole chain. Under memory-resident mode we only
// actually retain ledger_history ledgers, so the walk would
// either (a) duplicate bookkeeping we already have for the
// retained range, or (b) mark older seqs we can't actually
// serve. Skip it and let mCompleteLedgers track only the
// ledgers mRetainedLedgers structurally holds.
if (fillInProgress == 0 &&
!app_.getSHAMapStore().memoryResidentMode() &&
app_.getRelationalDatabase().getHashByIndex(seq - 1) ==
ledger->info().parentHash)
{
@@ -2156,14 +2006,7 @@ LedgerMaster::doAdvance(std::unique_lock<std::recursive_mutex>& sl)
auto const pubLedgers = findNewLedgersToPublish(sl);
if (pubLedgers.empty())
{
// History backfill is pointless in memory-resident mode: our
// retention IS ledger_history, and prevMissing finds gaps just
// below the retention window that we'd re-fetch only to
// immediately retire again — producing the classic flicker
// where mCompleteLedgers oscillates between ledger_history
// and ledger_history+1.
if (!standalone_ && !app_.getSHAMapStore().memoryResidentMode() &&
!app_.getFeeTrack().isLoadedLocal() &&
if (!standalone_ && !app_.getFeeTrack().isLoadedLocal() &&
(app_.getJobQueue().getJobCount(jtPUBOLDLEDGER) < 10) &&
(mValidLedgerSeq == mPubLedgerSeq) &&
(getValidatedLedgerAge() < MAX_LEDGER_AGE_ACQUIRE) &&

View File

@@ -897,11 +897,9 @@ NetworkOPsImp::setHeartbeatTimer()
heartbeatTimer_,
mConsensus.parms().ledgerGRANULARITY,
[this]() {
// Run the heartbeat directly on the io_service thread instead
// of posting to the JobQueue. This prevents heavy RPC load
// from starving the consensus heartbeat timer — the io_service
// thread pool is independent of the JobQueue worker pool.
processHeartbeatTimer();
m_job_queue.addJob(jtNETOP_TIMER, "NetOPs.heartbeat", [this]() {
processHeartbeatTimer();
});
},
[this]() { setHeartbeatTimer(); });
}
@@ -941,82 +939,66 @@ NetworkOPsImp::processHeartbeatTimer()
RclConsensusLogger clog(
"Heartbeat Timer", mConsensus.validating(), m_journal);
{
// Use try_to_lock so the heartbeat never blocks on masterMutex.
// If apply() or another operation is holding it, skip the non-critical
// peer/mode checks and proceed directly to timerEntry() — ensuring
// consensus timing is never delayed by mutex contention.
std::unique_lock lock{app_.getMasterMutex(), std::try_to_lock};
std::unique_lock lock{app_.getMasterMutex()};
if (lock.owns_lock())
// VFALCO NOTE This is for diagnosing a crash on exit
LoadManager& mgr(app_.getLoadManager());
mgr.resetDeadlockDetector();
std::size_t const numPeers = app_.overlay().size();
// do we have sufficient peers? If not, we are disconnected.
if (numPeers < minPeerCount_)
{
// VFALCO NOTE This is for diagnosing a crash on exit
LoadManager& mgr(app_.getLoadManager());
mgr.resetDeadlockDetector();
std::size_t const numPeers = app_.overlay().size();
// do we have sufficient peers? If not, we are disconnected.
if (numPeers < minPeerCount_)
if (mMode != OperatingMode::DISCONNECTED)
{
if (mMode != OperatingMode::DISCONNECTED)
{
setMode(OperatingMode::DISCONNECTED);
std::stringstream ss;
ss << "Node count (" << numPeers << ") has fallen "
<< "below required minimum (" << minPeerCount_ << ").";
JLOG(m_journal.warn()) << ss.str();
CLOG(clog.ss()) << "set mode to DISCONNECTED: " << ss.str();
}
else
{
CLOG(clog.ss())
<< "already DISCONNECTED. too few peers (" << numPeers
<< "), need at least " << minPeerCount_;
}
// MasterMutex lock need not be held to call
// setHeartbeatTimer()
lock.unlock();
// We do not call mConsensus.timerEntry until there are
// enough peers providing meaningful inputs to consensus
setHeartbeatTimer();
return;
setMode(OperatingMode::DISCONNECTED);
std::stringstream ss;
ss << "Node count (" << numPeers << ") has fallen "
<< "below required minimum (" << minPeerCount_ << ").";
JLOG(m_journal.warn()) << ss.str();
CLOG(clog.ss()) << "set mode to DISCONNECTED: " << ss.str();
}
if (mMode == OperatingMode::DISCONNECTED)
{
setMode(OperatingMode::CONNECTED);
JLOG(m_journal.info())
<< "Node count (" << numPeers << ") is sufficient.";
CLOG(clog.ss()) << "setting mode to CONNECTED based on "
<< numPeers << " peers. ";
}
// Check if the last validated ledger forces a change between
// these states.
auto origMode = mMode.load();
CLOG(clog.ss()) << "mode: " << strOperatingMode(origMode, true);
if (mMode == OperatingMode::SYNCING)
setMode(OperatingMode::SYNCING);
else if (mMode == OperatingMode::CONNECTED)
setMode(OperatingMode::CONNECTED);
auto newMode = mMode.load();
if (origMode != newMode)
else
{
CLOG(clog.ss())
<< ", changing to " << strOperatingMode(newMode, true);
<< "already DISCONNECTED. too few peers (" << numPeers
<< "), need at least " << minPeerCount_;
}
CLOG(clog.ss()) << ". ";
// MasterMutex lock need not be held to call setHeartbeatTimer()
lock.unlock();
// We do not call mConsensus.timerEntry until there are enough
// peers providing meaningful inputs to consensus
setHeartbeatTimer();
return;
}
else
if (mMode == OperatingMode::DISCONNECTED)
{
JLOG(m_journal.debug())
<< "Heartbeat: masterMutex contended, skipping "
"peer/mode checks";
CLOG(clog.ss())
<< "masterMutex contended, skipping peer/mode checks. ";
setMode(OperatingMode::CONNECTED);
JLOG(m_journal.info())
<< "Node count (" << numPeers << ") is sufficient.";
CLOG(clog.ss()) << "setting mode to CONNECTED based on " << numPeers
<< " peers. ";
}
// Check if the last validated ledger forces a change between these
// states.
auto origMode = mMode.load();
CLOG(clog.ss()) << "mode: " << strOperatingMode(origMode, true);
if (mMode == OperatingMode::SYNCING)
setMode(OperatingMode::SYNCING);
else if (mMode == OperatingMode::CONNECTED)
setMode(OperatingMode::CONNECTED);
auto newMode = mMode.load();
if (origMode != newMode)
{
CLOG(clog.ss())
<< ", changing to " << strOperatingMode(newMode, true);
}
CLOG(clog.ss()) << ". ";
}
mConsensus.timerEntry(app_.timeKeeper().closeTime(), clog.ss());

View File

@@ -97,29 +97,6 @@ public:
*/
virtual std::optional<LedgerIndex>
minimumOnline() const = 0;
/** True if this store is configured for memory-resident retention.
In memory-resident mode (null nodestore) the rotation thread does
not run; ledgers are retired one at a time as new validated ledgers
arrive (see retireLedger), and online_delete is effectively
ignored. The retention bound is ledger_history.
*/
virtual bool
memoryResidentMode() const = 0;
/** Retire a batch of ledgers from memory-resident retention.
Called by LedgerMaster when one or more Ledgers drop off the back
of the retention deque. Synchronously prunes mCompleteLedgers, the
LedgerHistory cache, and per-seq relational rows for these ledgers.
Relational/cache pruning collapses to a single prefix-delete at the
highest retired sequence, so plural calls are no costlier than a
singular one. No-op outside memory-resident mode.
*/
virtual void
retireLedgers(
std::vector<std::shared_ptr<Ledger const>> const& ledgers) = 0;
};
//------------------------------------------------------------------------------

View File

@@ -31,45 +31,7 @@
#include <boost/algorithm/string/predicate.hpp>
#include <cstdlib>
#include <string_view>
namespace ripple {
namespace {
constexpr std::uint32_t minimumDeletionIntervalExperimental = 8;
bool
isRWDBNullMode()
{
static bool const enabled = [] {
char const* e = std::getenv("XAHAU_RWDB_NULL");
return e && *e && std::string_view{e} != "0";
}();
return enabled;
}
std::uint32_t
minimumDeleteIntervalForMode(Config const& config, bool isMemoryBackend)
{
if (config.standalone())
return minimumDeletionIntervalExperimental;
if (isMemoryBackend && isRWDBNullMode())
return minimumDeletionIntervalExperimental;
return 256;
}
bool
skipNodeStoreRotateForMode(bool isMemoryBackend)
{
return isMemoryBackend && isRWDBNullMode();
}
} // namespace
void
SHAMapStoreImp::SavedStateDB::init(
BasicConfig const& config,
@@ -154,43 +116,6 @@ SHAMapStoreImp::SHAMapStoreImp(
}
get_if_exists(section, "online_delete", deleteInterval_);
auto const backendType = get(section, "type");
isMemoryBackend_ = boost::iequals(backendType, "rwdb") ||
boost::iequals(backendType, "none");
// type=none is the declared null-nodestore config (via NullFactory).
// Propagate to XAHAU_RWDB_NULL so isRWDBNullMode() in other components
// (SHAMapSync, InboundLedger, Ledger) picks up null-mode semantics via
// their file-local helpers. overwrite=0 preserves any value the user
// has already set.
if (boost::iequals(backendType, "none"))
::setenv("XAHAU_RWDB_NULL", "1", 0);
// Memory-resident mode is implied by null-mode semantics. The rotation
// thread doesn't run; per-ledger retirement happens via retireLedger
// called from LedgerMaster::setFullLedger when a ledger drops off the
// back of mRetainedLedgers.
memoryResidentMode_ = isRWDBNullMode();
if (memoryResidentMode_)
{
// No rotation thread will run, so working_ stays false and
// rendezvous() short-circuits cleanly.
working_ = false;
JLOG(journal_.info())
<< "Memory-resident retention mode enabled (no rotation thread); "
<< "ledger_history=" << config.LEDGER_HISTORY
<< " is the retention bound";
}
// For RWDB, default online_delete to ledger_history only if user did not
// explicitly set online_delete. Clamp to the minimum so an implicit
// value never triggers the "online_delete must be at least …" throw.
if (isMemoryBackend_ && deleteInterval_ == 0)
{
auto const minInterval =
minimumDeleteIntervalForMode(config, isMemoryBackend_);
deleteInterval_ = std::max(config.LEDGER_HISTORY, minInterval);
}
if (deleteInterval_)
{
@@ -210,8 +135,9 @@ SHAMapStoreImp::SHAMapStoreImp(
get_if_exists(section, "advisory_delete", advisoryDelete_);
auto const minInterval =
minimumDeleteIntervalForMode(config, isMemoryBackend_);
auto const minInterval = config.standalone()
? minimumDeletionIntervalSA_
: minimumDeletionInterval_;
if (deleteInterval_ < minInterval)
{
Throw<std::runtime_error>(
@@ -228,7 +154,7 @@ SHAMapStoreImp::SHAMapStoreImp(
}
state_db_.init(config, dbName_);
if (!isMemoryBackend_)
if (!config.mem_backend())
dbPaths();
}
}
@@ -399,152 +325,64 @@ SHAMapStoreImp::run()
if (healthWait() == stopping)
return;
if (isMemoryBackend_)
{
//@@start rwdb-null-skip-rotation
if (skipNodeStoreRotateForMode(isMemoryBackend_))
{
JLOG(journal_.debug())
<< "RWDB null mode: skipping node store rotation";
JLOG(journal_.debug()) << "copying ledger " << validatedSeq;
std::uint64_t nodeCount = 0;
try
{
validatedLedger->stateMap().snapShot(false)->visitNodes(
std::bind(
&SHAMapStoreImp::copyNode,
this,
std::ref(nodeCount),
std::placeholders::_1));
}
catch (SHAMapMissingNode const& e)
{
JLOG(journal_.error())
<< "Missing node while copying ledger before rotate: "
<< e.what();
continue;
}
if (healthWait() == stopping)
return;
// Only log if we completed without a "health" abort
JLOG(journal_.debug()) << "copied ledger " << validatedSeq
<< " nodecount " << nodeCount;
JLOG(journal_.debug()) << "freshening caches";
freshenCaches();
if (healthWait() == stopping)
return;
// Only log if we completed without a "health" abort
JLOG(journal_.debug()) << validatedSeq << " freshened caches";
JLOG(journal_.debug()) << "Making a new backend";
auto newBackend = makeBackendRotating();
JLOG(journal_.debug())
<< validatedSeq << " new backend " << newBackend->getName();
clearCaches(validatedSeq);
if (healthWait() == stopping)
return;
lastRotated = validatedSeq;
dbRotating_->rotate(
std::move(newBackend),
[&](std::string const& writableName,
std::string const& archiveName) {
SavedState savedState;
savedState.writableDb = writableName;
savedState.archiveDb = archiveName;
savedState.lastRotated = lastRotated;
state_db_.setState(savedState);
lastRotated = validatedSeq;
state_db_.setLastRotated(lastRotated);
clearCaches(validatedSeq);
continue;
}
//@@end rwdb-null-skip-rotation
});
// For RWDB: copy only the current validated ledger's live
// state nodes into a fresh backend that is not yet shared,
// avoiding both exclusive-lock contention on the live
// writable backend AND stale-node accumulation.
//
// copyArchiveTo would carry forward ALL archive entries
// (including stale nodes from older ledger versions that
// were promoted via fetch duplication), causing unbounded
// memory growth across rotation cycles.
JLOG(journal_.debug()) << "RWDB: copying live state for rotation";
auto newBackend = makeBackendRotating();
std::uint64_t nodeCount = 0;
bool aborted = false;
try
{
//@@start rwdb-visit-copy
validatedLedger->stateMap().snapShot(false)->visitNodes(
[&](SHAMapTreeNode& node) -> bool {
auto const hash = node.getHash().as_uint256();
// Fetch the NodeObject from the rotating DB
// (checks writable then archive) and store it
// directly in the new unshared backend.
auto obj = dbRotating_->fetchNodeObject(
hash,
0,
NodeStore::FetchType::synchronous,
false);
if (obj)
newBackend->store(obj);
if ((++nodeCount % checkHealthInterval_) == 0)
{
if (healthWait() == stopping)
{
aborted = true;
return false;
}
}
return true;
});
//@@end rwdb-visit-copy
}
catch (SHAMapMissingNode const& e)
{
JLOG(journal_.error())
<< "Missing node while copying state before rotate: "
<< e.what();
continue;
}
if (aborted)
return;
JLOG(journal_.debug())
<< "RWDB: copied " << nodeCount << " live nodes";
ledgerMaster_->clearLedgerCachePrior(validatedSeq);
lastRotated = validatedSeq;
dbRotating_->rotate(
std::move(newBackend),
[&](std::string const& writableName,
std::string const& archiveName) {
SavedState savedState;
savedState.writableDb = writableName;
savedState.archiveDb = archiveName;
savedState.lastRotated = lastRotated;
state_db_.setState(savedState);
ledgerMaster_->clearLedgerCachePrior(validatedSeq);
});
JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
}
else
{
JLOG(journal_.debug()) << "copying ledger " << validatedSeq;
std::uint64_t nodeCount = 0;
try
{
validatedLedger->stateMap().snapShot(false)->visitNodes(
std::bind(
&SHAMapStoreImp::copyNode,
this,
std::ref(nodeCount),
std::placeholders::_1));
}
catch (SHAMapMissingNode const& e)
{
JLOG(journal_.error())
<< "Missing node while copying ledger before rotate: "
<< e.what();
continue;
}
if (healthWait() == stopping)
return;
JLOG(journal_.debug()) << "copied ledger " << validatedSeq
<< " nodecount " << nodeCount;
JLOG(journal_.debug()) << "freshening caches";
freshenCaches();
if (healthWait() == stopping)
return;
JLOG(journal_.debug()) << validatedSeq << " freshened caches";
JLOG(journal_.trace()) << "Making a new backend";
auto newBackend = makeBackendRotating();
JLOG(journal_.debug())
<< validatedSeq << " new backend " << newBackend->getName();
clearCaches(validatedSeq);
if (healthWait() == stopping)
return;
lastRotated = validatedSeq;
dbRotating_->rotate(
std::move(newBackend),
[&](std::string const& writableName,
std::string const& archiveName) {
SavedState savedState;
savedState.writableDb = writableName;
savedState.archiveDb = archiveName;
savedState.lastRotated = lastRotated;
state_db_.setState(savedState);
clearCaches(validatedSeq);
});
JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
}
JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
}
}
}
@@ -842,74 +680,6 @@ SHAMapStoreImp::minimumOnline() const
return app_.getLedgerMaster().minSqlSeq();
}
void
SHAMapStoreImp::retireLedgers(
std::vector<std::shared_ptr<Ledger const>> const& ledgers)
{
if (!memoryResidentMode_ || ledgers.empty())
return;
// Memory-resident retirement: bulk-prefix prune everything at or
// below the max retired seq. This single pattern handles both the
// steady-state case (one ledger in `ledgers`) and the post-catch-up
// case where LedgerHistory and the relational tables accumulated
// many seqs below the retention window during catch-up — retireLedgers
// is only called once the node is FULL, so the first invocation
// after catch-up collapses all that accumulation in one pass.
//
// This function runs on a JobQueue worker, off the publish thread,
// so the expensive work doesn't block doAdvance:
//
// - clearPriorLedgers is idempotent here. LedgerMaster::setFullLedger
// already pruned mCompleteLedgers synchronously before posting
// this job, keeping the reported complete_ledgers range tight.
// Still called here for safety / external callers of retireLedgers.
//
// - clearLedgerCachePrior iterates the LedgerHistory cache and
// drops the shared_ptrs held there. This is where the heavy
// destruction cascade happens: Ledger → stateMap() SHAMap →
// canonical inner nodes → their children_ → etc. Thousands of
// shared_ptr decrements and TaggedCache weak_ptr bookkeeping
// per ledger. Kept off the publish thread by the job post.
//
// - Relational deletes are prefix operations; under RWDB-relational
// these are in-memory map.erase() calls (fast).
//
// - The `ledgers` vector going out of scope when this function
// returns drops the last strong references held by the job
// closure, kicking off destruction of any Ledgers that were
// only still alive via that capture.
//
// clearPriorLedgers preserves pinned ledgers.
LedgerIndex maxSeq = 0;
for (auto const& ledger : ledgers)
{
if (ledger && ledger->info().seq > maxSeq)
maxSeq = ledger->info().seq;
}
if (maxSeq == 0)
return;
auto& lm = app_.getLedgerMaster();
lm.clearPriorLedgers(maxSeq + 1);
lm.clearLedgerCachePrior(maxSeq + 1);
if (auto* db = dynamic_cast<SQLiteDatabase*>(&app_.getRelationalDatabase()))
{
if (app_.config().useTxTables())
{
db->deleteTransactionsBeforeLedgerSeq(maxSeq + 1);
db->deleteAccountTransactionsBeforeLedgerSeq(maxSeq + 1);
}
db->deleteBeforeLedgerSeq(maxSeq + 1);
}
JLOG(journal_.info()) << "retireLedgers: pruned everything at or below seq "
<< maxSeq << " (" << ledgers.size()
<< " popped this batch)";
}
//------------------------------------------------------------------------------
std::unique_ptr<SHAMapStore>

View File

@@ -101,10 +101,6 @@ private:
std::uint32_t deleteInterval_ = 0;
bool advisoryDelete_ = false;
bool isMemoryBackend_ = false;
// Memory-resident mode: skip the rotation thread entirely; per-ledger
// retirement happens via retireLedger called from LedgerMaster.
bool memoryResidentMode_ = false;
std::uint32_t deleteBatch_ = 100;
std::chrono::milliseconds backOff_{100};
std::chrono::seconds ageThreshold_{60};
@@ -180,16 +176,6 @@ public:
std::optional<LedgerIndex>
minimumOnline() const override;
bool
memoryResidentMode() const override
{
return memoryResidentMode_;
}
void
retireLedgers(
std::vector<std::shared_ptr<Ledger const>> const& ledgers) override;
private:
// callback for visitNodes
bool
@@ -251,8 +237,6 @@ public:
void
start() override
{
if (memoryResidentMode_)
return;
if (deleteInterval_)
thread_ = std::thread(&SHAMapStoreImp::run, this);
}

View File

@@ -55,15 +55,6 @@ public:
std::function<void(
std::string const& writableName,
std::string const& archiveName)> const& f) = 0;
/** Populate @a dest with every object in the archive backend.
Used by in-memory (RWDB) backends to pre-populate a new writable
backend before rotation, avoiding per-node write-lock contention on
the live writable backend. @a dest must not yet be shared.
*/
virtual void
copyArchiveTo(Backend& dest) = 0;
};
} // namespace NodeStore

View File

@@ -3,16 +3,12 @@
#include <xrpld/nodestore/detail/DecodedBlob.h>
#include <xrpld/nodestore/detail/EncodedBlob.h>
#include <xrpld/nodestore/detail/codec.h>
#include <xrpl/basics/ReaderPreferringSharedMutex.h>
#include <xrpl/basics/contract.h>
#include <boost/beast/core/string.hpp>
#include <boost/core/ignore_unused.hpp>
#include <boost/unordered/concurrent_flat_map.hpp>
#include <cstdlib>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <string_view>
namespace ripple {
namespace NodeStore {
@@ -38,7 +34,8 @@ private:
using DataStore =
std::map<uint256, std::vector<std::uint8_t>>; // Store compressed blob
// data
mutable reader_preferring_shared_mutex mutex_;
mutable std::recursive_mutex
mutex_; // Only needed for std::map implementation
DataStore table_;
@@ -68,7 +65,7 @@ public:
void
open(bool createIfMissing) override
{
std::unique_lock lock(mutex_);
std::lock_guard lock(mutex_);
if (isOpen_)
Throw<std::runtime_error>("already open");
isOpen_ = true;
@@ -77,44 +74,26 @@ public:
bool
isOpen() override
{
std::shared_lock lock(mutex_);
return isOpen_;
}
void
close() override
{
DataStore old;
{
std::unique_lock lock(mutex_);
isOpen_ = false;
old.swap(table_); // O(1) swap; release lock before destructor runs
}
// 'old' is now destroyed outside the lock — no fetch() can be
// blocked by the (potentially millions-of-entries) map destructor.
}
static bool
nullMode()
{
static bool const v = [] {
char const* e = std::getenv("XAHAU_RWDB_NULL");
return e && *e && std::string_view{e} != "0";
}();
return v;
std::lock_guard lock(mutex_);
table_.clear();
isOpen_ = false;
}
Status
fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
{
if (nullMode())
if (!isOpen_)
return notFound;
uint256 const hash(uint256::fromVoid(key));
std::shared_lock lock(mutex_);
if (!isOpen_)
return notFound;
std::lock_guard lock(mutex_);
auto it = table_.find(hash);
if (it == table_.end())
return notFound;
@@ -155,17 +134,6 @@ public:
if (!object)
return;
if (nullMode())
return;
static bool const discardHotAccountNode = [] {
char const* v = std::getenv("XAHAU_RWDB_DISCARD_HOT_ACCOUNT_NODE");
return v && *v && std::string_view{v} != "0";
}();
if (discardHotAccountNode && object->getType() == hotACCOUNT_NODE)
return;
EncodedBlob encoded(object);
nudb::detail::buffer bf;
auto const result =
@@ -194,9 +162,10 @@ public:
void
for_each(std::function<void(std::shared_ptr<NodeObject>)> f) override
{
std::shared_lock lock(mutex_);
if (!isOpen_)
return;
std::lock_guard lock(mutex_);
for (const auto& entry : table_)
{
nudb::detail::buffer bf;

View File

@@ -44,21 +44,6 @@ DatabaseRotatingImp::DatabaseRotatingImp(
fdRequired_ += archiveBackend_->fdRequired();
}
void
DatabaseRotatingImp::copyArchiveTo(Backend& dest)
{
// Snapshot the archive backend pointer under lock, then iterate it
// outside the lock. dest is not yet shared so its store() calls are
// uncontested — no live-backend write-lock contention.
auto archive = [&] {
std::lock_guard const lock(mutex_);
return archiveBackend_;
}();
archive->for_each(
[&](std::shared_ptr<NodeObject> obj) { dest.store(obj); });
}
void
DatabaseRotatingImp::rotate(
std::unique_ptr<NodeStore::Backend>&& newBackend,
@@ -126,11 +111,8 @@ DatabaseRotatingImp::rotate(
// Execute the lambda
ensurePinnedLedgersInWritable();
// Do NOT call setDeletePath() inside this lock. For in-memory
// backends, setDeletePath() calls close() which destructs the entire
// table_ map (millions of shared_ptr<NodeObject> ref-count decrements)
// while the lock is held, blocking every concurrent fetchNodeObject
// call for several seconds and starving consensus reads.
// Now it's safe to mark the archive backend for deletion
archiveBackend_->setDeletePath();
oldArchiveBackend = std::move(archiveBackend_);
// Complete the rotation
@@ -140,9 +122,6 @@ DatabaseRotatingImp::rotate(
writableBackend_ = std::move(newBackend);
}
// Lock released — clear the old archive now without blocking fetches.
oldArchiveBackend->setDeletePath();
f(newWritableBackendName, newArchiveBackendName);
}

View File

@@ -51,9 +51,6 @@ public:
stop();
}
void
copyArchiveTo(Backend& dest) override;
void
rotate(
std::unique_ptr<NodeStore::Backend>&& newBackend,

View File

@@ -32,14 +32,11 @@
#include <xrpld/overlay/detail/PeerImp.h>
#include <xrpld/overlay/detail/Tuning.h>
#include <xrpld/perflog/PerfLog.h>
#include <xrpld/shamap/Family.h>
#include <xrpld/shamap/SHAMapTreeNode.h>
#include <xrpl/basics/UptimeClock.h>
#include <xrpl/basics/base64.h>
#include <xrpl/basics/random.h>
#include <xrpl/basics/safe_cast.h>
#include <xrpl/beast/core/LexicalCast.h>
#include <xrpl/protocol/HashPrefix.h>
#include <xrpl/protocol/digest.h>
#include <boost/algorithm/string/predicate.hpp>
@@ -2466,53 +2463,14 @@ PeerImp::onMessage(std::shared_ptr<protocol::TMGetObjectByHash> const& m)
// VFALCO TODO Move this someplace more sensible so we dont
// need to inject the NodeStore interfaces.
std::uint32_t seq{obj.has_ledgerseq() ? obj.ledgerseq() : 0};
//@@start peerimp-node-fallback
auto nodeObject{app_.getNodeStore().fetchNodeObject(hash, seq)};
void const* dataPtr = nullptr;
std::size_t dataSize = 0;
Blob treeBlob;
if (nodeObject)
{
dataPtr = nodeObject->getData().data();
dataSize = nodeObject->getData().size();
}
else if (
auto treeNode =
app_.getNodeFamily().getTreeNodeCache()->fetch(hash))
{
// SHAMap tree node fallback — works for state/tx nodes
// held via the retained Ledgers' SHAMap inner nodes.
Serializer s;
treeNode->serializeWithPrefix(s);
treeBlob = std::move(s.modData());
dataPtr = treeBlob.data();
dataSize = treeBlob.size();
}
else if (packet.type() == protocol::TMGetObjectByHash::otLEDGER)
{
// Ledger header fallback — look up by hash in the
// in-memory ledger set and serialize the header in the
// same wire format used by the node store.
if (auto ledger =
app_.getLedgerMaster().getLedgerByHash(hash))
{
Serializer s(sizeof(LedgerInfo) + 4);
s.add32(HashPrefix::ledgerMaster);
addRaw(ledger->info(), s);
treeBlob = std::move(s.modData());
dataPtr = treeBlob.data();
dataSize = treeBlob.size();
}
}
//@@end peerimp-node-fallback
if (dataPtr)
{
protocol::TMIndexedObject& newObj = *reply.add_objects();
newObj.set_hash(hash.begin(), hash.size());
newObj.set_data(dataPtr, dataSize);
newObj.set_data(
&nodeObject->getData().front(),
nodeObject->getData().size());
if (obj.has_nodeid())
newObj.set_index(obj.nodeid());

View File

@@ -21,37 +21,8 @@
#include <xrpld/shamap/SHAMapSyncFilter.h>
#include <xrpl/basics/random.h>
#include <cstdlib>
#include <string_view>
namespace ripple {
namespace {
bool
isRWDBNullMode()
{
static bool const v = [] {
char const* e = std::getenv("XAHAU_RWDB_NULL");
return e && *e && std::string_view{e} != "0";
}();
return v;
}
bool
useFullBelowCache()
{
// FullBelowCache is enabled in both disk-backed and null modes. In
// null mode the FBC short-circuit sites (addKnownNode and
// gmn_ProcessNodes) additionally validate the claim via TreeNodeCache
// liveness and anchor the canonical into the current SHAMap's spine,
// so the claim cannot outlive the canonical node it vouches for. See
// .ai-docs/null-nodestore-backend.md for the full reasoning.
return true;
}
} // namespace
void
SHAMap::visitLeaves(
std::function<void(boost::intrusive_ptr<SHAMapItem const> const&
@@ -218,37 +189,10 @@ SHAMap::gmn_ProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
{
// we already know this child node is missing
fullBelow = false;
continue;
}
//@@start gmn-fullbelow-check
if (backed_ && useFullBelowCache() &&
f_.getFullBelowCache()->touch_if_exists(childHash.as_uint256()))
{
// Disk-backed mode: trust the claim (self-healing via lazy DB
// refetch on later reads).
if (!isRWDBNullMode())
continue;
// Null mode: validate via TreeNodeCache liveness AND the
// canonical's own full-below flag, then anchor the canonical
// into THIS SHAMap's spine. Same reasoning as addKnownNode.
if (auto canonical =
f_.getTreeNodeCache()->fetch(childHash.as_uint256());
canonical && canonical->isInner() &&
static_cast<SHAMapInnerNode*>(canonical.get())
->isFullBelow(mn.generation_))
{
node->canonicalizeChild(branch, std::move(canonical));
continue;
}
// fetch() returned null (canonical gone) OR the fetched
// canonical isn't marked full-below in the current
// generation (it's fresh — built from wire bytes with empty
// children_). Fall through to descend and walk properly.
}
//@@end gmn-fullbelow-check
else if (
!backed_ ||
!f_.getFullBelowCache()->touch_if_exists(childHash.as_uint256()))
{
bool pending = false;
auto d = descendAsync(
@@ -284,9 +228,7 @@ SHAMap::gmn_ProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
}
else if (
d->isInner() &&
(!useFullBelowCache() ||
!static_cast<SHAMapInnerNode*>(d)->isFullBelow(
mn.generation_)))
!static_cast<SHAMapInnerNode*>(d)->isFullBelow(mn.generation_))
{
mn.stack_.push(se);
@@ -306,7 +248,7 @@ SHAMap::gmn_ProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
if (fullBelow)
{ // No partial node encountered below this node
node->setFullBelowGen(mn.generation_);
if (backed_ && useFullBelowCache())
if (backed_)
{
f_.getFullBelowCache()->insert(node->getHash().as_uint256());
}
@@ -384,9 +326,8 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter* filter)
f_.getFullBelowCache()->getGeneration());
if (!root_->isInner() ||
(useFullBelowCache() &&
std::static_pointer_cast<SHAMapInnerNode>(root_)->isFullBelow(
mn.generation_)))
std::static_pointer_cast<SHAMapInnerNode>(root_)->isFullBelow(
mn.generation_))
{
clearSynching();
return std::move(mn.missingNodes_);
@@ -456,8 +397,7 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter* filter)
{
// Recheck nodes we could not finish before
for (auto const& [innerNode, nodeId] : mn.resumes_)
if (!useFullBelowCache() ||
!innerNode->isFullBelow(mn.generation_))
if (!innerNode->isFullBelow(mn.generation_))
mn.stack_.push(std::make_tuple(
innerNode, nodeId, rand_int(255), 0, true));
@@ -652,8 +592,7 @@ SHAMap::addKnownNode(
auto iNode = root_.get();
while (iNode->isInner() &&
(!useFullBelowCache() ||
!static_cast<SHAMapInnerNode*>(iNode)->isFullBelow(generation)) &&
!static_cast<SHAMapInnerNode*>(iNode)->isFullBelow(generation) &&
(iNodeID.getDepth() < node.getDepth()))
{
int branch = selectBranch(iNodeID, node.getNodeID());
@@ -666,65 +605,10 @@ SHAMap::addKnownNode(
}
auto childHash = inner->getChildHash(branch);
//@@start fullbelow-short-circuit
if (useFullBelowCache() &&
f_.getFullBelowCache()->touch_if_exists(childHash.as_uint256()))
if (f_.getFullBelowCache()->touch_if_exists(childHash.as_uint256()))
{
// Disk-backed mode: the FBC claim is self-healing (stale
// entries surface as lazy DB refetches). Return duplicate
// without further work.
if (!isRWDBNullMode())
return SHAMapAddNode::duplicate();
// Null mode: no DB to fall back on. Before trusting the FBC
// claim we verify two things about the canonical at this
// hash:
//
// 1. Liveness — TreeNodeCache::fetch returns non-null.
// Proves SOME shared_ptr to the canonical exists
// somewhere, so anchoring it via canonicalizeChild
// makes retention structural (not dependent on
// whichever ledger originally marked the subtree
// full-below).
//
// 2. Fully-walked — the canonical's own fullBelowGen_
// matches the current FBC generation. This is a
// strictly stronger property than liveness: it is
// only set by gmn_ProcessNodes AFTER successfully
// descending every child, which means children_[i]
// are populated.
//
// Why both matter: the FBC claim is tied to a hash, not to
// a specific canonical object. If the canonical that
// established the claim dies (last holder retires) and a
// fresh one is later materialised from wire bytes (e.g.
// via addKnownNode's `iNode == nullptr` branch or
// descend → filter), the fresh canonical has
// fullBelowGen_ == 0 and empty children_[i]. Liveness
// alone would pass, and we'd short-circuit onto an empty
// subtree; subsequent reads through unwired branches would
// then throw SHAMapMissingNode. The fullBelowGen_ check
// rejects fresh canonicals and forces descent, which
// populates children_ as it walks.
//
// In null mode the FBC generation is stable (no clear()
// calls — we removed rotation), so a canonical walked at
// any point since process start remains full-below for
// its lifetime.
if (auto canonical =
f_.getTreeNodeCache()->fetch(childHash.as_uint256());
canonical && canonical->isInner() &&
static_cast<SHAMapInnerNode*>(canonical.get())
->isFullBelow(generation))
{
inner->canonicalizeChild(branch, std::move(canonical));
return SHAMapAddNode::duplicate();
}
// Either no canonical, or canonical is fresh (not walked).
// Fall through to normal descent, which will populate this
// canonical's children_ as we walk toward the target.
return SHAMapAddNode::duplicate();
}
//@@end fullbelow-short-circuit
auto prevNode = inner;
std::tie(iNode, iNodeID) = descend(inner, iNodeID, branch, filter);
@@ -760,7 +644,6 @@ SHAMap::addKnownNode(
return SHAMapAddNode::useful();
}
//@@start addknown-hook-seq
if (backed_)
canonicalize(childHash, newNode);
@@ -777,7 +660,6 @@ SHAMap::addKnownNode(
std::move(s.modData()),
newNode->getType());
}
//@@end addknown-hook-seq
return SHAMapAddNode::useful();
}