ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-07-24 07:30:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve node store rotation

Browse Source 
- Add a special case to healthWait() to not pause when the server is
  DISCONNECTED.
- Log sequence differences at the start and end of rotation.
- Limit copy-forward to ledgers we're not about to delete, and unknown.
  (Changes rotationInFlight_ from a bool to a LedgerIndex.)
- Set the "rotation in flight" index right at the beginning of the rotation
    - Because the node copy process can take a long time, other ledgers may
      get validated. Any reads for those ledgers have the potential to be
      served by the archive DB and thus lost, too. Why wait?
- Add an assertion suggested on @vlntb in #7763.
- Don't wait as long for ledgers that should be built soon.
- Rescue nodes from the tree node cache, too.
This commit is contained in:
Ed Hennis
2026-07-23 21:07:35 -04:00
parent a2446e8f99
commit 5f087f6446
6 changed files with 184 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
include/xrpl/nodestore/DatabaseRotating.h
View File

@@ -4,6 +4,7 @@
#include <xrpl/nodestore/Backend.h>
#include <xrpl/nodestore/Database.h>
#include <xrpl/nodestore/Scheduler.h>
#include <xrpl/protocol/Protocol.h>
#include <functional>
#include <memory>
@@ -45,15 +46,15 @@ public:
/**
* Marks an online-delete rotation as in progress (or completed).
*
* While in flight, a read served by the archive backend is copied
* forward into the writable backend even for ordinary
* (duplicate == false) fetches: the archive is about to be deleted,
* and a node body canonicalized into caches during the rotation
* window would otherwise survive only in RAM once the archive is
* dropped.
* While in flight, a read for ledgers after the inFlight value served by the archive backend is
* copied forward into the writable backend even for ordinary (duplicate == false) fetches: the
* archive is about to be deleted, and a node body canonicalized into caches during the rotation
* window would otherwise survive only in RAM once the archive is dropped.
*/
virtual void
setRotationInFlight(bool inFlight) = 0;
setRotationInFlight(LedgerIndex inFlight) = 0;
virtual LedgerIndex
getRotationInFlight() const = 0;
};
} // namespace xrpl::NodeStore

20
include/xrpl/nodestore/detail/DatabaseRotatingImp.h
View File

@@ -8,6 +8,7 @@
#include <xrpl/nodestore/DatabaseRotating.h>
#include <xrpl/nodestore/NodeObject.h>
#include <xrpl/nodestore/Scheduler.h>
#include <xrpl/protocol/Protocol.h>
#include <atomic>
#include <cstdint>
@@ -71,20 +72,27 @@ public:
sweep() override;
void
setRotationInFlight(bool inFlight) override;
setRotationInFlight(LedgerIndex inFlight) override;
LedgerIndex
getRotationInFlight() const override;
private:
std::shared_ptr<Backend> writableBackend_;
std::shared_ptr<Backend> archiveBackend_;
mutable std::mutex mutex_;
// True between SHAMapStore starting the cache-freshen phase and the
// completion of rotate(). While true, archive hits on ordinary
// (duplicate == false) fetches are copied forward into the writable
// backend; copyForwardCount_ tallies them per rotation for the
// Set to the index of the last rotated ledger between SHAMapStore
// starting the cache-freshen phase and the completion of rotate().
// While non-zero, archive hits on ordinary (duplicate == false)
// fetches are copied forward into the writable backend if they are
// for that ledger or later, since those are the ones we'll keep.
// To be safe, copy forward if the provided ledger index is 0.
// copyForwardCount_ tallies them per rotation for the
// summary line logged at swap.
std::atomic<bool> rotationInFlight_{false};
// copyRejectCount_ tallies the ones that weren't copied.
std::atomic<LedgerIndex> rotationInFlight_{0};
std::atomic<std::uint64_t> copyForwardCount_{0};
std::atomic<std::uint64_t> copyRejectCount_{0};
std::shared_ptr<NodeObject>
fetchNodeObject(uint256 const& hash, std::uint32_t, FetchReport& fetchReport, bool duplicate)

46
src/libxrpl/nodestore/DatabaseRotatingImp.cpp
View File

