fix: Use the renamed PartitionMapType alias

develop renamed partition_map_type to PartitionMapType (#8177). The const map() overload this branch adds still used the old name, so every build failed after the develop merge.

include/xrpl/basics/TaggedCache.h

@@ -245,6 +245,37 @@ public:
     std::vector<key_type>
     getKeys() const;
 
+    /**
+     * Visit every key, one map partition at a time.
+     *
+     * The mutex is held only while one partition's keys are copied out, and
+     * `f` runs with the mutex released. So the longest hold is one partition,
+     * not the whole cache. getKeys() holds the mutex across every entry, and
+     * on a cache of tens of millions of entries that hold lasts seconds,
+     * during which every other user of the cache waits.
+     *
+     * `f` is called once per partition, in partition order, with that
+     * partition's keys as they were when that partition was copied. There is
+     * no single instant at which the whole cache was observed, so the snapshot
+     * is weaker than getKeys() in one direction and stronger in the other: a
+     * key inserted into a partition already copied is not visited, while one
+     * inserted into a later partition is, and a key erased before its
+     * partition is copied is not visited even though getKeys() would have
+     * returned it.
+     *
+     * @note `f` receives a reference to one buffer that is cleared and refilled
+     *       for each partition. Any pointer, iterator or reference into it is
+     *       dangling by the next call, so a callback that needs the keys beyond
+     *       its own invocation must copy them.
+     *
+     * @param f Callable taking `std::vector<key_type> const&` and returning
+     *          false to stop the walk before the next partition.
+     * @return true if every partition was visited, false if `f` stopped early.
+     */
+    template <class F>
+    [[nodiscard]] bool
+    forEachKeyPartition(F&& f) const;
+
     // CachedSLEs functions.
     /**
      * Returns the fraction of cache hits.

include/xrpl/basics/TaggedCache.ipp

@@ -6,6 +6,7 @@
 #include <xrpl/basics/scope.h>
 
 #include <algorithm>
+#include <utility>
 
 namespace xrpl {
 
@@ -647,6 +648,43 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
     return v;
 }
 
+template <
+    class Key,
+    class T,
+    bool IsKeyCache,
+    class SharedWeakUnionPointer,
+    class SharedPointerType,
+    class Hash,
+    class KeyEqual,
+    class Mutex>
+template <class F>
+inline bool
+TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash, KeyEqual, Mutex>::
+    forEachKeyPartition(F&& f) const
+{
+    // One buffer for the whole walk. clear() keeps the capacity, so it only
+    // reallocates for a partition larger than every earlier one.
+    std::vector<key_type> keys;
+    // partitions() and map() are fixed at construction, so the bound and the
+    // indexing need no lock.
+    for (std::size_t p = 0; p < cache_.partitions(); ++p)
+    {
+        keys.clear();
+        {
+            std::scoped_lock const lock(mutex_);
+            auto const& partition = cache_.map()[p];
+            keys.reserve(partition.size());
+            for (auto const& entry : partition)
+                keys.push_back(entry.first);
+        }
+        // as_const so the documented `const&` parameter is the enforced one: f
+        // must not mutate or move from the buffer the next partition reuses.
+        if (!f(std::as_const(keys)))
+            return false;
+    }
+    return true;
+}
+
 template <
     class Key,
     class T,

include/xrpl/basics/partitioned_unordered_map.h

@@ -240,6 +240,15 @@ public:
         return map_;
     }
 
+    /**
+     * Read-only access to the partitions, for a const walk over one at a time.
+     */
+    PartitionMapType const&
+    map() const
+    {
+        return map_;
+    }
+
     Iterator
     begin()
     {

src/libxrpl/nodestore/DatabaseRotatingImp.cpp

@@ -200,8 +200,8 @@ DatabaseRotatingImp::fetchNodeObject(
             // 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.
+            // into the cache after the freshen already walked its partition
+            // would otherwise survive only in RAM once the archive is dropped.
             if (duplicate || rotationInFlight_.load(std::memory_order_acquire))
             {
                 if (!duplicate)

src/tests/libxrpl/basics/TaggedCache.cpp

@@ -10,9 +10,13 @@
 #include <gtest/gtest.h>
 #include <helpers/TestSink.h>
 
+#include <algorithm>
+#include <cstddef>
 #include <memory>
 #include <string>
+#include <thread>
 #include <utility>
+#include <vector>
 
 namespace xrpl {
 
@@ -243,4 +247,158 @@ TEST(TaggedCacheTest, tagged_cache)
     }
 }
 
+TEST(TaggedCacheTest, for_each_key_partition_visits_every_key_exactly_once)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+    TestStopwatch clock;
+    clock.set(0);
+
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+    Cache c("partitions", 0, 1s, clock, journal);
+
+    constexpr LedgerIndex keyCount = 10'000;
+    for (LedgerIndex i = 0; i < keyCount; ++i)
+        c.insert(i, "v");
+    ASSERT_EQ(c.getTrackSize(), static_cast<int>(keyCount));
+
+    std::vector<LedgerIndex> visited;
+    std::size_t calls = 0;
+    EXPECT_TRUE(c.forEachKeyPartition([&](std::vector<LedgerIndex> const& keys) {
+        ++calls;
+        visited.insert(visited.end(), keys.begin(), keys.end());
+        return true;
+    }));
+
+    // The cache's map is built with one partition per hardware thread, and the
+    // callback runs once per partition.
+    EXPECT_EQ(calls, static_cast<std::size_t>(std::thread::hardware_concurrency()));
+
+    auto expected = c.getKeys();
+    std::ranges::sort(expected);
+    std::ranges::sort(visited);
+    ASSERT_EQ(visited.size(), keyCount);
+    EXPECT_EQ(visited, expected);
+}
+
+TEST(TaggedCacheTest, for_each_key_partition_on_empty_cache_calls_back_with_empty_batches)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+    TestStopwatch clock;
+    clock.set(0);
+
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+    Cache const c("empty", 0, 1s, clock, journal);
+
+    std::size_t calls = 0;
+    std::size_t keysSeen = 0;
+    EXPECT_TRUE(c.forEachKeyPartition([&](std::vector<LedgerIndex> const& keys) {
+        ++calls;
+        keysSeen += keys.size();
+        return true;
+    }));
+
+    EXPECT_EQ(calls, static_cast<std::size_t>(std::thread::hardware_concurrency()));
+    EXPECT_EQ(keysSeen, 0u);
+}
+
+TEST(TaggedCacheTest, for_each_key_partition_releases_the_mutex_while_the_callback_runs)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+    TestStopwatch clock;
+    clock.set(0);
+
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+    Cache c("unlocked", 0, 1s, clock, journal);
+    for (LedgerIndex i = 0; i < 1'000; ++i)
+        c.insert(i, "v");
+
+    // A recursive mutex lets the calling thread re-lock it, so the probe has to
+    // run on another thread: try_lock there succeeds only if this thread is not
+    // holding the mutex while the callback runs.
+    // With no partitions the callback never runs and every assertion below is
+    // vacuously true, which would hide the one thing this test exists to prove.
+    ASSERT_GT(std::thread::hardware_concurrency(), 0u);
+
+    std::size_t calls = 0;
+    std::size_t unlockedDuringCallback = 0;
+    EXPECT_TRUE(c.forEachKeyPartition([&](std::vector<LedgerIndex> const&) {
+        ++calls;
+        bool acquired = false;
+        std::thread probe([&] {
+            acquired = c.peekMutex().try_lock();
+            if (acquired)
+                c.peekMutex().unlock();
+        });
+        probe.join();
+        if (acquired)
+            ++unlockedDuringCallback;
+        return true;
+    }));
+
+    EXPECT_EQ(calls, static_cast<std::size_t>(std::thread::hardware_concurrency()));
+    EXPECT_EQ(unlockedDuringCallback, calls);
+}
+
+TEST(TaggedCacheTest, for_each_key_partition_tolerates_inserts_from_another_thread)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+    TestStopwatch clock;
+    clock.set(0);
+
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+    Cache c("concurrent", 0, 1s, clock, journal);
+
+    constexpr LedgerIndex original = 5'000;
+    for (LedgerIndex i = 0; i < original; ++i)
+        c.insert(i, "v");
+
+    // While one partition's keys are being handled, another thread inserts a
+    // fresh key. The mutex is free during the callback, so the insert must
+    // complete -- a held mutex would deadlock the join -- and every original key
+    // must still be visited exactly once. Whether a late key is visited depends
+    // on which partition it lands in, so this asserts nothing about those.
+    std::vector<LedgerIndex> visited;
+    LedgerIndex next = original;
+    EXPECT_TRUE(c.forEachKeyPartition([&](std::vector<LedgerIndex> const& keys) {
+        visited.insert(visited.end(), keys.begin(), keys.end());
+        std::thread inserter([&] { c.insert(next, "late"); });
+        inserter.join();
+        ++next;
+        return true;
+    }));
+
+    std::ranges::sort(visited);
+    EXPECT_EQ(std::ranges::adjacent_find(visited), visited.end());
+    auto const originals =
+        std::ranges::count_if(visited, [](LedgerIndex k) { return k < original; });
+    EXPECT_EQ(static_cast<LedgerIndex>(originals), original);
+    EXPECT_EQ(c.getTrackSize(), static_cast<int>(next));
+}
+
+TEST(TaggedCacheTest, for_each_key_partition_stops_when_the_callback_returns_false)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+    TestStopwatch clock;
+    clock.set(0);
+
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+    Cache c("early-stop", 0, 1s, clock, journal);
+    for (LedgerIndex i = 0; i < 10'000; ++i)
+        c.insert(i, "v");
+
+    // The callback refuses the first partition, so no later partition is even
+    // copied out, and the walk reports that it did not finish.
+    std::size_t calls = 0;
+    EXPECT_FALSE(c.forEachKeyPartition([&](std::vector<LedgerIndex> const&) {
+        ++calls;
+        return false;
+    }));
+    EXPECT_EQ(calls, 1u);
+}
+
 }  // namespace xrpl

