feat: generational NodeStore GC for online_delete (O(churn) rotation)

include/xrpl/config/Constants.h

@@ -131,6 +131,7 @@ struct Keys
     static constexpr auto kNormalConsensusIncreasePercent = "normal_consensus_increase_percent";
     static constexpr auto kNudbBlockSize = "nudb_block_size";
     static constexpr auto kOnlineDelete = "online_delete";
+    static constexpr auto kOnlineDeleteGenerations = "online_delete_generations";
     static constexpr auto kOpenFiles = "open_files";
     static constexpr auto kOptions = "options";
     static constexpr auto kOverlay = "overlay";

include/xrpl/nodestore/DatabaseRotating.h

@@ -5,20 +5,30 @@
 #include <xrpl/nodestore/Database.h>
 #include <xrpl/nodestore/Scheduler.h>
 
+#include <cstddef>
+#include <cstdint>
 #include <functional>
 #include <memory>
 #include <string>
+#include <vector>
 
 namespace xrpl::node_store {
 
-/* This class has two key-value store Backend objects for persisting SHAMap
- * records. This facilitates online deletion of data. New backends are
- * rotated in. Old ones are rotated out and deleted.
+/* This class keeps a ring of append-only key-value Backend objects (generations)
+ * for persisting SHAMap records, to facilitate online deletion of data. New nodes
+ * are written to the newest (writable) generation; reads probe newest -> oldest.
+ * Rather than copying the entire live state into a fresh backend every rotation
+ * (O(total state)), a generation is dropped only once its still-live nodes have been
+ * evacuated forward, so a cold node is re-stored ~once per ring cycle instead of every
+ * rotation (O(churn)).
  */
 
 class DatabaseRotating : public Database
 {
 public:
+    // Receives the full generation ring, ordered oldest -> newest, to persist durably.
+    using RingPersist = std::function<void(std::vector<std::string> const& generations)>;
+
     DatabaseRotating(
         Scheduler& scheduler,
         int readThreads,
@@ -29,31 +39,51 @@ public:
     }
 
     /**
-     * Rotates the backends.
+     * Append a fresh writable generation. The prior writable becomes a sealed,
+     * read-only generation that remains in the ring (still served by reads).
      *
-     * @param newBackend New writable backend
-     * @param f A function executed after the rotation outside of lock. The
-     * values passed to f will be the new backend database names _after_
-     * rotation.
+     * @param newWritable The new (empty) writable backend.
+     * @param persist Executed after the push, outside the lock, with the full ring
+     *        (oldest -> newest) so the caller can durably record it.
      */
     virtual void
-    rotate(
-        std::unique_ptr<node_store::Backend>&& newBackend,
-        std::function<void(std::string const& writableName, std::string const& archiveName)> const&
-            f) = 0;
+    advance(std::unique_ptr<Backend>&& newWritable, RingPersist const& persist) = 0;
 
     /**
-     * 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.
+     * Number of live generations currently in the ring.
+     */
+    virtual std::size_t
+    generationCount() const = 0;
+
+    /**
+     * Number of live nodes copied forward out of the retiring generation during the
+     * current retire window (reset by beginRetire). This is the evacuation volume — the
+     * churn the ring pays in place of copying the whole live state every rotation — so it
+     * quantifies that reclamation is O(churn) rather than O(total state).
+     */
+    virtual std::uint64_t
+    copyForwardCount() const = 0;
+
+    /**
+     * Begin/end retiring the oldest generation. While a retire is in progress, any
+     * read served by the retiring generation is copied forward into the writable
+     * backend (even ordinary reads): that generation is about to be dropped, so its
+     * still-live nodes must be preserved. Copy-forward is scoped to the retiring
+     * generation only — reads served by other sealed generations are NOT copied, which
+     * is what keeps evacuation O(churn) rather than O(total state).
      */
     virtual void
-    setRotationInFlight(bool inFlight) = 0;
+    beginRetire() = 0;
+    virtual void
+    endRetire() = 0;
+
+    /**
+     * Drop the oldest generation (its survivors already evacuated during the retire
+     * window). The generation's directory is deleted only after @p persist records the
+     * shortened ring, so a crash never leaves a persisted name without its backend.
+     */
+    virtual void
+    retireOldest(RingPersist const& persist) = 0;
 };
 
 }  // namespace xrpl::node_store

include/xrpl/nodestore/detail/DatabaseRotatingImp.h

@@ -10,11 +10,13 @@
 #include <xrpl/nodestore/Scheduler.h>
 
 #include <atomic>
+#include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <memory>
 #include <mutex>
 #include <string>
+#include <vector>
 
 namespace xrpl::node_store {
 
@@ -26,11 +28,11 @@ public:
     DatabaseRotatingImp&
     operator=(DatabaseRotatingImp const&) = delete;
 
+    // Generations are ordered oldest -> newest; the last is the writable backend.
     DatabaseRotatingImp(
         Scheduler& scheduler,
         int readThreads,
-        std::shared_ptr<Backend> writableBackend,
-        std::shared_ptr<Backend> archiveBackend,
+        std::vector<std::shared_ptr<Backend>> generations,
         Section const& config,
         beast::Journal j);
 
@@ -40,10 +42,22 @@ public:
     }
 
     void
-    rotate(
-        std::unique_ptr<node_store::Backend>&& newBackend,
-        std::function<void(std::string const& writableName, std::string const& archiveName)> const&
-            f) override;
+    advance(std::unique_ptr<Backend>&& newWritable, RingPersist const& persist) override;
+
+    std::size_t
+    generationCount() const override;
+
+    std::uint64_t
+    copyForwardCount() const override;
+
+    void
+    beginRetire() override;
+
+    void
+    endRetire() override;
+
+    void
+    retireOldest(RingPersist const& persist) override;
 
     std::string
     getName() const override;
@@ -70,20 +84,22 @@ public:
     void
     sweep() override;
 
-    void
-    setRotationInFlight(bool inFlight) override;
-
 private:
-    std::shared_ptr<Backend> writableBackend_;
-    std::shared_ptr<Backend> archiveBackend_;
+    // Immutable snapshot of the generation ring, ordered oldest (front) -> newest (back);
+    // back() is the writable backend. Replaced copy-on-write under mutex_ by advance() /
+    // retireOldest(); fetchNodeObject takes a shared_ptr copy and iterates it lock-free,
+    // so the hot read path never allocates and never blocks writers.
+    std::shared_ptr<std::vector<std::shared_ptr<Backend>>> ring_;
+    // The oldest generation while it is being retired (evacuated then dropped), else
+    // null. A read served by this generation is copied forward into the writable
+    // backend so its survivors are preserved before it is dropped; copy-forward is
+    // scoped to this generation only (reads from other sealed generations are not
+    // copied), which keeps evacuation O(churn) instead of O(total state).
+    std::shared_ptr<Backend> retiring_;
     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
-    // summary line logged at swap.
-    std::atomic<bool> rotationInFlight_{false};
+    // Tally of nodes copied forward out of the retiring generation, for the summary
+    // line logged when the generation is dropped.
     std::atomic<std::uint64_t> copyForwardCount_{0};
 
     std::shared_ptr<NodeObject>

include/xrpl/server/State.h

@@ -7,14 +7,30 @@
 #include <boost/filesystem.hpp>
 
 #include <string>
