From 5f087f64464b8cb84cd9b268ad2bf6e2c3d20ade Mon Sep 17 00:00:00 2001 From: Ed Hennis Date: Thu, 23 Jul 2026 21:07:35 -0400 Subject: [PATCH] Improve node store rotation - 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. --- include/xrpl/nodestore/DatabaseRotating.h | 15 +- .../nodestore/detail/DatabaseRotatingImp.h | 20 ++- src/libxrpl/nodestore/DatabaseRotatingImp.cpp | 46 +++-- src/xrpld/app/misc/SHAMapStoreImp.cpp | 159 ++++++++++++------ src/xrpld/app/misc/SHAMapStoreImp.h | 17 +- src/xrpld/app/rdb/backend/detail/Node.cpp | 4 +- 6 files changed, 184 insertions(+), 77 deletions(-) diff --git a/include/xrpl/nodestore/DatabaseRotating.h b/include/xrpl/nodestore/DatabaseRotating.h index 5381b5c435..5872e553d6 100644 --- a/include/xrpl/nodestore/DatabaseRotating.h +++ b/include/xrpl/nodestore/DatabaseRotating.h @@ -4,6 +4,7 @@ #include #include #include +#include #include #include @@ -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 diff --git a/include/xrpl/nodestore/detail/DatabaseRotatingImp.h b/include/xrpl/nodestore/detail/DatabaseRotatingImp.h index ecbe9a513d..2bec58c674 100644 --- a/include/xrpl/nodestore/detail/DatabaseRotatingImp.h +++ b/include/xrpl/nodestore/detail/DatabaseRotatingImp.h @@ -8,6 +8,7 @@ #include #include #include +#include #include #include @@ -71,20 +72,27 @@ public: sweep() override; void - setRotationInFlight(bool inFlight) override; + setRotationInFlight(LedgerIndex inFlight) override; + LedgerIndex + getRotationInFlight() const override; private: std::shared_ptr writableBackend_; std::shared_ptr 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 rotationInFlight_{false}; + // copyRejectCount_ tallies the ones that weren't copied. + std::atomic rotationInFlight_{0}; std::atomic copyForwardCount_{0}; + std::atomic copyRejectCount_{0}; std::shared_ptr fetchNodeObject(uint256 const& hash, std::uint32_t, FetchReport& fetchReport, bool duplicate) diff --git a/src/libxrpl/nodestore/DatabaseRotatingImp.cpp b/src/libxrpl/nodestore/DatabaseRotatingImp.cpp index 23a48a3bf3..0a5850d52b 100644 --- a/src/libxrpl/nodestore/DatabaseRotatingImp.cpp +++ b/src/libxrpl/nodestore/DatabaseRotatingImp.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include @@ -54,6 +55,7 @@ DatabaseRotatingImp::rotate( // deleted. std::shared_ptr 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 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); + } } } diff --git a/src/xrpld/app/misc/SHAMapStoreImp.cpp b/src/xrpld/app/misc/SHAMapStoreImp.cpp index bfe86a9205..6e332a55e9 100644 --- a/src/xrpld/app/misc/SHAMapStoreImp.cpp +++ b/src/xrpld/app/misc/SHAMapStoreImp.cpp @@ -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(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(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(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 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 { + 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(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; diff --git a/src/xrpld/app/misc/SHAMapStoreImp.h b/src/xrpld/app/misc/SHAMapStoreImp.h index e02f64f56f..9090aadd63 100644 --- a/src/xrpld/app/misc/SHAMapStoreImp.h +++ b/src/xrpld/app/misc/SHAMapStoreImp.h @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -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) + { + if (!obj) + { + auto const node = cache.fetch(key); + if (node) + rescueNode(*node); + } + } if (!(++check % checkHealthInterval_) && healthWait() == HealthResult::Stopping) return true; } diff --git a/src/xrpld/app/rdb/backend/detail/Node.cpp b/src/xrpld/app/rdb/backend/detail/Node.cpp index b2f14c71ea..d97fbce6f7 100644 --- a/src/xrpld/app/rdb/backend/detail/Node.cpp +++ b/src/xrpld/app/rdb/backend/detail/Node.cpp @@ -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.