@@ -12,6 +12,7 @@
#include <xrpl/nodestore/NodeObject.h>
#include <xrpl/nodestore/Scheduler.h>
#include <xrpl/nodestore/Types.h>
#include <xrpl/protocol/Protocol.h>
#include <atomic>
#include <cstdint>
@@ -54,6 +55,7 @@ DatabaseRotatingImp::rotate(
// deleted.
std::shared_ptr<NodeStore::Backend> oldArchiveBackend;
std::uint64_t copyForwards = 0;
std::uint64_t copyRejects = 0;
{
std::scoped_lock const lock(mutex_);
@@ -66,24 +68,31 @@ DatabaseRotatingImp::rotate(
writableBackend_ = std::move(newBackend);
copyForwards = copyForwardCount_.exchange(0, std::memory_order_relaxed);
copyRejects = copyRejectCount_.exchange(0, std::memory_order_relaxed);
}
if (copyForwards > 0)
if (copyForwards > 0 || copyRejects > 0)
{
JLOG(j_.warn()) << "Rotating: copied forward " << copyForwards
<< " archive-served reads into the writable backend "
"during the rotation window";
"during the rotation window. Rejected "
<< copyRejects;
}
f(newWritableBackendName, newArchiveBackendName);
}
void
DatabaseRotatingImp::setRotationInFlight(bool inFlight)
DatabaseRotatingImp::setRotationInFlight(LedgerIndex inFlight)
{
rotationInFlight_.store(inFlight, std::memory_order_release);
JLOG(j_.debug()) << "Rotating: copy-forward on archive reads "
<< (inFlight ? "enabled" : "disabled");
JLOG(j_.debug()) << "Rotating: copy-forward on archive reads from " << inFlight << " forward";
}
LedgerIndex
DatabaseRotatingImp::getRotationInFlight() const
{
return rotationInFlight_.load(std::memory_order_acquire);
}
std::string
@@ -141,7 +150,7 @@ DatabaseRotatingImp::sweep()
std::shared_ptr<NodeObject>
DatabaseRotatingImp::fetchNodeObject(
uint256 const& hash,
std::uint32_t,
std::uint32_t ledgerSeq,
FetchReport& fetchReport,
bool duplicate)
{
@@ -190,24 +199,35 @@ DatabaseRotatingImp::fetchNodeObject(
nodeObject = fetch(archive);
if (nodeObject)
{
{
// Refresh the writable backend pointer
std::scoped_lock const lock(mutex_);
writable = writableBackend_;
}
// Update writable backend with data from the archive backend.
// While a rotation is in flight, ordinary (duplicate == false)
// reads served by the archive are copied forward too: the
// archive is about to be deleted, and a body canonicalized
// into the cache after the freshen getKeys() snapshot would
// otherwise survive only in RAM once the archive is dropped.
if (duplicate || rotationInFlight_.load(std::memory_order_acquire))
auto const inFlight = getRotationInFlight();
if (duplicate || (inFlight != 0 && (ledgerSeq == 0 || ledgerSeq >= inFlight)))
{
{
// Refresh the writable backend pointer
std::scoped_lock const lock(mutex_);
writable = writableBackend_;
}
if (!duplicate)
{
JLOG(j_.debug()) << "Rotating: copy node for ledger " << ledgerSeq
<< " from archive to writable backend: " << hash;
copyForwardCount_.fetch_add(1, std::memory_order_relaxed);
}
writable->store(nodeObject);
}
else if (inFlight != 0)
{
JLOG(j_.debug()) << "Rotating: DO NOT copy node for ledger " << ledgerSeq
<< " from archive to writable backend: " << hash;
copyRejectCount_.fetch_add(1, std::memory_order_relaxed);
}
}
}

159
src/xrpld/app/misc/SHAMapStoreImp.cpp
View File

@@ -262,6 +262,45 @@ SHAMapStoreImp::fdRequired() const
return fdRequired_;
}
void
SHAMapStoreImp::rescueNode(SHAMapTreeNode const& node)
{
XRPL_ASSERT(node.cowid() == 0, "SHAMapStoreImp::copyNode : rescued node must be clean");
// Reachable from the validated state map in memory, but present in
// neither backend: its only on-disk copy lived in a backend removed by
// an earlier rotation, and it was never rewritten because it is clean
// (cowid == 0, so flushDirty skips it). Persist the in-memory body
// directly into the writable backend so it survives this rotation
// instead of later surfacing as an unresolvable SHAMapMissingNode.
auto const nodeType = node.getType();
auto const objectType = std::invoke([nodeType] {
switch (nodeType)
{
case SHAMapNodeType::TnAccountState:
return NodeObjectType::AccountNode;
case SHAMapNodeType::TnTransactionNm:
return NodeObjectType::TransactionNode;
default:
return NodeObjectType::Unknown;
}
});
auto const hash = node.getHash().asUInt256();
if (objectType == NodeObjectType::Unknown)
{
JLOG(journal_.warn()) << "copyNode: unable to re-store node with unknown type, hash="
<< hash << " type=" << static_cast<int>(nodeType);
return;
}
Serializer s;
node.serializeWithPrefix(s);
dbRotating_->store(objectType, std::move(s.modData()), hash, 0);
JLOG(journal_.info()) << "copyNode: re-stored node missing from both backends, hash=" << hash
<< " type=" << static_cast<int>(nodeType);
}
bool
SHAMapStoreImp::copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node)
{
@@ -270,19 +309,7 @@ SHAMapStoreImp::copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node)
node.getHash().asUInt256(), 0, NodeStore::FetchType::Synchronous, true);
if (!obj)
{
XRPL_ASSERT(node.cowid() == 0, "SHAMapStoreImp::copyNode : rescued node must be clean");
// Reachable from the validated state map in memory, but present in
// neither backend: its only on-disk copy lived in a backend removed by
// an earlier rotation, and it was never rewritten because it is clean
// (cowid == 0, so flushDirty skips it). Persist the in-memory body
// directly into the writable backend so it survives this rotation
// instead of later surfacing as an unresolvable SHAMapMissingNode.
auto const hash = node.getHash().asUInt256();
Serializer s;
node.serializeWithPrefix(s);
dbRotating_->store(NodeObjectType::AccountNode, std::move(s.modData()), hash, 0);
JLOG(journal_.warn()) << "copyNode: re-stored node missing from both backends, hash="
<< hash << " type=" << static_cast<int>(node.getType());
rescueNode(node);
}
if ((++nodeCount % checkHealthInterval_) == 0u)
{
@@ -308,6 +335,11 @@ SHAMapStoreImp::run()
while (true)
{
XRPL_ASSERT(
dbRotating_->getRotationInFlight() == 0,
"SHAMapStoreImp::run : rotationInFlight_ must be zero "
"outside rotation window");
healthy_ = true;
std::shared_ptr<Ledger const> validatedLedger;
@@ -353,13 +385,33 @@ SHAMapStoreImp::run()
// will delete up to (not including) lastRotated
if (readyToRotate)
{
auto const diff = validatedSeq - lastRotated;
JLOG(journal_.warn()) << "rotating validatedSeq " << validatedSeq << " lastRotated "
<< lastRotated << " deleteInterval " << deleteInterval_
<< " canDelete_ " << canDelete_ << " state "
<< lastRotated << " diff " << diff << " deleteInterval "
<< deleteInterval_ << " canDelete_ " << canDelete_ << " state "
<< app_.getOPs().strOperatingMode(false) << " age "
<< ledgerMaster_->getValidatedLedgerAge().count()
<< "s. Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
// Close the getKeys()->swap exposure window: from here until
// rotate() completes, an ordinary read for new ledgers served by the archive is
// copied forward into the writable backend, so a node fetched
// from the doomed archive cannot be left RAM-only when the
// archive is deleted. RAII so the early returns below (and any
// exception) also clear the flag.
struct RotationExposureGuard
{
NodeStore::DatabaseRotating& db;
~RotationExposureGuard()
{
db.setRotationInFlight(0);
}
};
RotationExposureGuard const rotationExposureGuard{*dbRotating_};
// Anything before lastRotated is going to get deleted soon, so we don't care about
// moving it to the writable DB.
dbRotating_->setRotationInFlight(lastRotated);
clearPrior(lastRotated);
if (healthWait() == HealthResult::Stopping)
return;
@@ -387,23 +439,6 @@ SHAMapStoreImp::run()
JLOG(journal_.debug())
<< "copied ledger " << validatedSeq << " nodecount " << nodeCount;
// Close the getKeys()->swap exposure window: from here until
// rotate() completes, an ordinary read served by the archive is
// copied forward into the writable backend, so a node fetched
// from the doomed archive cannot be left RAM-only when the
// archive is deleted. RAII so the early returns below (and any
// exception) also clear the flag.
struct RotationExposureGuard
{
NodeStore::DatabaseRotating& db;
~RotationExposureGuard()
{
db.setRotationInFlight(false);
}
};
RotationExposureGuard const rotationExposureGuard{*dbRotating_};
dbRotating_->setRotationInFlight(true);
JLOG(journal_.debug()) << "freshening caches";
freshenCaches();
if (healthWait() == HealthResult::Stopping)
@@ -433,9 +468,14 @@ SHAMapStoreImp::run()
clearCaches(validatedSeq);
});
JLOG(journal_.warn()) << "finished rotation. validatedSeq: " << validatedSeq
<< ", lastRotated: " << lastRotated
<< ". Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
auto const currentValidatedSeq = ledgerMaster_->getValidLedgerIndex();
auto const processingDiff = currentValidatedSeq - validatedSeq;
JLOG(journal_.warn())
<< "finished rotation. validatedSeq: " << validatedSeq
<< ", lastRotated: " << lastRotated << " diff " << diff
<< ". Updated validated seq is " << currentValidatedSeq << ", " << processingDiff
<< " ledgers were validated during the rotation processs. Complete ledgers: "
<< ledgerMaster_->getCompleteLedgers();
}
}
}
@@ -630,8 +670,7 @@ SHAMapStoreImp::freshenCaches()
{
if (freshenCache(*treeNodeCache_))
return;
if (freshenCache(app_.getMasterTransaction().getCache()))
return;
freshenCache(app_.getMasterTransaction().getCache());
}
void
@@ -701,7 +740,6 @@ SHAMapStoreImp::healthWait()
numMissing =
lowerBound == 0 ? 0 : ledgerMaster_->missingFromCompleteLedgerRange(lowerBound, index);
};
// Tracked server status properties
LedgerIndex index = 0;
std::chrono::seconds age;
@@ -720,7 +758,24 @@ SHAMapStoreImp::healthWait()
readServerStatus(index, age, mode, numMissing, lowerBound, unlock);
}
while (!stop_ && (mode != OperatingMode::FULL || age > ageThreshold || numMissing > 0))
auto healthy = [&]() {
// Special case: If the server is disconnected, it's not doing any ledger I/O, because
// it's focused on trying to get peers. A disconnected state is should never be caused by
// the activity of the server. It's usually limited to hardware or connectivity issues. Take
// advantage of that to run as much rotation I/O as possible before it comes back online.
if (mode == OperatingMode::DISCONNECTED)
return true;
if (age > ageThreshold)
return false;
if (numMissing > 0)
return false;
if (mode != OperatingMode::FULL)
return false;
return true;
};
while (!stop_ && !healthy())
{
// this value shouldn't change, so grab it while we have the
// lock
@@ -728,18 +783,26 @@ SHAMapStoreImp::healthWait()
ScopeUnlock const unlock(lock);
auto const stream = std::invoke([mode, age, ageThreshold, index, lastLedger, this]() {
if (mode != OperatingMode::FULL || age > ageThreshold)
return journal_.warn();
if (index != lastLedger)
return journal_.trace();
return journal_.info();
});
JLOG(stream) << "Waiting " << waitTime.count() << "s for node to stabilize. state: "
auto const [stream, waitMs] = std::invoke(
[mode, age, ageThreshold, index, lastLedger, waitTime, this]()
-> std::pair<beast::Journal::Stream, std::chrono::milliseconds> {
if (mode != OperatingMode::FULL || age > ageThreshold)
return {journal_.warn(), waitTime};
if (index != lastLedger)
{
// We expect this ledger to be built soon, so log at a lower level, and don't
// wait as long.
return {
journal_.trace(),
std::chrono::duration_cast<std::chrono::milliseconds>(waitTime) / 4};
}
return {journal_.info(), waitTime};
});
JLOG(stream) << "Waiting " << waitMs.count() << "ms for node to stabilize. state: "
<< app_.getOPs().strOperatingMode(mode, false) << ". age " << age.count()
<< "s. Missing ledgers: " << numMissing << ". Expect: " << lowerBound << "-"
<< index << ". Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
std::this_thread::sleep_for(waitTime);
std::this_thread::sleep_for(waitMs);
readServerStatus(index, age, mode, numMissing, lowerBound, unlock);
lastLedger = index;

17
src/xrpld/app/misc/SHAMapStoreImp.h
View File

@@ -21,6 +21,7 @@
#include <algorithm>
#include <atomic>
#include <chrono>
#include <concepts>
#include <condition_variable>
#include <cstdint>
#include <functional>
@@ -177,6 +178,9 @@ public:
minimumOnline() const override;
private:
// Force write a node to the writable backend during rotation so it doesn't get lost
void
rescueNode(SHAMapTreeNode const& node);
// callback for visitNodes
bool
copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node);
@@ -196,7 +200,18 @@ private:
for (auto const& key : cache.getKeys())
{
dbRotating_->fetchNodeObject(key, 0, NodeStore::FetchType::Synchronous, true);
[[maybe_unused]]
auto const obj =
dbRotating_->fetchNodeObject(key, 0, NodeStore::FetchType::Synchronous, true);
if constexpr (std::derived_from<typename CacheInstance::mapped_type, SHAMapTreeNode>)
{
if (!obj)
{
auto const node = cache.fetch(key);
if (node)
rescueNode(*node);
}
}
if (!(++check % checkHealthInterval_) && healthWait() == HealthResult::Stopping)
return true;
}

4
src/xrpld/app/rdb/backend/detail/Node.cpp
View File

@@ -268,9 +268,9 @@ saveValidatedLedger(
app.getAcceptedLedgerCache().canonicalizeReplaceClient(ledger->header().hash, aLedger);
}
}
catch (std::exception const&)
catch (std::exception const& e)
{
JLOG(j.warn()) << "An accepted ledger was missing nodes";
JLOG(j.warn()) << "An accepted ledger was missing nodes " << e.what();
app.getLedgerMaster().failedSave(seq, ledger->header().hash);
// Clients can now trust the database for information about this
// ledger sequence.