+#include <vector>
 
 namespace xrpl {
 
 struct SavedState
 {
+    // Legacy two-backend fields. Retained for on-disk back-compat: a state written
+    // by an older (two-backend) build has only these. On read, an empty `generations`
+    // is reconstructed as {archiveDb, writableDb} (oldest -> newest). On write, these
+    // are kept in sync with the ring ends (archiveDb = generations.front(),
+    // writableDb = generations.back()). CAUTION: a downgraded build boots from the
+    // pair alone — it deletes middle-generation directories as orphans (losing any
+    // node whose only copy lives there) and its rotations leave DbGenerations rows
+    // stale; getSavedState detects that staleness (ring ends disagreeing with the
+    // pair) and falls back to the pair.
     std::string writableDb;
     std::string archiveDb;
     LedgerIndex lastRotated{};
+
+    // The online_delete generation ring, ordered oldest -> newest. New nodes are
+    // written to generations.back() (the writable generation); reads probe
+    // newest -> oldest. Persisted as a newline-delimited list so a variable number
+    // of generations round-trips through the single-row state table.
+    std::vector<std::string> generations;
 };
 
 /**

src/libxrpl/nodestore/DatabaseRotatingImp.cpp

@@ -5,6 +5,7 @@
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/basics/contract.h>
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/config/BasicConfig.h>
 #include <xrpl/nodestore/Backend.h>
 #include <xrpl/nodestore/Database.h>
@@ -14,90 +15,167 @@
 #include <xrpl/nodestore/Types.h>
 
 #include <atomic>
+#include <cstddef>
 #include <cstdint>
 #include <exception>
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <ranges>
 #include <string>
 #include <utility>
+#include <vector>
 
 namespace xrpl::node_store {
 
+namespace {
+
+using Ring = std::vector<std::shared_ptr<Backend>>;
+
+// Names of the ring, ordered oldest -> newest.
+std::vector<std::string>
+ringNames(Ring const& ring)
+{
+    std::vector<std::string> names;
+    names.reserve(ring.size());
+    for (auto const& backend : ring)
+        names.push_back(backend->getName());
+    return names;
+}
+
+}  // namespace
+
 DatabaseRotatingImp::DatabaseRotatingImp(
     Scheduler& scheduler,
     int readThreads,
-    std::shared_ptr<Backend> writableBackend,
-    std::shared_ptr<Backend> archiveBackend,
+    std::vector<std::shared_ptr<Backend>> generations,
     Section const& config,
     beast::Journal j)
     : DatabaseRotating(scheduler, readThreads, config, j)
-    , writableBackend_(std::move(writableBackend))
-    , archiveBackend_(std::move(archiveBackend))
+    , ring_(std::make_shared<Ring>(std::move(generations)))
 {
-    if (writableBackend_)
-        fdRequired_ += writableBackend_->fdRequired();
-    if (archiveBackend_)
-        fdRequired_ += archiveBackend_->fdRequired();
+    XRPL_ASSERT(
+        !ring_->empty(), "xrpl::node_store::DatabaseRotatingImp : non-empty generation ring");
+    for (auto const& backend : *ring_)
+    {
+        if (backend)
+            fdRequired_ += backend->fdRequired();
+    }
 }
 
 void
-DatabaseRotatingImp::rotate(
-    std::unique_ptr<node_store::Backend>&& newBackend,
-    std::function<void(std::string const& writableName, std::string const& archiveName)> const& f)
+DatabaseRotatingImp::advance(std::unique_ptr<Backend>&& newWritable, RingPersist const& persist)
 {
-    // Pass these two names to the callback function
-    std::string const newWritableBackendName = newBackend->getName();
-    std::string newArchiveBackendName;
-    // Hold on to current archive backend pointer until after the
-    // callback finishes. Only then will the archive directory be
-    // deleted.
-    std::shared_ptr<node_store::Backend> oldArchiveBackend;
+    // Flush the outgoing writable's buffered writes before it is sealed read-only, so
+    // every node written to it is durable in its own generation and can't later surface
+    // as missing when a cold read falls through to it. Best-effort: a store racing this
+    // sync lands after it either way (stores write outside the lock), and such nodes are
+    // flushed by the backend's own cadence. Runs outside the lock so a slow flush never
+    // stalls concurrent fetches.
+    std::shared_ptr<Backend> outgoing;
+    {
+        std::scoped_lock const lock(mutex_);
+        if (!ring_->empty())
+            outgoing = ring_->back();
+    }
+    if (outgoing)
+        outgoing->sync();
+
+    std::vector<std::string> names;
+    {
+        std::scoped_lock const lock(mutex_);
+        // Copy-on-write: the prior writable stays in the ring as a sealed, read-only
+        // generation; the new backend becomes the writable one at the back.
+        auto next = std::make_shared<Ring>(*ring_);
+        next->push_back(std::shared_ptr<Backend>(std::move(newWritable)));
+        ring_ = std::move(next);
+        names = ringNames(*ring_);
+    }
+    persist(names);
+}
+
+std::size_t
+DatabaseRotatingImp::generationCount() const
+{
+    std::scoped_lock const lock(mutex_);
+    return ring_->size();
+}
+
+std::uint64_t
+DatabaseRotatingImp::copyForwardCount() const
+{
+    return copyForwardCount_.load(std::memory_order_relaxed);
+}
+
+void
+DatabaseRotatingImp::beginRetire()
+{
+    std::scoped_lock const lock(mutex_);
+    retiring_ = ring_->empty() ? nullptr : ring_->front();
+    copyForwardCount_.store(0, std::memory_order_relaxed);
+}
+
+void
+DatabaseRotatingImp::endRetire()
+{
+    std::scoped_lock const lock(mutex_);
+    retiring_.reset();
+}
+
+void
+DatabaseRotatingImp::retireOldest(RingPersist const& persist)
+{
+    // Keep the dropped generation alive until after persist so its directory is deleted
+    // only once the shortened ring is durably recorded (a crash never leaves a persisted
+    // name without its backend, nor a deleted backend still named in the ring).
+    std::shared_ptr<Backend> dropped;
+    std::vector<std::string> names;
     std::uint64_t copyForwards = 0;
     {
         std::scoped_lock const lock(mutex_);
-
-        archiveBackend_->setDeletePath();
-        oldArchiveBackend = std::move(archiveBackend_);
-
-        archiveBackend_ = std::move(writableBackend_);
-        newArchiveBackendName = archiveBackend_->getName();
-
-        writableBackend_ = std::move(newBackend);
-
+        // Never drop the writable generation.
+        if (ring_->size() <= 1)
+        {
+            retiring_.reset();
+            return;
+        }
+        dropped = ring_->front();
+        // Copy-on-write: publish a new ring with the oldest generation removed.
+        auto next = std::make_shared<Ring>(ring_->begin() + 1, ring_->end());
+        ring_ = std::move(next);
+        names = ringNames(*ring_);
         copyForwards = copyForwardCount_.exchange(0, std::memory_order_relaxed);
+        // Release the retiring reference to the dropped backend so `dropped` holds the
+        // last one and its directory is removed when this function returns.
+        if (retiring_ == dropped)
+            retiring_.reset();
     }
 
     if (copyForwards > 0)
     {
-        JLOG(j_.warn()) << "Rotating: copied forward " << copyForwards
-                        << " archive-served reads into the writable backend "
-                           "during the rotation window";
+        JLOG(j_.warn()) << "online_delete: evacuated " << copyForwards
+                        << " live nodes from the retired generation into the writable backend";
     }
 
-    f(newWritableBackendName, newArchiveBackendName);
-}
-
-void
-DatabaseRotatingImp::setRotationInFlight(bool inFlight)
-{
-    rotationInFlight_.store(inFlight, std::memory_order_release);
-    JLOG(j_.debug()) << "Rotating: copy-forward on archive reads "
-                     << (inFlight ? "enabled" : "disabled");
+    persist(names);
+    // Arm deletion only after the shortened ring is durably recorded: if persist throws,
+    // the directory survives for the name the state db still holds.
+    dropped->setDeletePath();
+    // `dropped` is destroyed here: its directory (armed by setDeletePath) is removed.
 }
 
 std::string
 DatabaseRotatingImp::getName() const
 {
     std::scoped_lock const lock(mutex_);
-    return writableBackend_->getName();
+    return ring_->empty() ? std::string() : ring_->back()->getName();
 }
 
 std::int32_t
 DatabaseRotatingImp::getWriteLoad() const
 {
     std::scoped_lock const lock(mutex_);
-    return writableBackend_->getWriteLoad();
+    return ring_->empty() ? 0 : ring_->back()->getWriteLoad();
 }
 
 void
@@ -105,17 +183,22 @@ DatabaseRotatingImp::importDatabase(Database& source)
 {
     auto const backend = [&] {
         std::scoped_lock const lock(mutex_);
-        return writableBackend_;
+        return ring_->empty() ? std::shared_ptr<Backend>() : ring_->back();
     }();
-
-    importInternal(*backend, source);
+    XRPL_ASSERT(backend, "xrpl::node_store::DatabaseRotatingImp::importDatabase : have writable");
+    if (backend)
+        importInternal(*backend, source);
 }
 
 void
 DatabaseRotatingImp::sync()
 {
-    std::scoped_lock const lock(mutex_);
-    writableBackend_->sync();
+    auto const backend = [&] {
+        std::scoped_lock const lock(mutex_);
+        return ring_->empty() ? std::shared_ptr<Backend>() : ring_->back();
+    }();
+    if (backend)
+        backend->sync();
 }
 
 void
@@ -125,8 +208,11 @@ DatabaseRotatingImp::store(NodeObjectType type, Blob&& data, uint256 const& hash
 
     auto const backend = [&] {
         std::scoped_lock const lock(mutex_);
-        return writableBackend_;
+        return ring_->empty() ? std::shared_ptr<Backend>() : ring_->back();
     }();
+    XRPL_ASSERT(backend, "xrpl::node_store::DatabaseRotatingImp::store : have writable");
+    if (!backend)
+        return;
 
     backend->store(nObj);
     storeStats(1, nObj->getData().size());
@@ -143,7 +229,7 @@ DatabaseRotatingImp::fetchNodeObject(
     uint256 const& hash,
     std::uint32_t,
     FetchReport& fetchReport,
-    bool duplicate)
+    bool)
 {
     auto fetch = [&](std::shared_ptr<Backend> const& backend) {
         Status status = Status::Ok;
@@ -174,41 +260,37 @@ DatabaseRotatingImp::fetchNodeObject(
         return nodeObject;
     };
 
-    // See if the node object exists in the cache
-    std::shared_ptr<NodeObject> nodeObject;
-
-    auto [writable, archive] = [&] {
-        std::scoped_lock const lock(mutex_);
-        return std::make_pair(writableBackend_, archiveBackend_);
-    }();
-
-    // Try to fetch from the writable backend
-    nodeObject = fetch(writable);
-    if (!nodeObject)
+    // Take a shared_ptr copy of the immutable ring snapshot (plus the retiring
+    // generation and writable backend) under the lock, then probe lock-free with no
+    // allocation.
+    std::shared_ptr<Ring const> ring;
+    std::shared_ptr<Backend> retiring;
+    std::shared_ptr<Backend> writable;
     {
-        // Otherwise try to fetch from the archive backend
-        nodeObject = fetch(archive);
-        if (nodeObject)
-        {
-            {
-                // Refresh the writable backend pointer
-                std::scoped_lock const lock(mutex_);
-                writable = writableBackend_;
-            }
+        std::scoped_lock const lock(mutex_);
+        ring = ring_;
+        retiring = retiring_;
+        writable = ring_->empty() ? nullptr : ring_->back();
+    }
 
-            // 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))
-            {
-                if (!duplicate)
-                    copyForwardCount_.fetch_add(1, std::memory_order_relaxed);
-                writable->store(nodeObject);
-            }
+    // Probe newest -> oldest; first hit wins.
+    std::shared_ptr<NodeObject> nodeObject;
+    for (auto const& backend : std::ranges::reverse_view(*ring))
+    {
+        nodeObject = fetch(backend);
+        if (!nodeObject)
+            continue;
+
+        // Copy forward only when the hit is served by the retiring generation: it is
+        // about to be dropped, so its still-live nodes must be preserved in the writable
+        // backend. Reads served by other sealed generations are deliberately NOT copied,
+        // which is what keeps evacuation O(churn) rather than O(total state).
+        if (retiring && backend == retiring && writable && backend != writable)
+        {
+            writable->store(nodeObject);
+            copyForwardCount_.fetch_add(1, std::memory_order_relaxed);
         }
+        break;
     }
 
     if (nodeObject)
@@ -220,16 +302,14 @@ DatabaseRotatingImp::fetchNodeObject(
 void
 DatabaseRotatingImp::forEach(std::function<void(std::shared_ptr<NodeObject>)> f)
 {
-    auto [writable, archive] = [&] {
+    std::shared_ptr<Ring const> ring;
+    {
         std::scoped_lock const lock(mutex_);
-        return std::make_pair(writableBackend_, archiveBackend_);
-    }();
+        ring = ring_;
+    }
 
-    // Iterate the writable backend
-    writable->forEach(f);
-
-    // Iterate the archive backend
-    archive->forEach(f);
+    for (auto const& backend : *ring)
+        backend->forEach(f);
 }
 
 }  // namespace xrpl::node_store

src/libxrpl/server/State.cpp

@@ -9,9 +9,12 @@
 
 #include <soci/boost-optional.h>  // IWYU pragma: keep
 #include <soci/into.h>
+#include <soci/rowset.h>
 #include <soci/session.h>
+#include <soci/transaction.h>
 #include <soci/use.h>
 
+#include <cstddef>
 #include <cstdint>
 #include <stdexcept>
 #include <string>
@@ -32,6 +35,16 @@ initStateDB(soci::session& session, BasicConfig const& config, std::string const
                "  LastRotatedLedger      INTEGER"
                ");";
 
+    // The online_delete generation ring, one row per generation ordered oldest ->
+    // newest by Ordinal. A dedicated table (rather than a delimited column) round-trips
+    // arbitrary backend directory names with no escaping and is rewritten atomically
+    // in a transaction by setSavedState. Absent/empty for an older two-backend state,
+    // in which case getSavedState reconstructs the ring from {ArchiveDb, WritableDb}.
+    session << "CREATE TABLE IF NOT EXISTS DbGenerations ("
+               "  Ordinal                INTEGER PRIMARY KEY,"
+               "  Name                   TEXT NOT NULL"
+               ");";
+
     session << "CREATE TABLE IF NOT EXISTS CanDelete ("
                "  Key                    INTEGER PRIMARY KEY,"
                "  CanDeleteSeq           INTEGER"
@@ -92,18 +105,62 @@ getSavedState(soci::session& session)
                " FROM DbState WHERE Key = 1;",
         soci::into(state.writableDb), soci::into(state.archiveDb), soci::into(state.lastRotated);
 
+    // Read the generation ring, oldest -> newest.
+    soci::rowset<std::string> const rs =
+        (session.prepare << "SELECT Name FROM DbGenerations ORDER BY Ordinal ASC;");
+    for (auto const& name : rs)
+        state.generations.push_back(name);
+
+    // Stale ring detection: setSavedState always writes the pair as the ring ends, so
+    // a mismatch means an older two-backend build rotated after this ring was written
+    // (it updates the pair but not DbGenerations, and deletes middle-generation
+    // directories as orphans). Trust the pair; the stale rows are rewritten on the
+    // next setSavedState.
+    if (!state.generations.empty() &&
+        (state.generations.back() != state.writableDb ||
+         state.generations.front() != state.archiveDb))
+    {
+        state.generations.clear();
+    }
+
+    // Legacy two-backend state (written before the generation ring existed, or
+    // invalidated above): no usable rows in DbGenerations but the pair is populated.
+    // Reconstruct the ring oldest -> newest so boot opens the same on-disk backends.
+    if (state.generations.empty() && !state.writableDb.empty())
+    {
+        if (!state.archiveDb.empty())
+            state.generations.push_back(state.archiveDb);
+        state.generations.push_back(state.writableDb);
+    }
+
     return state;
 }
 
 void
 setSavedState(soci::session& session, SavedState const& state)
 {
+    // Rewrite the state row and the whole generation ring atomically: a crash between
+    // the two would otherwise leave the persisted ring inconsistent with the pair,
+    // and on restart the node could open the wrong backends (missing nodes).
+    soci::transaction tr(session);
+
     session << "UPDATE DbState"
                " SET WritableDb = :writableDb,"
                " ArchiveDb = :archiveDb,"
                " LastRotatedLedger = :lastRotated"
                " WHERE Key = 1;",
         soci::use(state.writableDb), soci::use(state.archiveDb), soci::use(state.lastRotated);
+
+    session << "DELETE FROM DbGenerations;";
+    for (std::size_t i = 0; i < state.generations.size(); ++i)
+    {
+        auto const ordinal = static_cast<std::int64_t>(i);
+        auto const& name = state.generations[i];
+        session << "INSERT INTO DbGenerations (Ordinal, Name) VALUES (:ordinal, :name);",
+            soci::use(ordinal), soci::use(name);
+    }
+
+    tr.commit();
 }
 
 void

src/test/app/SHAMapStore_test.cpp

@@ -1,6 +1,7 @@
 #include <test/jtx/Env.h>
 #include <test/jtx/amount.h>
 #include <test/jtx/envconfig.h>
+#include <test/jtx/pay.h>
 
 #include <xrpld/app/main/Application.h>
 #include <xrpld/app/main/NodeStoreScheduler.h>
@@ -8,6 +9,7 @@
 #include <xrpld/app/rdb/backend/SQLiteDatabase.h>
 #include <xrpld/core/Config.h>
 
+#include <xrpl/basics/Blob.h>
 #include <xrpl/basics/ByteUtilities.h>
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/beast/unit_test/suite.h>
@@ -15,24 +17,35 @@
 #include <xrpl/config/Constants.h>
 #include <xrpl/json/json_value.h>
 #include <xrpl/nodestore/Backend.h>
+#include <xrpl/nodestore/DatabaseRotating.h>
 #include <xrpl/nodestore/Manager.h>
+#include <xrpl/nodestore/NodeObject.h>
+#include <xrpl/nodestore/Scheduler.h>
 #include <xrpl/nodestore/detail/DatabaseRotatingImp.h>
 #include <xrpl/protocol/ErrorCodes.h>
 #include <xrpl/protocol/LedgerHeader.h>
 #include <xrpl/protocol/Protocol.h>
 #include <xrpl/protocol/XRPAmount.h>
 #include <xrpl/protocol/jss.h>
+#include <xrpl/server/State.h>
+#include <xrpl/shamap/Family.h>
 
+#include <boost/filesystem/operations.hpp>
 #include <boost/filesystem/path.hpp>
 
+#include <soci/session.h>
+
 #include <atomic>
 #include <cstdint>
+#include <exception>
 #include <limits>
 #include <map>
 #include <memory>
 #include <optional>
 #include <string>
+#include <thread>
 #include <utility>
+#include <vector>
 
 namespace xrpl::test {
 
@@ -57,6 +70,46 @@ class SHAMapStore_test : public beast::unit_test::Suite
         return cfg;
     }
 
+    static auto
+    generationalDelete(std::unique_ptr<Config> cfg)
+    {
+        // The smallest budget: retirement runs on every rotation.
+        cfg = onlineDelete(std::move(cfg));
+        cfg->section(Sections::kNodeDatabase).set(Keys::kOnlineDeleteGenerations, "2");
+        return cfg;
+    }
+
+    // On-disk NuDB config rooted at fixed paths, so the ring and state survive an Env
+    // restart within a test.
+    static std::unique_ptr<Config>
+    diskConfig(std::string const& nodeDb, std::string const& stateDir, std::string const& budget)
+    {
+        return jtx::envconfig([&](std::unique_ptr<Config> cfg) {
+            cfg = onlineDelete(std::move(cfg));
+            auto& section = cfg->section(Sections::kNodeDatabase);
+            section.set(Keys::kType, "NuDB");
+            section.set(Keys::kPath, nodeDb);
+            section.set(Keys::kOnlineDeleteGenerations, budget);
+            cfg->legacy(Sections::kDatabasePath, stateDir);
+            return cfg;
+        });
+    }
+
+    // Close ledgers until the next rotation fires, asserting it landed where expected.
+    void
+    rotateOnce(jtx::Env& env, int& ledgerSeq)
+    {
+        auto& store = env.app().getSHAMapStore();
+        auto const target = store.getLastRotated() + kDeleteInterval;
+        while (ledgerSeq <= static_cast<int>(target))
+        {
+            env.close();
+            ++ledgerSeq;
+            store.rendezvous();
+        }
+        BEAST_EXPECT(store.getLastRotated() == target);
+    }
+
     static bool
     goodLedger(jtx::Env& env, json::Value const& json, std::string ledgerID, bool checkDB = false)
     {
@@ -543,17 +596,16 @@ public:
 
         NodeStoreScheduler scheduler(env.app().getJobQueue());
 
-        std::string const writableDb = "write";
-        std::string const archiveDb = "archive";
-        auto writableBackend = makeBackendRotating(env, scheduler, writableDb);
-        auto archiveBackend = makeBackendRotating(env, scheduler, archiveDb);
+        // Open a two-generation ring, oldest -> newest: {"archive", "write"}.
+        std::vector<std::shared_ptr<node_store::Backend>> generations;
+        generations.emplace_back(makeBackendRotating(env, scheduler, "archive"));
+        generations.emplace_back(makeBackendRotating(env, scheduler, "write"));
 
         static constexpr int kReadThreads = 4;
         auto dbr = std::make_unique<node_store::DatabaseRotatingImp>(
             scheduler,
             kReadThreads,
-            std::move(writableBackend),
-            std::move(archiveBackend),
+            std::move(generations),
             nscfg,
             env.app().getJournal("NodeStoreTest"));
 
@@ -562,46 +614,630 @@ public:
         using namespace std::chrono_literals;
         std::atomic<int> threadNum = 0;
 
+        BEAST_EXPECT(dbr->getName() == "write");
+        BEAST_EXPECT(dbr->generationCount() == 2);
+
+        // advance: append a fresh writable generation "1". The prior writable stays in
+        // the ring; the persist callback receives the whole ring, oldest -> newest.
         {
             auto newBackend = makeBackendRotating(env, scheduler, std::to_string(++threadNum));
-
-            auto const cb = [&](std::string const& writableName, std::string const& archiveName) {
-                BEAST_EXPECT(writableName == "1");
-                BEAST_EXPECT(archiveName == "write");
-                // Ensure that dbr functions can be called from within the
-                // callback
+            std::vector<std::string> persisted;
+            dbr->advance(std::move(newBackend), [&](std::vector<std::string> const& generations) {
+                persisted = generations;
+                // Ensure that dbr functions can be called from within the callback
                 BEAST_EXPECT(dbr->getName() == "1");
-            };
-
-            dbr->rotate(std::move(newBackend), cb);
+            });
+            BEAST_EXPECT((persisted == std::vector<std::string>{"archive", "write", "1"}));
         }
         BEAST_EXPECT(threadNum == 1);
         BEAST_EXPECT(dbr->getName() == "1");
+        BEAST_EXPECT(dbr->generationCount() == 3);
 
-        /////////////////////////////////////////////////////////////
-        // Do something stupid. Try to re-enter rotate from inside the callback.
+        // retire the oldest generation ("archive"); the ring shrinks and the persist
+        // callback receives the shortened ring.
         {
-            auto const cb = [&](std::string const& writableName, std::string const& archiveName) {
-                BEAST_EXPECT(writableName == "3");
-                BEAST_EXPECT(archiveName == "2");
-                // Ensure that dbr functions can be called from within the
-                // callback
-                BEAST_EXPECT(dbr->getName() == "3");
-            };
-            auto const cbReentrant = [&](std::string const& writableName,
-                                         std::string const& archiveName) {
-                BEAST_EXPECT(writableName == "2");
-                BEAST_EXPECT(archiveName == "1");
-                auto newBackend = makeBackendRotating(env, scheduler, std::to_string(++threadNum));
-                // Reminder: doing this is stupid and should never happen
-                dbr->rotate(std::move(newBackend), cb);
-            };
-            auto newBackend = makeBackendRotating(env, scheduler, std::to_string(++threadNum));
-            dbr->rotate(std::move(newBackend), cbReentrant);
+            dbr->beginRetire();
+            std::vector<std::string> persisted;
+            dbr->retireOldest([&](std::vector<std::string> const& generations) {
+                persisted = generations;
+                BEAST_EXPECT(dbr->getName() == "1");
+            });
+            dbr->endRetire();
+            BEAST_EXPECT((persisted == std::vector<std::string>{"write", "1"}));
+        }
+        BEAST_EXPECT(dbr->getName() == "1");
+        BEAST_EXPECT(dbr->generationCount() == 2);
+
+        // retireOldest never drops the sole writable generation.
+        {
+            dbr->retireOldest([&](std::vector<std::string> const& generations) {
+                BEAST_EXPECT((generations == std::vector<std::string>{"1"}));
+            });
+            BEAST_EXPECT(dbr->generationCount() == 1);
+
+            bool retiredWritable = false;
+            dbr->retireOldest([&](std::vector<std::string> const&) { retiredWritable = true; });
+            BEAST_EXPECT(!retiredWritable);
+            BEAST_EXPECT(dbr->generationCount() == 1);
+            BEAST_EXPECT(dbr->getName() == "1");
         }
 
+        /////////////////////////////////////////////////////////////
+        // Do something stupid. Re-enter advance from inside the persist callback.
+        {
+            auto const cbInner = [&](std::vector<std::string> const& generations) {
+                BEAST_EXPECT((generations == std::vector<std::string>{"1", "2", "3"}));
+                BEAST_EXPECT(dbr->getName() == "3");
+            };
+            auto const cbReentrant = [&](std::vector<std::string> const& generations) {
+                BEAST_EXPECT((generations == std::vector<std::string>{"1", "2"}));
+                auto newBackend = makeBackendRotating(env, scheduler, std::to_string(++threadNum));
+                // Reminder: doing this is stupid and should never happen
+                dbr->advance(std::move(newBackend), cbInner);
+            };
+            auto newBackend = makeBackendRotating(env, scheduler, std::to_string(++threadNum));
+            dbr->advance(std::move(newBackend), cbReentrant);
+        }
         BEAST_EXPECT(threadNum == 3);
         BEAST_EXPECT(dbr->getName() == "3");
+        BEAST_EXPECT(dbr->generationCount() == 3);
+
+        // Equally stupid: re-enter retireOldest from inside its persist callback.
+        {
+            bool innerRan = false;
+            dbr->beginRetire();
+            dbr->retireOldest([&](std::vector<std::string> const& generations) {
+                BEAST_EXPECT((generations == std::vector<std::string>{"2", "3"}));
+                dbr->retireOldest([&](std::vector<std::string> const& inner) {
+                    BEAST_EXPECT((inner == std::vector<std::string>{"3"}));
+                    innerRan = true;
+                });
+            });
+            dbr->endRetire();
+            BEAST_EXPECT(innerRan);
+            BEAST_EXPECT(dbr->generationCount() == 1);
+            BEAST_EXPECT(dbr->getName() == "3");
+        }
+    }
+
+    // Store a node into the writable generation and return its hash. A distinct tag byte
+    // gives each node a distinct key and payload.
+    static uint256
+    storeNode(node_store::DatabaseRotating& dbr, std::uint8_t tag)
+    {
+        uint256 hash;
+        hash.begin()[0] = tag;
+        Blob blob{tag, tag, tag};
+        dbr.store(NodeObjectType::AccountNode, std::move(blob), hash, 1);
+        return hash;
+    }
+
+    static bool
+    hasNode(node_store::DatabaseRotating& dbr, uint256 const& hash)
+    {
+        return dbr.fetchNodeObject(hash, 0, node_store::FetchType::Synchronous, false) != nullptr;
+    }
+
+    void
+    testRetention()
+    {
+        // Prove the generational invariant the whole feature rests on: retiring the oldest
+        // generation preserves its still-live nodes (evacuated forward) and reclaims only
+        // its dead ones, and evacuation is scoped to the retiring generation so nodes in
+        // other sealed generations are never needlessly copied. This is also the recovery
+        // guarantee — a node retained across a rotation is still fetchable afterwards.
+        testcase("generational retention and evacuation");
+
+        using namespace jtx;
+        Env env(*this, envconfig(onlineDelete));
+        NodeStoreScheduler scheduler(env.app().getJobQueue());
+        auto nscfg = env.app().config().section(Sections::kNodeDatabase);
+
+        auto const noop = [](std::vector<std::string> const&) {};
+
+        // Start with one generation (g0, writable) and grow the ring to three:
+        // g0 (oldest) -> g1 (middle) -> g2 (writable).
+        std::vector<std::shared_ptr<node_store::Backend>> generations;
+        generations.emplace_back(makeBackendRotating(env, scheduler, "g0"));
+        auto dbr = std::make_unique<node_store::DatabaseRotatingImp>(
+            scheduler, 4, std::move(generations), nscfg, env.app().getJournal("NodeStoreTest"));
+
+        // Into g0: a live node (X, will be evacuated) and a dead node (Z, never touched
+        // during the retire window, so it must be reclaimed with the generation).
+        auto const x = storeNode(*dbr, 0x11);
+        auto const z = storeNode(*dbr, 0x22);
+
+        dbr->advance(makeBackendRotating(env, scheduler, "g1"), noop);
+        // Into g1: a live node (Y) in a generation that will NOT be retired.
+        auto const y = storeNode(*dbr, 0x33);
+
+        dbr->advance(makeBackendRotating(env, scheduler, "g2"), noop);
+        BEAST_EXPECT(dbr->generationCount() == 3);
+        BEAST_EXPECT(dbr->getName() == "g2");
+
+        // Retire the oldest generation (g0). During the window, evacuate live nodes by
+        // fetching them — exactly what SHAMapStore does via visitNodes(copyNode).
+        dbr->beginRetire();
+
+        // X is served by the retiring generation, so it is copied forward into the
+        // writable backend.
+        BEAST_EXPECT(hasNode(*dbr, x));
+        BEAST_EXPECT(dbr->copyForwardCount() == 1);
+
+        // Y is served by a sealed but non-retiring generation: found, but NOT copied — the
+        // property that keeps evacuation O(churn) rather than O(total state).
+        BEAST_EXPECT(hasNode(*dbr, y));
+        BEAST_EXPECT(dbr->copyForwardCount() == 1);
+
+        // Fetching X again now hits the writable copy first, so it is not copied twice.
+        BEAST_EXPECT(hasNode(*dbr, x));
+        BEAST_EXPECT(dbr->copyForwardCount() == 1);
+
+        dbr->endRetire();
+        dbr->retireOldest(noop);
+        BEAST_EXPECT(dbr->generationCount() == 2);
+
+        // X lived only in g0; its survival proves it was evacuated to the writable backend.
+        BEAST_EXPECT(hasNode(*dbr, x));
+        // Z lived only in g0 and was never evacuated: reclaimed with the dropped generation.
+        BEAST_EXPECT(!hasNode(*dbr, z));
+        // Y lives in g1, which was not dropped: it survives without ever being copied.
+        BEAST_EXPECT(hasNode(*dbr, y));
+
+        // The rescue mechanism copyNode relies on: re-storing a lost node's in-memory
+        // body into the writable generation makes it fetchable again.
+        {
+            Blob blob{0x22, 0x22, 0x22};
+            dbr->store(NodeObjectType::AccountNode, std::move(blob), z, 0);
+            BEAST_EXPECT(hasNode(*dbr, z));
+        }
+
+        // Single-generation edge: with only the writable left, a retire window copies
+        // nothing forward (a writable hit is not an evacuation) and retireOldest
+        // refuses to drop the sole generation.
+        {
+            dbr->retireOldest(noop);
+            BEAST_EXPECT(dbr->generationCount() == 1);
+
+            dbr->beginRetire();
+            auto const w = storeNode(*dbr, 0x44);
+            BEAST_EXPECT(hasNode(*dbr, w));
+            BEAST_EXPECT(dbr->copyForwardCount() == 0);
+            dbr->retireOldest(noop);
+            BEAST_EXPECT(dbr->generationCount() == 1);
+            BEAST_EXPECT(hasNode(*dbr, w));
+            dbr->endRetire();
+        }
+    }
+
+    void
+    testConcurrentAccess()
+    {
+        // Race readers against the ring lifecycle: fetches snapshot the ring lock-free
+        // while the maintenance path advances, opens retire windows, and drops
+        // generations. Every node fetched during each retire window must survive.
+        testcase("concurrent fetch during advance and retire");
+
+        using namespace jtx;
+        Env env(*this, envconfig(onlineDelete));
+        NodeStoreScheduler scheduler(env.app().getJobQueue());
+        auto nscfg = env.app().config().section(Sections::kNodeDatabase);
+        auto const noop = [](std::vector<std::string> const&) {};
+
+        std::vector<std::shared_ptr<node_store::Backend>> generations;
+        generations.emplace_back(makeBackendRotating(env, scheduler, "c0"));
+        auto dbr = std::make_unique<node_store::DatabaseRotatingImp>(
+            scheduler, 4, std::move(generations), nscfg, env.app().getJournal("NodeStoreTest"));
+
+        std::vector<uint256> hashes;
+        for (std::uint8_t tag = 1; tag <= 16; ++tag)
+            hashes.push_back(storeNode(*dbr, tag));
+
+        std::atomic<bool> done{false};
+        std::vector<std::thread> readers;
+        readers.reserve(4);
+        for (int t = 0; t < 4; ++t)
+        {
+            readers.emplace_back([&] {
+                while (!done.load(std::memory_order_relaxed))
+                {
+                    for (auto const& h : hashes)
+                        hasNode(*dbr, h);
+                }
+            });
+        }
+        // A writer racing advance/retire: store() snapshots the writable outside the
+        // ring lock, the exact window the lifecycle mutates.
+        readers.emplace_back([&] {
+            std::uint8_t tag = 0;
+            while (!done.load(std::memory_order_relaxed))
+                storeNode(*dbr, 100 + (tag++ % 100));
+        });
+
+        for (int cycle = 1; cycle <= 20; ++cycle)
+        {
+            dbr->advance(makeBackendRotating(env, scheduler, "c" + std::to_string(cycle)), noop);
+            dbr->beginRetire();
+            // Evacuate by fetching, as SHAMapStore does.
+            for (auto const& h : hashes)
+                hasNode(*dbr, h);
+            dbr->retireOldest(noop);
+            dbr->endRetire();
+        }
+        done = true;
+        for (auto& r : readers)
+            r.join();
+
+        BEAST_EXPECT(dbr->generationCount() == 1);
+        for (auto const& h : hashes)
+            BEAST_EXPECT(hasNode(*dbr, h));
+    }
+
+    void
+    testStateMigration()
+    {
+        // The saved-state ring round-trips, reconstructs from a legacy two-backend
+        // state, and detects rows left stale by a downgraded build's rotation.
+        testcase("saved state ring migration");
+
+        using namespace jtx;
+        Env env(*this, envconfig(onlineDelete));
+
+        soci::session session;
+        initStateDB(session, env.app().config(), "state_migration_test");
+
+        // Legacy pair only (pre-ring build): the ring is reconstructed oldest -> newest.
+        session << "UPDATE DbState SET WritableDb = 'write', ArchiveDb = 'archive',"
+                   " LastRotatedLedger = 42 WHERE Key = 1;";
+        auto state = getSavedState(session);
+        BEAST_EXPECT((state.generations == std::vector<std::string>{"archive", "write"}));
+        BEAST_EXPECT(state.writableDb == "write");
+        BEAST_EXPECT(state.archiveDb == "archive");
+        BEAST_EXPECT(state.lastRotated == 42);
+
+        // A ring round-trips, with the legacy pair mirroring the ring ends.
+        SavedState ring;
+        ring.generations = {"g0", "g1", "g2"};
+        ring.archiveDb = "g0";
+        ring.writableDb = "g2";
+        ring.lastRotated = 43;
+        setSavedState(session, ring);
+        state = getSavedState(session);
+        BEAST_EXPECT((state.generations == std::vector<std::string>{"g0", "g1", "g2"}));
+        BEAST_EXPECT(state.archiveDb == "g0");
+        BEAST_EXPECT(state.writableDb == "g2");
+        BEAST_EXPECT(state.lastRotated == 43);
+
+        // Downgrade simulation: an old build's rotation rewrites the pair but leaves
+        // DbGenerations untouched. The stale ring is discarded in favor of the pair.
+        session << "UPDATE DbState SET WritableDb = 'new', ArchiveDb = 'g2' WHERE Key = 1;";
+        state = getSavedState(session);
+        BEAST_EXPECT((state.generations == std::vector<std::string>{"g2", "new"}));
+        BEAST_EXPECT(state.archiveDb == "g2");
+        BEAST_EXPECT(state.writableDb == "new");
+    }
+
+    void
+    testGenerationalHistory()
+    {
+        // End-to-end retirement through the SHAMapStore run loop with the smallest
+        // generation budget (2), with churn every ledger so superseded node versions
+        // exist. After several retirements, the full state of the oldest retained
+        // ledgers must still resolve — the retained-history guarantee.
+        testcase("retirement preserves retained history");
+
+        using namespace jtx;
+        Env env(*this, envconfig(generationalDelete));
+        auto& store = env.app().getSHAMapStore();
+
+        Account const alice{"alice"};
+        env.fund(XRP(10000), noripple(alice));
+
+        auto ledgerSeq = waitForReady(env);
+        LedgerIndex prevRotated = 0;
+
+        for (int cycle = 0; cycle < 3; ++cycle)
+        {
+            prevRotated = store.getLastRotated();
+            auto const target = prevRotated + kDeleteInterval;
+            while (ledgerSeq <= static_cast<int>(target))
+            {
+                env(pay(env.master, alice, XRP(1)));
+                env.close();
+
+                auto const ledger = env.rpc("ledger", "validated");
+                BEAST_EXPECT(goodLedger(env, ledger, std::to_string(ledgerSeq++)));
+                store.rendezvous();
+            }
+            BEAST_EXPECT(store.getLastRotated() == target);
+        }
+
+        // prevRotated is now the oldest retained ledger; versions its state references
+        // that were superseded during the last interval sat in the generation retired
+        // at the final rotation and survive only via boundary-root evacuation.
+        for (auto const seq : {prevRotated, store.getLastRotated()})
+        {
+            json::Value params;
+            params[jss::ledger_index] = seq;
+            params[jss::limit] = 4096;
+            auto const res = env.rpc("json", "ledger_data", params.toStyledString());
+            BEAST_EXPECT(res.isMember(jss::result) && !rpc::containsError(res[jss::result]));
+            BEAST_EXPECT(res[jss::result][jss::state].size() > 0);
+            // No marker: the whole state tree resolved in one page.
+            BEAST_EXPECT(!res[jss::result].isMember(jss::marker));
+        }
+    }
+
+    void
+    testTwoRootEvacuation()
+    {
+        // Deterministic proof of boundary-root evacuation. Emptying the TreeNodeCache
+        // right before each rotation-triggering close leaves cache freshening nothing
+        // to rescue, so the walk of the oldest retained ledger's root is the ONLY
+        // mechanism that can preserve node versions that ledger references but the
+        // current state no longer does (they live in the generation being retired).
+        testcase("boundary-root evacuation preserves superseded versions");
+
+        using namespace jtx;
+        Env env(*this, envconfig(generationalDelete));
+        auto& store = env.app().getSHAMapStore();
+
+        Account const alice{"alice"};
+        env.fund(XRP(10000), noripple(alice));
+
+        auto ledgerSeq = waitForReady(env);
+        LedgerIndex prevRotated = 0;
+
+        for (int cycle = 0; cycle < 2; ++cycle)
+        {
+            prevRotated = store.getLastRotated();
+            auto const target = prevRotated + kDeleteInterval;
+            while (ledgerSeq <= static_cast<int>(target))
+            {
+                // Churn every ledger: alice's account root (and the inner nodes above
+                // it) is superseded on every close.
+                env(pay(env.master, alice, XRP(1)));
+                if (ledgerSeq == static_cast<int>(target))
+                    env.app().getNodeFamily().getTreeNodeCache()->clear();
+                env.close();
+
+                auto const ledger = env.rpc("ledger", "validated");
+                BEAST_EXPECT(goodLedger(env, ledger, std::to_string(ledgerSeq++)));
+                store.rendezvous();
+            }
+            BEAST_EXPECT(store.getLastRotated() == target);
+        }
+
+        // The boundary ledger's state references versions written before the previous
+        // rotation and superseded since -- exactly the contents of the generation just
+        // retired. A complete walk proves they were evacuated via the boundary root.
+        json::Value params;
+        params[jss::ledger_index] = prevRotated;
+        params[jss::limit] = 4096;
+        auto const res = env.rpc("json", "ledger_data", params.toStyledString());
+        BEAST_EXPECT(res.isMember(jss::result) && !rpc::containsError(res[jss::result]));
+        BEAST_EXPECT(res[jss::result][jss::state].size() > 0);
+        BEAST_EXPECT(!res[jss::result].isMember(jss::marker));
+    }
+
+    void
+    testRingConvergence()
+    {
+        // A ring persisted over budget -- rotations interrupted between advance and
+        // retire, or a lowered online_delete_generations -- must converge back to
+        // budget on the next rotation instead of staying inflated forever. Phase 1
+        // grows a 4-generation ring under a budget of 4 (no retirement) on real disk
+        // backends; phase 2 reboots the same data with a budget of 2 and expects the
+        // first rotation to retire all the way back down.
+        testcase("ring converges to budget after over-budget boot");
+
+        using namespace jtx;
+        namespace bfs = boost::filesystem;
+
+        auto const root = bfs::temp_directory_path() / bfs::unique_path("shamapstore_conv_%%%%");
+        auto const nodeDb = (root / "nudb").string();
+        auto const stateDir = (root / "state").string();
+        bfs::create_directories(nodeDb);
+        bfs::create_directories(stateDir);
+
+        {
+            Env env(*this, diskConfig(nodeDb, stateDir, "4"));
+            auto ledgerSeq = waitForReady(env);
+
+            // Two rotations grow the ring 2 -> 3 -> 4, never exceeding the budget,
+            // so no generation is ever retired.
+            rotateOnce(env, ledgerSeq);
+            rotateOnce(env, ledgerSeq);
+
+            auto const& dbr = dynamic_cast<node_store::DatabaseRotating&>(env.app().getNodeStore());
+            BEAST_EXPECT(dbr.generationCount() == 4);
+        }
+
+        {
+            Env env(*this, diskConfig(nodeDb, stateDir, "2"));
+            auto& store = env.app().getSHAMapStore();
+            auto const& dbr = dynamic_cast<node_store::DatabaseRotating&>(env.app().getNodeStore());
+
+            // The persisted 4-generation ring reopened, now over a budget of 2.
+            BEAST_EXPECT(dbr.generationCount() == 4);
+            auto const bootRotated = store.getLastRotated();
+            BEAST_EXPECT(bootRotated != 0);
+
+            // The fresh genesis chain must catch up to bootRotated + interval before
+            // the next rotation fires.
+            for (int i = 0; i < static_cast<int>(bootRotated) + (2 * kDeleteInterval) &&
+                 store.getLastRotated() == bootRotated;
+                 ++i)
+            {
+                env.close();
+                store.rendezvous();
+            }
+            BEAST_EXPECT(store.getLastRotated() != bootRotated);
+
+            // One rotation: advance made it 5; the retire loop must shed 3, not 1.
+            BEAST_EXPECT(dbr.generationCount() == 2);
+        }
+
+        bfs::remove_all(root);
+    }
+
+    void
+    testMinGenerationsClamp()
+    {
+        // online_delete_generations below the floor is clamped to 2, preserving the
+        // writable + one archive invariant: a budget of 1 would drop the only sealed
+        // generation immediately after every rotation.
+        testcase("online_delete_generations clamps to the minimum");
+
+        using namespace jtx;
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg = onlineDelete(std::move(cfg));
+            cfg->section(Sections::kNodeDatabase).set(Keys::kOnlineDeleteGenerations, "1");
+            return cfg;
+        }));
+        auto const& dbr = dynamic_cast<node_store::DatabaseRotating&>(env.app().getNodeStore());
+
+        auto ledgerSeq = waitForReady(env);
+        rotateOnce(env, ledgerSeq);
+
+        // Clamped to 2: the rotation retired down to two generations, not one.
+        BEAST_EXPECT(dbr.generationCount() == 2);
+    }
+
+    void
+    testCorruptedStateRefusal()
+    {
+        // A generation named in the persisted ring but missing on disk means the data
+        // is unusable; the server must refuse to start rather than silently open a
+        // ring with a hole in it.
+        testcase("boot refuses a ring with a missing generation");
+
+        using namespace jtx;
+        namespace bfs = boost::filesystem;
+
+        auto const root = bfs::temp_directory_path() / bfs::unique_path("shamapstore_miss_%%%%");
+        auto const nodeDb = (root / "nudb").string();
+        auto const stateDir = (root / "state").string();
+        bfs::create_directories(nodeDb);
+        bfs::create_directories(stateDir);
+
+        {
+            Env env(*this, diskConfig(nodeDb, stateDir, "4"));
+            auto ledgerSeq = waitForReady(env);
+            rotateOnce(env, ledgerSeq);
+            auto const& dbr = dynamic_cast<node_store::DatabaseRotating&>(env.app().getNodeStore());
+            BEAST_EXPECT(dbr.generationCount() == 3);
+        }
+
+        std::vector<bfs::path> generations;
+        for (bfs::directory_iterator it(nodeDb); it != bfs::directory_iterator(); ++it)
+            generations.push_back(it->path());
+        BEAST_EXPECT(generations.size() == 3);
+        bfs::remove_all(generations.front());
+
+        bool threw = false;
+        try
+        {
+            Env const env(*this, diskConfig(nodeDb, stateDir, "4"));
+        }
+        catch (std::exception const&)
+        {
+            threw = true;
+        }
+        BEAST_EXPECT(threw);
+
+        bfs::remove_all(root);
+    }
+
+    void
+    testOrphanCleanup()
+    {
+        // A directory with the backend prefix that is not in the persisted ring is an
+        // orphan (created but never persisted before a crash) and is removed at boot;
+        // unrelated directories are left alone.
+        testcase("orphan generation directories are removed at boot");
+
+        using namespace jtx;
+        namespace bfs = boost::filesystem;
+
+        auto const root = bfs::temp_directory_path() / bfs::unique_path("shamapstore_orph_%%%%");
+        auto const nodeDb = (root / "nudb").string();
+        auto const stateDir = (root / "state").string();
+        bfs::create_directories(nodeDb);
+        bfs::create_directories(stateDir);
+
+        {
+            Env env(*this, diskConfig(nodeDb, stateDir, "4"));
+            auto ledgerSeq = waitForReady(env);
+            rotateOnce(env, ledgerSeq);
+        }
+
+        auto const orphan = bfs::path(nodeDb) / "rippledb.orphan";  // cspell: disable-line
+        auto const unrelated = bfs::path(nodeDb) / "unrelated";
+        bfs::create_directories(orphan);
+        bfs::create_directories(unrelated);
+
+        {
+            Env env(*this, diskConfig(nodeDb, stateDir, "4"));
+            auto const& dbr = dynamic_cast<node_store::DatabaseRotating&>(env.app().getNodeStore());
+            BEAST_EXPECT(dbr.generationCount() == 3);
+            BEAST_EXPECT(!bfs::exists(orphan));
+            BEAST_EXPECT(bfs::exists(unrelated));
+        }
+
+        bfs::remove_all(root);
+    }
+
+    void
+    testPathRelocation()
+    {
+        // When the configured node_db path changes, every stored generation name is
+        // rewritten to the new directory (keeping filenames) and the ring reopens.
+        testcase("ring survives a node_db path change");
+
+        using namespace jtx;
+        namespace bfs = boost::filesystem;
+
+        auto const root = bfs::temp_directory_path() / bfs::unique_path("shamapstore_relo_%%%%");
+        auto const nodeDbA = (root / "nudb_a").string();
+        auto const nodeDbB = (root / "nudb_b").string();
+        auto const stateDir = (root / "state").string();
+        bfs::create_directories(nodeDbA);
+        bfs::create_directories(nodeDbB);
+        bfs::create_directories(stateDir);
+
+        {
+            Env env(*this, diskConfig(nodeDbA, stateDir, "4"));
+            auto ledgerSeq = waitForReady(env);
+            rotateOnce(env, ledgerSeq);
+            auto const& dbr = dynamic_cast<node_store::DatabaseRotating&>(env.app().getNodeStore());
+            BEAST_EXPECT(dbr.generationCount() == 3);
+        }
+
+        // The operator moves the data and repoints the config.
+        for (bfs::directory_iterator it(nodeDbA); it != bfs::directory_iterator(); ++it)
+            bfs::rename(it->path(), bfs::path(nodeDbB) / it->path().filename());
+
+        {
+            Env env(*this, diskConfig(nodeDbB, stateDir, "4"));
+            auto& store = env.app().getSHAMapStore();
+            auto const& dbr = dynamic_cast<node_store::DatabaseRotating&>(env.app().getNodeStore());
+            BEAST_EXPECT(dbr.generationCount() == 3);
+
+            // Still operational: the relocated ring rotates normally.
+            auto const bootRotated = store.getLastRotated();
+            for (int i = 0; i < static_cast<int>(bootRotated) + (2 * kDeleteInterval) &&
+                 store.getLastRotated() == bootRotated;
+                 ++i)
+            {
+                env.close();
+                store.rendezvous();
+            }
+            BEAST_EXPECT(store.getLastRotated() != bootRotated);
+            BEAST_EXPECT(dbr.generationCount() == 4);
+        }
+
+        bfs::remove_all(root);
     }
 
     void
@@ -611,6 +1247,16 @@ public:
         testAutomatic();
         testCanDelete();
         testRotate();
+        testRetention();
+        testConcurrentAccess();
+        testStateMigration();
+        testGenerationalHistory();
+        testTwoRootEvacuation();
+        testRingConvergence();
+        testMinGenerationsClamp();
+        testCorruptedStateRefusal();
+        testOrphanCleanup();
+        testPathRelocation();
     }
 };
 

src/xrpld/app/misc/SHAMapStoreImp.cpp

@@ -14,6 +14,7 @@
 #include <xrpl/config/BasicConfig.h>
 #include <xrpl/config/Constants.h>
 #include <xrpl/ledger/Ledger.h>
+#include <xrpl/nodestore/Backend.h>
 #include <xrpl/nodestore/Database.h>
 #include <xrpl/nodestore/Manager.h>
 #include <xrpl/nodestore/NodeObject.h>
@@ -32,6 +33,7 @@
 #include <boost/filesystem/path.hpp>
 
 #include <algorithm>
+#include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <limits>
@@ -130,6 +132,11 @@ SHAMapStoreImp::SHAMapStoreImp(
     {
         // Configuration that affects the behavior of online delete
         getIfExists(section, Keys::kDeleteBatch, deleteBatch_);
+        getIfExists(section, Keys::kOnlineDeleteGenerations, numGenerations_);
+        // A ring needs at least a writable generation plus one archive (the historical
+        // two-backend behavior); fewer would drop data still referenced by the network.
+        // The upper bound caps file-descriptor and directory growth.
+        numGenerations_ = std::clamp(numGenerations_, kMinGenerations, kMaxGenerations);
         std::uint32_t temp = 0;
         if (getIfExists(section, Keys::kBackOffMilliseconds, temp) ||
             // Included for backward compatibility with an undocumented setting
@@ -190,26 +197,50 @@ SHAMapStoreImp::makeNodeStore(int readThreads)
     if (deleteInterval_ != 0u)
     {
         SavedState state = stateDb_.getState();
-        auto writableBackend = makeBackendRotating(state.writableDb);
-        auto archiveBackend = makeBackendRotating(state.archiveDb);
-        if (state.writableDb.empty())
+
+        // Open the generation ring, oldest -> newest. New nodes are written to the
+        // newest (writable) generation; reads probe newest -> oldest.
+        std::vector<std::shared_ptr<node_store::Backend>> generations;
+        if (state.generations.empty())
         {
-            state.writableDb = writableBackend->getName();
-            state.archiveDb = archiveBackend->getName();
+            // First run: bootstrap a two-generation ring (a sealed archive plus an
+            // empty writable), matching the historical initial two-backend state.
+            auto archive = makeBackendRotating();
+            auto writable = makeBackendRotating();
+            state.archiveDb = archive->getName();
+            state.writableDb = writable->getName();
+            state.generations = {archive->getName(), writable->getName()};
             stateDb_.setState(state);
+            generations.emplace_back(std::move(archive));
+            generations.emplace_back(std::move(writable));
+        }
+        else
+        {
+            for (auto const& name : state.generations)
+                generations.emplace_back(makeBackendRotating(name));
         }
 
-        // Create NodeStore with two backends to allow online deletion of
-        // data
+        // Create the rotating NodeStore over the generation ring to allow online
+        // deletion of data.
         auto dbr = std::make_unique<node_store::DatabaseRotatingImp>(
             scheduler_,
             readThreads,
-            std::move(writableBackend),
-            std::move(archiveBackend),
+            std::move(generations),
             nscfg,
             app_.getJournal(kNodeStoreName));
         fdRequired_ += dbr->fdRequired();
+        // The ring grows to numGenerations_ (+1 transiently between advance and retire),
+        // so budget descriptors for the full ring, not just the generations open at boot.
+        if (auto const bootCount = dbr->generationCount();
+            bootCount > 0 && numGenerations_ + 1 > bootCount)
+        {
+            fdRequired_ += static_cast<int>(dbr->fdRequired() / bootCount) *
+                static_cast<int>(numGenerations_ + 1 - bootCount);
+        }
         dbRotating_ = dbr.get();
+        JLOG(journal_.warn()) << "online_delete generation ring: opened " << dbr->generationCount()
+                              << " generations, budget " << numGenerations_
+                              << " (retire when above budget)";
         db.reset(dynamic_cast<node_store::Database*>(dbr.release()));
     }
     else
@@ -253,7 +284,10 @@ SHAMapStoreImp::fdRequired() const
 }
 
 bool
-SHAMapStoreImp::copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node)
+SHAMapStoreImp::copyNode(
+    std::uint64_t& nodeCount,
+    SHAMapTreeNode const& node,
+    NodeObjectType rescueType)
 {
     // Copy a single record from node to dbRotating_
     auto obj = dbRotating_->fetchNodeObject(
@@ -261,8 +295,8 @@ SHAMapStoreImp::copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node)
     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
+        // Reachable from the walked map in memory, but present in no
+        // generation: 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
@@ -270,7 +304,7 @@ SHAMapStoreImp::copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node)
         auto const hash = node.getHash().asUInt256();
         Serializer s;
         node.serializeWithPrefix(s);
-        dbRotating_->store(NodeObjectType::AccountNode, std::move(s.modData()), hash, 0);
+        dbRotating_->store(rescueType, 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());
     }
@@ -343,74 +377,151 @@ SHAMapStoreImp::run()
             if (healthWait() == HealthResult::Stopping)
                 return;
 
-            JLOG(journal_.debug()) << "copying ledger " << validatedSeq;
-            std::uint64_t nodeCount = 0;
-
-            try
-            {
-                validatedLedger->stateMap().snapShot(false)->visitNodes(
-                    [this, &nodeCount](SHAMapTreeNode const& node) {
-                        return copyNode(nodeCount, node);
-                    });
-            }
-            catch (SHAMapMissingNode const& e)
-            {
-                JLOG(journal_.error())
-                    << "Missing node while copying ledger before rotate: " << e.what();
-                continue;
-            }
-
-            if (healthWait() == HealthResult::Stopping)
-                return;
-            // Only log if we completed without a "health" abort
-            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
-            {
-                node_store::DatabaseRotating& db;
-                ~RotationExposureGuard()
-                {
-                    db.setRotationInFlight(false);
-                }
+            // Persist the whole generation ring durably. archiveDb/writableDb are kept
+            // in sync with the ring ends so an older two-backend build could still boot.
+            auto const persistRing = [&](std::vector<std::string> const& generations) {
+                SavedState savedState;
+                savedState.generations = generations;
+                savedState.archiveDb = generations.empty() ? std::string() : generations.front();
+                savedState.writableDb = generations.empty() ? std::string() : generations.back();
+                savedState.lastRotated = lastRotated;
+                stateDb_.setState(savedState);
             };
-            RotationExposureGuard const rotationExposureGuard{*dbRotating_};
-            dbRotating_->setRotationInFlight(true);
-
-            JLOG(journal_.debug()) << "freshening caches";
-            freshenCaches();
-            if (healthWait() == HealthResult::Stopping)
-                return;
-            // Only log if we completed without a "health" abort
-            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() == HealthResult::Stopping)
-                return;
 
+            // Oldest ledger still retained after clearPrior. Its state map is the second
+            // evacuation root: it anchors node versions that current state no longer
+            // references but retained historical ledgers still do.
+            LedgerIndex const oldestRetained = lastRotated;
             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;
-                    stateDb_.setState(savedState);
+            // Seal the current writable generation and open a fresh empty one. This is
+            // O(1): new nodes now accumulate in the new generation, and the whole live
+            // set is NOT re-stored (unlike the old full-state copy).
+            JLOG(journal_.trace()) << "Making a new writable generation";
+            auto newBackend = makeBackendRotating();
+            JLOG(journal_.debug())
+                << validatedSeq << " new writable generation " << newBackend->getName();
+            dbRotating_->advance(std::move(newBackend), persistRing);
 
-                    clearCaches(validatedSeq);
-                });
+            // Once the ring exceeds its generation budget, retire oldest generations:
+            // evacuate only their still-live nodes into the writable backend, then drop
+            // them. This is the O(churn) replacement for the old O(total state)
+            // copy-on-rotate. Loop until back at budget: normally one generation, but an
+            // earlier rotation interrupted between advance and retire leaves extras, and a
+            // single retire per rotation would never shrink the ring back (each rotation's
+            // advance adds one).
+            JLOG(journal_.warn()) << "rotation " << validatedSeq << ": ring has "
+                                  << dbRotating_->generationCount() << " generations, budget "
+                                  << numGenerations_
+                                  << (dbRotating_->generationCount() > numGenerations_
+                                          ? " -> retiring oldest"
+                                          : " -> below budget, no retirement");
+            while (dbRotating_->generationCount() > numGenerations_)
+            {
+                // RAII: close the retire window on any early return / exception, so a
+                // read is never copied forward from a generation we are no longer dropping.
+                struct RetireGuard
+                {
+                    node_store::DatabaseRotating& db;
+                    ~RetireGuard()
+                    {
+                        db.endRetire();
+                    }
+                };
+                RetireGuard const retireGuard{*dbRotating_};
+                dbRotating_->beginRetire();
+
+                // Copy forward the retiring generation's survivors. copyNode fetches each
+                // visited node; the scoped copy-forward re-stores only those served by the
+                // retiring generation (the cold survivors), not the whole live set. Two
+                // evacuation roots cover every version a retained ledger can reference:
+                // the current state, and the oldest retained ledger's state. A version
+                // referenced only by a ledger strictly between the two was created inside
+                // the retention window, so it lives in the newest generations, never the
+                // retiring one. Anything in the retiring generation reachable from neither
+                // root is dead.
+                JLOG(journal_.warn())
+                    << "evacuating retiring generation for ledger " << validatedSeq;
+                std::uint64_t nodeCount = 0;
+                bool evacuationComplete = true;
+                auto const evacuate =
+                    [&](SHAMap const& map, LedgerIndex seq, NodeObjectType rescueType) {
+                        // An empty map (zero root hash, e.g. the transaction tree of a
+                        // no-transaction ledger) has nothing on disk to evacuate.
+                        if (map.getHash().isZero())
+                            return;
+                        try
+                        {
+                            map.snapShot(false)->visitNodes(
+                                [this, &nodeCount, rescueType](SHAMapTreeNode const& node) {
+                                    return copyNode(nodeCount, node, rescueType);
+                                });
+                        }
+                        catch (SHAMapMissingNode const& e)
+                        {
+                            // A node absent from every generation AND memory is already lost:
+                            // it cannot be recovered whether or not we retire, so aborting
+                            // retirement only leaks the ring forever (the observed failure
+                            // mode). Log it, preserve everything reachable + cached
+                            // (freshenCaches below), and still drop the oldest generation so
+                            // the ring stays bounded.
+                            evacuationComplete = false;
+                            JLOG(journal_.warn())
+                                << "evacuation incomplete for ledger " << seq << " after "
+                                << nodeCount
+                                << " nodes -- a node is already lost (retiring anyway to keep "
+                                << "the ring bounded): " << e.what();
+                        }
+                    };
+                evacuate(validatedLedger->stateMap(), validatedSeq, NodeObjectType::AccountNode);
+                if (healthWait() == HealthResult::Stopping)
+                    return;
+                if (oldestRetained != validatedSeq)
+                {
+                    if (auto const boundary = ledgerMaster_->getLedgerBySeq(oldestRetained))
+                    {
+                        evacuate(boundary->stateMap(), oldestRetained, NodeObjectType::AccountNode);
+                        // The boundary ledger's transaction tree was written at its own
+                        // close, which lands in the retiring generation when the budget
+                        // is at its minimum; it is referenced by a retained ledger, so it
+                        // must survive too.
+                        evacuate(
+                            boundary->txMap(), oldestRetained, NodeObjectType::TransactionNode);
+                    }
+                    else
+                    {
+                        JLOG(journal_.warn())
+                            << "evacuation: oldest retained ledger " << oldestRetained
+                            << " unavailable; reads of retained ledgers below " << validatedSeq
+                            << " may miss superseded nodes";
+                    }
+                }
+
+                if (healthWait() == HealthResult::Stopping)
+                    return;
+                JLOG(journal_.warn()) << "evacuated ledger " << validatedSeq << " nodecount "
+                                      << nodeCount << (evacuationComplete ? "" : " (INCOMPLETE)");
+
+                // Any hot node still living only in the retiring generation is re-stored
+                // into the writable backend via the same scoped copy-forward.
+                JLOG(journal_.debug()) << "freshening caches";
+                freshenCaches();
+                if (healthWait() == HealthResult::Stopping)
+                    return;
+
+                // Invalidate FullBelow / ledger caches before the drop so nothing resolves
+                // to the removed backend.
+                clearCaches(validatedSeq);
+                if (healthWait() == HealthResult::Stopping)
+                    return;
+
+                auto const evacuated = dbRotating_->copyForwardCount();
+                dbRotating_->retireOldest(persistRing);
+                clearCaches(validatedSeq);
+                JLOG(journal_.warn()) << "retired oldest generation for ledger " << validatedSeq
+                                      << ": evacuated " << evacuated << " live nodes, ring now "
+                                      << dbRotating_->generationCount() << " generations";
+            }
 
             JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
         }
@@ -443,42 +554,41 @@ SHAMapStoreImp::dbPaths()
     SavedState state = stateDb_.getState();
 
     {
-        auto update = [&dbPath](std::string& sPath) {
+        // If the configured node_db "path" changed, rewrite every stored generation
+        // name (and the legacy pair) to point at the new directory, keeping filenames.
+        using namespace boost::filesystem;
+        bool changed = false;
+        auto relocate = [&dbPath, &changed](std::string& sPath) {
             if (sPath.empty())
-                return false;
-
-            // Check if configured "path" matches stored directory path
-            using namespace boost::filesystem;
+                return;
             auto const stored{path(sPath)};
             if (stored.parent_path() == dbPath)
-                return false;
-
+                return;
             sPath = (dbPath / stored.filename()).string();
-            return true;
+            changed = true;
         };
 
-        if (update(state.writableDb))
-        {
-            update(state.archiveDb);
+        for (auto& name : state.generations)
+            relocate(name);
+        relocate(state.writableDb);
+        relocate(state.archiveDb);
+        if (changed)
             stateDb_.setState(state);
-        }
     }
 
-    bool writableDbExists = false;
-    bool archiveDbExists = false;
-
+    // Every generation named in the ring must exist on disk. Any other directory whose
+    // stem is the backend prefix is an orphan (e.g. a generation whose directory was
+    // created but never persisted into the ring before a crash) and is removed.
+    std::size_t generationsFound = 0;
     std::vector<boost::filesystem::path> pathsToDelete;
     for (boost::filesystem::directory_iterator it(dbPath);
          it != boost::filesystem::directory_iterator();
          ++it)
     {
-        if (state.writableDb == it->path().string())
+        auto const name = it->path().string();
+        if (std::ranges::find(state.generations, name) != state.generations.end())
         {
-            writableDbExists = true;
-        }
-        else if (state.archiveDb == it->path().string())
-        {
-            archiveDbExists = true;
+            ++generationsFound;
         }
         else if (dbPrefix_ == it->path().stem().string())
         {
@@ -486,19 +596,15 @@ SHAMapStoreImp::dbPaths()
         }
     }
 
-    if ((!writableDbExists && !state.writableDb.empty()) ||
-        (!archiveDbExists && !state.archiveDb.empty()) || (writableDbExists != archiveDbExists) ||
-        state.writableDb.empty() != state.archiveDb.empty())
+    if (generationsFound != state.generations.size())
     {
         boost::filesystem::path stateDbPathName = app_.config().legacy(Sections::kDatabasePath);
         stateDbPathName /= dbName_;
         stateDbPathName += "*";
 
         journal_.error() << "state db error:\n"
-                         << "  writableDbExists " << writableDbExists << " archiveDbExists "
-                         << archiveDbExists << '\n'
-                         << "  writableDb '" << state.writableDb << "' archiveDb '"
-                         << state.archiveDb << "\n\n"
+                         << "  generations expected " << state.generations.size() << " found "
+                         << generationsFound << "\n\n"
                          << "The existing data is in a corrupted state.\n"
                          << "To resume operation, remove the files matching "
                          << stateDbPathName.string() << " and contents of the directory "

src/xrpld/app/misc/SHAMapStoreImp.h

@@ -10,6 +10,7 @@
 #include <xrpl/nodestore/Backend.h>
 #include <xrpl/nodestore/Database.h>
 #include <xrpl/nodestore/DatabaseRotating.h>
+#include <xrpl/nodestore/NodeObject.h>
 #include <xrpl/nodestore/Scheduler.h>
 #include <xrpl/protocol/Protocol.h>
 #include <xrpl/rdb/DatabaseCon.h>
@@ -99,6 +100,14 @@ private:
     std::uint32_t deleteInterval_ = 0;
     bool advisoryDelete_ = false;
     std::uint32_t deleteBatch_ = 100;
+    // Bounds for online_delete_generations: at least writable + one archive (the
+    // historical two-backend behavior); the maximum caps fd and directory growth.
+    static constexpr std::uint32_t kMinGenerations = 2;
+    static constexpr std::uint32_t kMaxGenerations = 64;
+    // Number of NodeStore generations to retain in the ring before retiring the oldest.
+    // The disk<->copy tradeoff: larger keeps more (transient) on-disk data but re-stores
+    // a cold node less often (~once per this many rotations instead of every rotation).
+    std::uint32_t numGenerations_ = 8;
     std::chrono::milliseconds backOff_{100};
     std::chrono::seconds ageThreshold_{60};
     /**
@@ -174,7 +183,7 @@ public:
 private:
     // callback for visitNodes
     bool
-    copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node);
+    copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node, NodeObjectType rescueType);
     void
     run();
     void