Use loser-tree merge for ConservationOfLumens

src/bucket/BucketBase.cpp

@@ -426,6 +426,13 @@ BucketBase<BucketT, IndexT>::merge(
     return out.getBucket(bucketManager, &mk);
 }
 
+template <typename BucketT, typename IndexT>
+std::optional<std::pair<std::streamoff, std::streamoff>>
+BucketBase<BucketT, IndexT>::getRangeForType(LedgerEntryType type) const
+{
+    return getIndex().getRangeForType(type);
+}
+
 template void BucketBase<LiveBucket, LiveBucket::IndexT>::mergeInternal<
     MemoryMergeInput<LiveBucket>, std::function<void(BucketEntry const&)>,
     std::vector<BucketInputIterator<LiveBucket>>&, bool&>(

src/bucket/BucketBase.h

@@ -187,6 +187,12 @@ class BucketBase : public NonMovableOrCopyable
     static std::string randomBucketName(std::string const& tmpDir);
     static std::string randomBucketIndexName(std::string const& tmpDir);
 
+    // Returns [lowerBound, upperBound) of file offsets for all entries of the
+    // given type in the bucket, or std::nullopt if no entries of this type
+    // exist.
+    std::optional<std::pair<std::streamoff, std::streamoff>>
+    getRangeForType(LedgerEntryType type) const;
+
 #ifdef BUILD_TESTS
     IndexT const&
     getIndexForTesting() const

src/bucket/BucketListSnapshot.cpp

@@ -709,9 +709,10 @@ namespace
 // to be positioned at the start of the type range. This is basically the same
 // as SearchableLiveBucketListSnapshot::scanForEntriesOfType's scanBucket except
 // with more control over when iteration happens.
-class BucketEntryIterator
+template <typename BucketT> class BucketEntryIterator
 {
-    BucketEntry mEntry;
+    BUCKET_TYPE_ASSERT(BucketT);
+    BucketT::EntryT mEntry;
     LedgerKey mKey;
     XDRInputFileStream mStream;
     LedgerEntryType const mType;
@@ -722,7 +723,7 @@ class BucketEntryIterator
     {
     }
 
-    BucketEntry const&
+    BucketT::EntryT const&
     getEntry() const
     {
         return mEntry;
@@ -738,7 +739,7 @@ class BucketEntryIterator
     {
         while (mStream.readOne(mEntry))
         {
-            if (isBucketMetaEntry<LiveBucket>(mEntry))
+            if (isBucketMetaEntry<BucketT>(mEntry))
             {
                 continue;
             }
@@ -758,10 +759,11 @@ class BucketEntryIterator
 };
 } // namespace
 
+template <class BucketT>
 void
-SearchableLiveBucketListSnapshot::scanForLiveEntriesOfType(
+SearchableBucketListSnapshot<BucketT>::scanForCurrentEntriesOfType(
     LedgerEntryType type,
-    std::function<void(LedgerEntry const&, LedgerKey const&)> callback) const
+    std::function<Loop(LedgerEntry const&, LedgerKey const&)> callback) const
 {
     ZoneScoped;
     // We implement this as a k-way merge over all buckets. We use a loser tree
@@ -777,9 +779,9 @@ SearchableLiveBucketListSnapshot::scanForLiveEntriesOfType(
     // intermediate nodes, we just store an index, since copying the XDR types
     // is probably more expensive than the extra indirection.
 
-    std::vector<BucketEntryIterator> iterators;
+    std::vector<BucketEntryIterator<BucketT>> iterators;
     loopAllBuckets(
-        [&iterators, type](std::shared_ptr<LiveBucket const> const& bucket) {
+        [&iterators, type](std::shared_ptr<BucketT const> const& bucket) {
             if (bucket->isEmpty())
             {
                 return Loop::INCOMPLETE;
@@ -878,9 +880,27 @@ SearchableLiveBucketListSnapshot::scanForLiveEntriesOfType(
         {
             last = key;
             auto& entry = iter.getEntry();
-            if (entry.type() == LIVEENTRY || entry.type() == INITENTRY)
+            if constexpr (std::is_same_v<BucketT, LiveBucket>)
             {
-                callback(entry.liveEntry(), key);
+                if (entry.type() == LIVEENTRY || entry.type() == INITENTRY)
+                {
+                    if (callback(entry.liveEntry(), key) == Loop::COMPLETE)
+                    {
+                        return;
+                    }
+                }
+            }
+            else
+            {
+                static_assert(std::is_same_v<BucketT, HotArchiveBucket>,
+                              "unexpected bucket type");
+                if (entry.type() == HOT_ARCHIVE_ARCHIVED)
+                {
+                    if (callback(entry.archivedEntry(), key) == Loop::COMPLETE)
+                    {
+                        return;
+                    }
+                }
             }
         }
         first = false;

src/bucket/BucketListSnapshot.h

@@ -201,6 +201,19 @@ template <class BucketT> class SearchableBucketListSnapshot
     // Access to underlying data (for copying/refreshing)
     std::shared_ptr<BucketListSnapshotData<BucketT> const> const&
     getSnapshotData() const;
+
+    // Iterate over the visible, non-shadowed entries of a given type, i.e., the
+    // newest entry for each key if it isn't tombstoned. Calls callback for
+    // these entries, stopping early if callback returns Loop::COMPLETE. For the
+    // live bucket list, this will be the current values (LIVEENTRY or
+    // INITENTRY) for all keys of the given type that aren't dead (DEADENTRY).
+    // For the hot archive bucket list, this will be the latest version of all
+    // keys of the given type that are still archived (HOT_ARCHIVE_ARCHIVED) and
+    // haven't been restored (HOT_ARCHIVE_LIVE).
+    void scanForCurrentEntriesOfType(
+        LedgerEntryType type,
+        std::function<Loop(LedgerEntry const&, LedgerKey const&)> callback)
+        const;
 };
 
 // Live bucket list snapshot with additional query methods
@@ -241,13 +254,6 @@ class SearchableLiveBucketListSnapshot
         LedgerEntryType type,
         std::function<Loop(BucketEntry const&)> callback) const;
 
-    // Iterate over all live entries of a given type. Note that this handles
-    // shadowing and only returns the latest live entry for each key.
-    void scanForLiveEntriesOfType(
-        LedgerEntryType type,
-        std::function<void(LedgerEntry const&, LedgerKey const&)> callback)
-        const;
-
     friend class ImmutableLedgerData;
     friend class ImmutableLedgerView;
 };

src/bucket/HotArchiveBucketIndex.cpp

@@ -28,6 +28,12 @@ HotArchiveBucketIndex::HotArchiveBucketIndex(
                mDiskIndex.getPageSize(), filename);
 }
 
+std::optional<std::pair<std::streamoff, std::streamoff>>
+HotArchiveBucketIndex::getRangeForType(LedgerEntryType type) const
+{
+    return mDiskIndex.getRangeForType(type);
+}
+
 std::streamoff
 HotArchiveBucketIndex::getPageSize(Config const& cfg, size_t bucketSize)
 {

src/bucket/HotArchiveBucketIndex.h

@@ -83,6 +83,9 @@ class HotArchiveBucketIndex : public NonMovableOrCopyable
     {
     }
 
+    std::optional<std::pair<std::streamoff, std::streamoff>>
+    getRangeForType(LedgerEntryType type) const;
+
     std::pair<IndexReturnT, IterT> scan(IterT start, LedgerKey const& k) const;
 
     BucketEntryCounters const&

src/bucket/LiveBucket.cpp

@@ -371,12 +371,6 @@ LiveBucket::getMaxCacheSize() const
 }
 #endif // BUILD_TESTS
 
-std::optional<std::pair<std::streamoff, std::streamoff>>
-LiveBucket::getRangeForType(LedgerEntryType type) const
-{
-    return getIndex().getRangeForType(type);
-}
-
 std::vector<BucketEntry>
 LiveBucket::convertToBucketEntry(bool useInit,
                                  std::vector<LedgerEntry> const& initEntries,

src/bucket/LiveBucket.h

@@ -105,14 +105,6 @@ class LiveBucket : public BucketBase<LiveBucket, LiveBucketIndex>,
     size_t getMaxCacheSize() const;
 #endif
 
-    // Returns [lowerBound, upperBound) of file offsets for all entries of the
-    // given type in the bucket, or std::nullopt if no entries of this type
-    // exist. Note that if the underlying index is a page based index, this is a
-    // rough bound such that entries of another type may also be present in the
-    // range.
-    std::optional<std::pair<std::streamoff, std::streamoff>>
-    getRangeForType(LedgerEntryType type) const;
-
     // Create a fresh bucket from given vectors of init (created) and live
     // (updated) LedgerEntries, and dead LedgerEntryKeys. The bucket will
     // be sorted, hashed, and adopted in the provided BucketManager.

src/bucket/test/BucketIndexTests.cpp

@@ -1607,9 +1607,9 @@ TEST_CASE("getRangeForType bounds verification", "[bucket][bucketindex]")
     testAllIndexTypes(f);
 }
 
-// Fixture for the scanForLiveEntriesOfType randomized test. Records the
-// live-entry set created by buildMultiVersionTest. Includes the genesis
-// root ACCOUNT but not genesis CONFIG_SETTINGs.
+// Fixture for the scanForCurrentEntriesOfType randomized test. Records the
+// live-entry set created by buildMultiVersionTest. Includes the genesis root
+// ACCOUNT but not genesis CONFIG_SETTINGs.
 class BucketIndexScanTest : public BucketIndexTest
 {
     UnorderedMap<LedgerKey, LedgerEntry> mAllEntries;
@@ -1654,7 +1654,7 @@ class BucketIndexScanTest : public BucketIndexTest
     }
 };
 
-TEST_CASE("scanForLiveEntriesOfType randomized testing",
+TEST_CASE("scanForCurrentEntriesOfType randomized testing",
           "[bucket][bucketindex]")
 {
     // Scan for each of the given types and check the result against the
@@ -1682,6 +1682,7 @@ TEST_CASE("scanForLiveEntriesOfType randomized testing",
 
                     // Each key should only be emitted once
                     REQUIRE(found.emplace(key, entry).second);
+                    return Loop::INCOMPLETE;
                 });
             REQUIRE(found == expected);
         }
@@ -1704,8 +1705,10 @@ TEST_CASE("scanForLiveEntriesOfType randomized testing",
             INFO(
                 "type = " << xdr::xdr_traits<LedgerEntryType>::enum_name(type));
             ledgerView.scanCurrentLiveEntriesOfType(
-                type,
-                [](LedgerEntry const&, LedgerKey const&) { REQUIRE(false); });
+                type, [](LedgerEntry const&, LedgerKey const&) {
+                    REQUIRE(false);
+                    return Loop::COMPLETE;
+                });
         }
     });
 
@@ -1721,7 +1724,7 @@ TEST_CASE("scanForLiveEntriesOfType randomized testing",
     });
 }
 
-TEST_CASE("scanForLiveEntriesOfType loser tree unit tests",
+TEST_CASE("scanForCurrentEntriesOfType loser tree unit tests",
           "[bucket][bucketindex]")
 {
     auto f = [&](Config& cfg) {
@@ -1794,15 +1797,18 @@ TEST_CASE("scanForLiveEntriesOfType loser tree unit tests",
 
                     // Each key must be emitted exactly once
                     REQUIRE(found.emplace(key, entry).second);
+                    return Loop::INCOMPLETE;
                 });
             return found;
         };
 
         auto requireNoCallback = [](ImmutableLedgerView const& view,
                                     LedgerEntryType type) {
             view.scanCurrentLiveEntriesOfType(
-                type,
-                [](LedgerEntry const&, LedgerKey const&) { REQUIRE(false); });
+                type, [](LedgerEntry const&, LedgerKey const&) {
+                    REQUIRE(false);
+                    return Loop::COMPLETE;
+                });
         };
 
         SECTION("k disjoint buckets")