src/xrpld/app/misc/SHAMapStoreImp.cpp

@@ -435,7 +435,7 @@ SHAMapStoreImp::run()
             JLOG(journal_.debug())
                 << "copied ledger " << validatedSeq << " nodecount " << nodeCount;
 
-            // Close the getKeys()->swap exposure window: from here until
+            // Close the freshen->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

src/xrpld/app/misc/SHAMapStoreImp.h

@@ -197,20 +197,37 @@ private:
     std::unique_ptr<node_store::Backend>
     makeBackendRotating(std::string path = std::string());
 
+    /**
+     * Re-fetch every key of one cache so any node only the archive still holds
+     * is copied into the writable backend before the archive is deleted.
+     *
+     * Keys are copied one map partition at a time, and every fetch runs with
+     * the cache mutex released. getKeys() would hold the mutex across the whole
+     * cache, which on a large tree-node cache stalls every job for seconds and
+     * can drop the node out of sync. The walk is abandoned as soon as
+     * healthWait() reports the node is no longer keeping up.
+     *
+     * @param cache The cache to walk.
+     * @return true if healthWait() said the rotation must stop or expire, so
+     *         the caller must abandon it; false if every key was fetched.
+     */
     template <class CacheInstance>
     bool
     freshenCache(CacheInstance& cache)
     {
         std::uint64_t check = 0;
 
-        for (auto const& key : cache.getKeys())
-        {
-            dbRotating_->fetchNodeObject(key, 0, node_store::FetchType::Synchronous, true);
-            if (!(++check % checkHealthInterval_) && healthWait() != HealthResult::KeepGoing)
-                return true;
-        }
-
-        return false;
+        // Returning false abandons the walk, so an unhealthy check skips the
+        // remaining partitions instead of copying keys nobody will fetch.
+        return !cache.forEachKeyPartition([&](auto const& keys) {
+            for (auto const& key : keys)
+            {
+                dbRotating_->fetchNodeObject(key, 0, node_store::FetchType::Synchronous, true);
+                if (!(++check % checkHealthInterval_) && healthWait() != HealthResult::KeepGoing)
+                    return false;
+            }
+            return true;
+        });
     }
 
     /**