diff --git a/src/bucket/Bucket.cpp b/src/bucket/Bucket.cpp index bd7022bd34..80876428e9 100644 --- a/src/bucket/Bucket.cpp +++ b/src/bucket/Bucket.cpp @@ -1071,23 +1071,6 @@ HotArchiveBucket::isTombstoneEntry(HotArchiveBucketEntry const& e) return e.type() == HOT_ARCHIVE_LIVE; } -template std::shared_ptr Bucket::merge( - BucketManager& bucketManager, uint32_t maxProtocolVersion, - std::shared_ptr const& oldBucket, - std::shared_ptr const& newBucket, - std::vector> const& shadows, - bool keepTombstoneEntries, bool countMergeEvents, asio::io_context& ctx, - bool doFsync); - -template std::shared_ptr Bucket::merge( - BucketManager& bucketManager, uint32_t maxProtocolVersion, - std::shared_ptr const& oldBucket, - std::shared_ptr const& newBucket, - std::vector> const& shadows, - bool keepTombstoneEntries, bool countMergeEvents, asio::io_context& ctx, - bool doFsync); -} - BucketEntryCounters& BucketEntryCounters::operator+=(BucketEntryCounters const& other) { @@ -1114,4 +1097,20 @@ BucketEntryCounters::operator!=(BucketEntryCounters const& other) const { return !(*this == other); } + +template std::shared_ptr Bucket::merge( + BucketManager& bucketManager, uint32_t maxProtocolVersion, + std::shared_ptr const& oldBucket, + std::shared_ptr const& newBucket, + std::vector> const& shadows, + bool keepTombstoneEntries, bool countMergeEvents, asio::io_context& ctx, + bool doFsync); + +template std::shared_ptr Bucket::merge( + BucketManager& bucketManager, uint32_t maxProtocolVersion, + std::shared_ptr const& oldBucket, + std::shared_ptr const& newBucket, + std::vector> const& shadows, + bool keepTombstoneEntries, bool countMergeEvents, asio::io_context& ctx, + bool doFsync); } \ No newline at end of file diff --git a/src/bucket/Bucket.h b/src/bucket/Bucket.h index 3fd3f3ad46..0f65acffa6 100644 --- a/src/bucket/Bucket.h +++ b/src/bucket/Bucket.h @@ -37,6 +37,14 @@ namespace stellar * Two buckets can be merged together efficiently (in a single pass): elements * from the newer bucket overwrite elements from the older bucket, the rest are * merged in sorted order, and all elements are hashed while being added. + * + * Different types of BucketList vary on the type of entries they contain and by + * extension the merge logic of those entries. Additionally, some types of + * BucketList may have special operations only relevant to that specific type. + * This pure virtual base class provides the core functionality of a BucketList + * container and must be extened for each specific BucketList type. In + * particular, the fresh and merge functions must be defined for the specific + * type, while other functionality can be shared. */ class AbstractLedgerTxn; @@ -45,6 +53,8 @@ class BucketManager; struct EvictionResultEntry; class EvictionStatistics; struct BucketEntryCounters; +template class SearchableBucketListSnapshot; +enum class LedgerEntryTypeAndDurability : uint32_t; class Bucket : public NonMovableOrCopyable { @@ -132,12 +142,18 @@ class Bucket : public NonMovableOrCopyable template friend class BucketSnapshotBase; }; -template class SearchableBucketListSnapshot; +/* + * Live Buckets are used by the LiveBucketList to store the current canonical + * state of the ledger. They contain entries of type BucketEntry. + */ class LiveBucket : public Bucket, public std::enable_shared_from_this { public: LiveBucket(); + virtual ~LiveBucket() + { + } LiveBucket(std::string const& filename, Hash const& hash, std::unique_ptr&& index); @@ -209,6 +225,10 @@ class LiveBucket : public Bucket, friend class LiveBucketSnapshot; }; +/* + * Hot Archive Buckets are used by the HotBucketList to store recently evicted + * entries. They contain entries of type HotArchiveBucketEntry. + */ class HotArchiveBucket : public Bucket, public std::enable_shared_from_this { @@ -219,6 +239,9 @@ class HotArchiveBucket : public Bucket, public: HotArchiveBucket(); + virtual ~HotArchiveBucket() + { + } HotArchiveBucket(std::string const& filename, Hash const& hash, std::unique_ptr&& index); uint32_t getBucketVersion() const override; @@ -237,7 +260,6 @@ class HotArchiveBucket : public Bucket, friend class HotArchiveBucketSnapshot; }; -enum class LedgerEntryTypeAndDurability : uint32_t; struct BucketEntryCounters { std::map entryTypeCounts; diff --git a/src/bucket/BucketIndexImpl.cpp b/src/bucket/BucketIndexImpl.cpp index cb41ca7abd..1e010d40d8 100644 --- a/src/bucket/BucketIndexImpl.cpp +++ b/src/bucket/BucketIndexImpl.cpp @@ -204,7 +204,11 @@ BucketIndexImpl::BucketIndexImpl(BucketManager& bm, { mData.keysToOffset.emplace_back(key, pos); } - countEntry(be); + + if constexpr (std::is_same::value) + { + countEntry(be); + } } pos = in.pos(); diff --git a/src/bucket/BucketList.cpp b/src/bucket/BucketList.cpp index cdb079b498..27364c34eb 100644 --- a/src/bucket/BucketList.cpp +++ b/src/bucket/BucketList.cpp @@ -27,8 +27,8 @@ namespace stellar template <> BucketListDepth BucketListBase::kNumLevels = 11; -// TODO: I made this number up, do some analysis and pick a better value or -// make this a configurable network config. +// TODO: This is an arbitrary number. Do some analysis and pick a better value +// or make this a configurable network config. template <> BucketListDepth BucketListBase::kNumLevels = 9; template @@ -577,12 +577,6 @@ HotArchiveBucketList::addBatch(Application& app, uint32_t currLedger, for (uint32_t i = static_cast(mLevels.size()) - 1; i != 0; --i) { - /* - CLOG_DEBUG(Bucket, "curr={}, half(i-1)={}, size(i-1)={}, - roundDown(curr,half)={}, roundDown(curr,size)={}", currLedger, - levelHalf(i-1), levelSize(i-1), roundDown(currLedger, levelHalf(i-1)), - roundDown(currLedger, levelSize(i-1))); - */ if (levelShouldSpill(currLedger, i - 1)) { /** @@ -771,7 +765,7 @@ LiveBucketList::addBatch(Application& app, uint32_t currLedger, } BucketEntryCounters -BucketList::sumBucketEntryCounters() const +LiveBucketList::sumBucketEntryCounters() const { BucketEntryCounters counters; for (auto const& lev : mLevels) diff --git a/src/bucket/BucketList.h b/src/bucket/BucketList.h index 744db303a9..3e464c006f 100644 --- a/src/bucket/BucketList.h +++ b/src/bucket/BucketList.h @@ -401,6 +401,12 @@ class BucketListDepth template friend class testutil::BucketListDepthModifier; }; +// While every BucketList shares the same high level structure wrt to spill +// schedules, merges at the bucket level, etc, each BucketList type hold +// different types of entries and has different merge logic at the individual +// entry level. This pure virtual base class defines the shared structure of all +// BucketLists. It must be extended for each specific BucketList type, where the +// template parameter BucketT refers to the underlying Bucket type. template class BucketListBase { static_assert(std::is_same_v || @@ -503,6 +509,11 @@ template class BucketListBase uint64_t getSize() const; }; +// The LiveBucketList stores the current canonical state of the ledger. It is +// made up of LiveBucket buckets, which in turn store individual entries of type +// BucketEntry. When an entry is "evicted" from the ledger, it is removed from +// the LiveBucketList. Depending on the evicted entry type, it may then be added +// to the HotArchiveBucketList. class LiveBucketList : public BucketListBase { public: @@ -545,6 +556,9 @@ class LiveBucketList : public BucketListBase BucketEntryCounters sumBucketEntryCounters() const; }; +// The HotArchiveBucketList stores recently evicted entries. It contains Buckets +// of type HotArchiveBucket, which store individual entries of type +// HotArchiveBucketEntry. class HotArchiveBucketList : public BucketListBase { private: diff --git a/src/bucket/BucketListSnapshot.cpp b/src/bucket/BucketListSnapshot.cpp index 96e2c62a46..b071567b09 100644 --- a/src/bucket/BucketListSnapshot.cpp +++ b/src/bucket/BucketListSnapshot.cpp @@ -12,13 +12,14 @@ #include "medida/timer.h" #include "util/GlobalChecks.h" +#include +#include namespace stellar { template BucketListSnapshot::BucketListSnapshot( - BucketListBase const& bl, - LedgerHeader header) + BucketListBase const& bl, LedgerHeader header) : mHeader(std::move(header)) { releaseAssert(threadIsMain()); @@ -32,7 +33,7 @@ BucketListSnapshot::BucketListSnapshot( template BucketListSnapshot::BucketListSnapshot( - BucketListSnapshot const& snapshot) + BucketListSnapshot const& snapshot) : mLevels(snapshot.mLevels), mHeader(snapshot.mHeader) { } @@ -51,10 +52,20 @@ BucketListSnapshot::getLedgerSeq() const return mHeader.ledgerSeq; } +template +LedgerHeader const& +SearchableBucketListSnapshotBase::getLedgerHeader() +{ + releaseAssert(mSnapshot); + mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); + return mSnapshot->getLedgerHeader(); +} + template void SearchableBucketListSnapshotBase::loopAllBuckets( - std::function f, BucketListSnapshot const& snapshot) const + std::function f, + BucketListSnapshot const& snapshot) const { for (auto const& lev : snapshot.getLevels()) { @@ -75,70 +86,6 @@ SearchableBucketListSnapshotBase::loopAllBuckets( } } -// Loads bucket entry for LedgerKey k. Returns , -// where bloomMiss is true if a bloom miss occurred during the load. -std::pair, bool> -getLedgerEntryInternal(LedgerKey const& k, BucketListSnapshot const& snapshot) -{ - std::shared_ptr result{}; - auto sawBloomMiss = false; - - auto f = [&](BucketSnapshot const& b) { - auto [be, bloomMiss] = b.getBucketEntry(k); - sawBloomMiss = sawBloomMiss || bloomMiss; - - if (be.has_value()) - { - result = - be.value().type() == DEADENTRY - ? nullptr - : std::make_shared(be.value().liveEntry()); - return true; - } - else - { - return false; - } - }; - - loopAllBuckets(f, snapshot); - return {result, sawBloomMiss}; -} - -std::vector -loadKeysInternal(std::set const& inKeys, - BucketListSnapshot const& snapshot, LedgerKeyMeter* lkMeter) -{ - std::vector entries; - - // Make a copy of the key set, this loop is destructive - auto keys = inKeys; - auto f = [&](BucketSnapshot const& b) { - b.loadKeysWithLimits(keys, entries, lkMeter); - return keys.empty(); - }; - - loopAllBuckets(f, snapshot); - return entries; -} - -} - -uint32_t -SearchableBucketListSnapshot::getLedgerSeq() const -{ - releaseAssert(mSnapshot); - return mSnapshot->getLedgerSeq(); -} - -LedgerHeader const& -SearchableBucketListSnapshot::getLedgerHeader() -{ - releaseAssert(mSnapshot); - mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); - return mSnapshot->getLedgerHeader(); -} - EvictionResult SearchableLiveBucketListSnapshot::scanForEviction( uint32_t ledgerSeq, EvictionCounters& counters, @@ -204,41 +151,61 @@ SearchableLiveBucketListSnapshot::loadKeysWithLimits( return keys.empty(); }; - mSnapshotManager.maybeUpdateSnapshot(mSnapshot); + mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); if (threadIsMain()) { auto timer = mSnapshotManager.recordBulkLoadMetrics("prefetch", inKeys.size()) .TimeScope(); - loopAllBuckets(loadKeysLoop); + loopAllBuckets(loadKeysLoop, *mSnapshot); } else { // TODO: Background metrics - loopAllBuckets(loadKeysLoop); + loopAllBuckets(loadKeysLoop, *mSnapshot); + } + + return entries; +} + +std::optional> +SearchableLiveBucketListSnapshot::loadKeysFromLedger( + std::set const& inKeys, uint32_t ledgerSeq) +{ + ZoneScoped; + + // Make a copy of the key set, this loop is destructive + auto keys = inKeys; + std::vector entries; + auto loadKeysLoop = [&](auto const& b) { + b.loadKeys(keys, entries, /*lkMeter=*/nullptr); + return keys.empty(); + }; + + mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); + + if (ledgerSeq == mSnapshot->getLedgerSeq()) + { + loopAllBuckets(loadKeysLoop, *mSnapshot); + } + else + { + auto iter = mHistoricalSnapshots.find(ledgerSeq); + if (iter == mHistoricalSnapshots.end()) + { + return std::nullopt; + } + + releaseAssert(iter->second); + loopAllBuckets(loadKeysLoop, *iter->second); } return entries; } std::shared_ptr -SearchableLiveBucketListSnapshot::getLedgerEntry(LedgerKey const& k) +SearchableLiveBucketListSnapshot::load(LedgerKey const& k) { - // if (ledgerSeq == mSnapshot->getLedgerSeq()) - // { - // auto result = loadKeysInternal(inKeys, *mSnapshot, /*lkMeter=*/nullptr); - // return {result, true}; - // } - - // auto iter = mHistoricalSnapshots.find(ledgerSeq); - // if (iter == mHistoricalSnapshots.end()) - // { - // return {{}, false}; - // } - - // releaseAssert(iter->second); - // auto result = loadKeysInternal(inKeys, *iter->second, /*lkMeter=*/nullptr); - // return {result, true}; ZoneScoped; std::shared_ptr result{}; @@ -263,18 +230,18 @@ SearchableLiveBucketListSnapshot::getLedgerEntry(LedgerKey const& k) } }; - mSnapshotManager.maybeUpdateSnapshot(mSnapshot); + mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); if (threadIsMain()) { mSnapshotManager.startPointLoadTimer(); - loopAllBuckets(loadKeyBucketLoop); + loopAllBuckets(loadKeyBucketLoop, *mSnapshot); mSnapshotManager.endPointLoadTimer(k.type(), sawBloomMiss); return result; } else { // TODO: Background metrics - loopAllBuckets(loadKeyBucketLoop); + loopAllBuckets(loadKeyBucketLoop, *mSnapshot); return result; } } @@ -324,7 +291,7 @@ SearchableLiveBucketListSnapshot::loadPoolShareTrustLinesByAccountAndAsset( return trustlinesToLoad.empty(); }; - loopAllBuckets(loadKeysLoop); + loopAllBuckets(loadKeysLoop, *mSnapshot); return result; } @@ -433,15 +400,8 @@ SearchableBucketListSnapshotBase::SearchableBucketListSnapshotBase( BucketSnapshotManager const& snapshotManager) : mSnapshotManager(snapshotManager), mHistoricalSnapshots() { - // Initialize snapshot from SnapshotManager - if constexpr (std::is_same_v) - { - mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); - } - else - { - mSnapshotManager.maybeUpdateHotArchiveSnapshot(mSnapshot); - } + + mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); } template @@ -462,7 +422,7 @@ SearchableHotArchiveBucketListSnapshot::SearchableHotArchiveBucketListSnapshot( } std::shared_ptr -SearchableHotArchiveBucketListSnapshot::getArchiveEntry(LedgerKey const& k) +SearchableHotArchiveBucketListSnapshot::load(LedgerKey const& k) { ZoneScoped; @@ -483,14 +443,23 @@ SearchableHotArchiveBucketListSnapshot::getArchiveEntry(LedgerKey const& k) }; // TODO: Metrics - mSnapshotManager.maybeUpdateHotArchiveSnapshot(mSnapshot); - loopAllBuckets(loadKeyBucketLoop); + mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); + loopAllBuckets(loadKeyBucketLoop, *mSnapshot); return result; } std::vector SearchableHotArchiveBucketListSnapshot::loadKeys( std::set const& inKeys) +{ + auto op = loadKeysFromLedger(inKeys, getLedgerSeq()); + releaseAssertOrThrow(op); + return std::move(*op); +} + +std::optional> +SearchableHotArchiveBucketListSnapshot::loadKeysFromLedger( + std::set const& inKeys, uint32_t ledgerSeq) { ZoneScoped; std::vector entries; @@ -502,10 +471,24 @@ SearchableHotArchiveBucketListSnapshot::loadKeys( return keys.empty(); }; - mSnapshotManager.maybeUpdateHotArchiveSnapshot(mSnapshot); + mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots); + + if (ledgerSeq == mSnapshot->getLedgerSeq()) + { + loopAllBuckets(loadKeysLoop, *mSnapshot); + } + else + { + auto iter = mHistoricalSnapshots.find(ledgerSeq); + if (iter == mHistoricalSnapshots.end()) + { + return std::nullopt; + } + + releaseAssert(iter->second); + loopAllBuckets(loadKeysLoop, *iter->second); + } - // TODO: Metrics - loopAllBuckets(loadKeysLoop); return entries; } diff --git a/src/bucket/BucketListSnapshot.h b/src/bucket/BucketListSnapshot.h index f7e2d66cc9..c4cd98450f 100644 --- a/src/bucket/BucketListSnapshot.h +++ b/src/bucket/BucketListSnapshot.h @@ -48,7 +48,7 @@ template class BucketListSnapshot : public NonMovable LedgerHeader const mHeader; public: - BucketListSnapshot(BucketListBase const& bl, LedgerHeader hhe); + BucketListSnapshot(BucketListBase const& bl, LedgerHeader hhe); // Only allow copies via constructor BucketListSnapshot(BucketListSnapshot const& snapshot); @@ -100,6 +100,15 @@ class SearchableBucketListSnapshotBase : public NonMovableOrCopyable SearchableBucketListSnapshotBase( BucketSnapshotManager const& snapshotManager); + + public: + uint32_t + getLedgerSeq() const + { + return mSnapshot->getLedgerSeq(); + } + + LedgerHeader const& getLedgerHeader(); }; class SearchableLiveBucketListSnapshot @@ -109,7 +118,7 @@ class SearchableLiveBucketListSnapshot BucketSnapshotManager const& snapshotManager); public: - std::shared_ptr getLedgerEntry(LedgerKey const& k); + std::shared_ptr load(LedgerKey const& k); std::vector loadKeysWithLimits(std::set const& inKeys, @@ -122,15 +131,13 @@ class SearchableLiveBucketListSnapshot std::vector loadInflationWinners(size_t maxWinners, int64_t minBalance); - std::shared_ptr load(LedgerKey const& k); - // Loads inKeys from the specified historical snapshot. Returns - // if the snapshot for the given ledger is - // available, otherwise. Note that ledgerSeq is defined + // load_result_vec if the snapshot for the given ledger is + // available, std::nullopt otherwise. Note that ledgerSeq is defined // as the state of the BucketList at the beginning of the ledger. This means // that for ledger N, the maximum lastModifiedLedgerSeq of any LedgerEntry // in the BucketList is N - 1. - std::pair, bool> + std::optional> loadKeysFromLedger(std::set const& inKeys, uint32_t ledgerSeq); @@ -140,8 +147,8 @@ class SearchableLiveBucketListSnapshot std::shared_ptr stats, StateArchivalSettings const& sas); - uint32_t getLedgerSeq() const; - LedgerHeader const& getLedgerHeader(); + friend std::shared_ptr + BucketSnapshotManager::copySearchableLiveBucketListSnapshot() const; }; class SearchableHotArchiveBucketListSnapshot @@ -151,12 +158,22 @@ class SearchableHotArchiveBucketListSnapshot BucketSnapshotManager const& snapshotManager); public: - std::shared_ptr getArchiveEntry(LedgerKey const& k); + std::shared_ptr load(LedgerKey const& k); std::vector loadKeys(std::set const& inKeys); + // Loads inKeys from the specified historical snapshot. Returns + // load_result_vec if the snapshot for the given ledger is + // available, std::nullopt otherwise. Note that ledgerSeq is defined + // as the state of the BucketList at the beginning of the ledger. This means + // that for ledger N, the maximum lastModifiedLedgerSeq of any LedgerEntry + // in the BucketList is N - 1. + std::optional> + loadKeysFromLedger(std::set const& inKeys, + uint32_t ledgerSeq); + friend std::shared_ptr - BucketSnapshotManager::getSearchableHotArchiveBucketListSnapshot() const; + BucketSnapshotManager::copySearchableHotArchiveBucketListSnapshot() const; }; } \ No newline at end of file diff --git a/src/bucket/BucketManager.h b/src/bucket/BucketManager.h index 5c928cdf8c..88e6280a46 100644 --- a/src/bucket/BucketManager.h +++ b/src/bucket/BucketManager.h @@ -285,8 +285,7 @@ class BucketManager : NonMovableOrCopyable std::vector const& liveEntries, std::vector const& deadEntries) = 0; virtual void - addHotArchiveBatch(Application& app, uint32_t currLedger, - uint32_t currLedgerProtocol, + addHotArchiveBatch(Application& app, LedgerHeader header, std::vector const& archivedEntries, std::vector const& restoredEntries, std::vector const& deletedEntries) = 0; @@ -402,8 +401,9 @@ class BucketManager : NonMovableOrCopyable virtual Config const& getConfig() const = 0; // Get bucketlist snapshot - virtual std::shared_ptr - getSearchableBucketListSnapshot() = 0; + virtual std::shared_ptr + getSearchableLiveBucketListSnapshot() = 0; + virtual void reportBucketEntryCountMetrics() = 0; }; } diff --git a/src/bucket/BucketManagerImpl.cpp b/src/bucket/BucketManagerImpl.cpp index 09a6eee274..dceb3c4463 100644 --- a/src/bucket/BucketManagerImpl.cpp +++ b/src/bucket/BucketManagerImpl.cpp @@ -135,9 +135,10 @@ BucketManagerImpl::initialize() mSnapshotManager = std::make_unique( mApp, std::make_unique>( - *mLiveBucketList, LedgerHeader()), + *mLiveBucketList, LedgerHeader()), std::make_unique>( - *mHotArchiveBucketList, LedgerHeader()), mApp.getConfig().QUERY_SNAPSHOT_LEDGERS); + *mHotArchiveBucketList, LedgerHeader()), + mApp.getConfig().QUERY_SNAPSHOT_LEDGERS); } } @@ -1027,10 +1028,9 @@ BucketManagerImpl::addLiveBatch(Application& app, LedgerHeader header, #endif auto timer = mBucketAddLiveBatch.TimeScope(); mBucketLiveObjectInsertBatch.Mark(initEntries.size() + liveEntries.size() + - deadEntries.size()); + deadEntries.size()); mLiveBucketList->addBatch(app, header.ledgerSeq, header.ledgerVersion, - initEntries, liveEntries, deadEntries); - + initEntries, liveEntries, deadEntries); mLiveBucketListSizeCounter.set_count(mLiveBucketList->getSize()); if (app.getConfig().isUsingBucketListDB()) @@ -1039,10 +1039,9 @@ BucketManagerImpl::addLiveBatch(Application& app, LedgerHeader header, } } -// TODO: Fix interface to match addLiveBatch void BucketManagerImpl::addHotArchiveBatch( - Application& app, uint32_t currLedger, uint32_t currLedgerProtocol, + Application& app, LedgerHeader header, std::vector const& archivedEntries, std::vector const& restoredEntries, std::vector const& deletedEntries) @@ -1050,12 +1049,12 @@ BucketManagerImpl::addHotArchiveBatch( ZoneScoped; releaseAssertOrThrow(app.getConfig().MODE_ENABLES_BUCKETLIST); releaseAssertOrThrow(protocolVersionStartsFrom( - currLedgerProtocol, + header.ledgerVersion, Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION)); #ifdef BUILD_TESTS if (mUseFakeTestValuesForNextClose) { - currLedgerProtocol = mFakeTestProtocolVersion; + header.ledgerVersion = mFakeTestProtocolVersion; } #endif auto timer = mBucketAddArchiveBatch.TimeScope(); @@ -1065,7 +1064,7 @@ BucketManagerImpl::addHotArchiveBatch( // Hot archive should never modify an existing entry, so there are never // live entries - mHotArchiveBucketList->addBatch(app, currLedger, currLedgerProtocol, + mHotArchiveBucketList->addBatch(app, header.ledgerSeq, header.ledgerVersion, archivedEntries, restoredEntries, deletedEntries); mArchiveBucketListSizeCounter.set_count(mHotArchiveBucketList->getSize()); @@ -1113,12 +1112,7 @@ BucketManagerImpl::snapshotLedger(LedgerHeader& currentHeader) Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION)) { // TODO: Hash Archive Bucket - // Holding off on this until buckets are written to history - - // SHA256 hasher; - // hasher.add(mLiveBucketList->getHash()); - // hasher.add(mHotArchiveBucketList->getHash()); - // hash = hasher.finish(); + // Dependency: HAS supports Hot Archive BucketList hash = mLiveBucketList->getHash(); } @@ -1171,7 +1165,8 @@ BucketManagerImpl::startBackgroundEvictionScan(uint32_t ledgerSeq) releaseAssert(!mEvictionFuture.valid()); releaseAssert(mEvictionStatistics); - auto searchableBL = mSnapshotManager->copySearchableBucketListSnapshot(); + auto searchableBL = + mSnapshotManager->copySearchableLiveBucketListSnapshot(); auto const& cfg = mApp.getLedgerManager().getSorobanNetworkConfig(); auto const& sas = cfg.stateArchivalSettings(); @@ -1188,7 +1183,7 @@ BucketManagerImpl::startBackgroundEvictionScan(uint32_t ledgerSeq) mEvictionFuture = task->get_future(); mApp.postOnEvictionBackgroundThread( bind(&task_t::operator(), task), - "SearchableBucketListSnapshot: eviction scan"); + "SearchableLiveBucketListSnapshot: eviction scan"); } void @@ -1328,6 +1323,7 @@ BucketManagerImpl::assumeState(HistoryArchiveState const& has, releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST); // TODO: Assume archival bucket state + // Dependency: HAS supports Hot Archive BucketList for (uint32_t i = 0; i < LiveBucketList::kNumLevels; ++i) { auto curr = @@ -1679,6 +1675,7 @@ BucketManagerImpl::scheduleVerifyReferencedBucketsWork() } // TODO: Update verify to for ArchiveBucket + // Dependency: HAS supports Hot Archive BucketList auto b = getBucketByHash(h); if (!b) { @@ -1698,8 +1695,8 @@ BucketManagerImpl::getConfig() const return mApp.getConfig(); } -std::shared_ptr -BucketManagerImpl::getSearchableBucketListSnapshot() +std::shared_ptr +BucketManagerImpl::getSearchableLiveBucketListSnapshot() { releaseAssert(mApp.getConfig().isUsingBucketListDB()); // Any other threads must maintain their own snapshot @@ -1707,7 +1704,7 @@ BucketManagerImpl::getSearchableBucketListSnapshot() if (!mSearchableBucketListSnapshot) { mSearchableBucketListSnapshot = - mSnapshotManager->copySearchableBucketListSnapshot(); + mSnapshotManager->copySearchableLiveBucketListSnapshot(); } return mSearchableBucketListSnapshot; @@ -1720,7 +1717,7 @@ BucketManagerImpl::reportBucketEntryCountMetrics() { return; } - auto bucketEntryCounters = mBucketList->sumBucketEntryCounters(); + auto bucketEntryCounters = mLiveBucketList->sumBucketEntryCounters(); for (auto [type, count] : bucketEntryCounters.entryTypeCounts) { auto countCounter = mBucketListEntryCountCounters.find(type); diff --git a/src/bucket/BucketManagerImpl.h b/src/bucket/BucketManagerImpl.h index 4f5f2a423e..8a59959afa 100644 --- a/src/bucket/BucketManagerImpl.h +++ b/src/bucket/BucketManagerImpl.h @@ -47,7 +47,7 @@ class BucketManagerImpl : public BucketManager std::unique_ptr mTmpDirManager; std::unique_ptr mWorkDir; std::map> mSharedBuckets; - std::shared_ptr + std::shared_ptr mSearchableBucketListSnapshot{}; // Lock for managing raw Bucket files or the bucket directory. This lock is @@ -187,8 +187,7 @@ class BucketManagerImpl : public BucketManager std::vector const& liveEntries, std::vector const& deadEntries) override; void - addHotArchiveBatch(Application& app, uint32_t currLedger, - uint32_t currLedgerProtocol, + addHotArchiveBatch(Application& app, LedgerHeader header, std::vector const& archivedEntries, std::vector const& restoredEntries, std::vector const& deletedEntries) override; @@ -243,8 +242,9 @@ class BucketManagerImpl : public BucketManager Config const& getConfig() const override; - std::shared_ptr - getSearchableBucketListSnapshot() override; + std::shared_ptr + getSearchableLiveBucketListSnapshot() override; + void reportBucketEntryCountMetrics() override; }; diff --git a/src/bucket/BucketSnapshotManager.cpp b/src/bucket/BucketSnapshotManager.cpp index 98cea9b197..703da5c21f 100644 --- a/src/bucket/BucketSnapshotManager.cpp +++ b/src/bucket/BucketSnapshotManager.cpp @@ -3,6 +3,7 @@ // of this distribution or at http://www.apache.org/licenses/LICENSE-2.0 #include "bucket/BucketSnapshotManager.h" +#include "bucket/Bucket.h" #include "bucket/BucketListSnapshot.h" #include "main/Application.h" #include "util/GlobalChecks.h" @@ -16,16 +17,17 @@ namespace stellar { BucketSnapshotManager::BucketSnapshotManager( - Application& app, std::unique_ptr const>&& snapshot, + Application& app, + std::unique_ptr const>&& snapshot, std::unique_ptr const>&& hotArchiveSnapshot, - uint32_t numHistoricalSnapshots) + uint32_t numLiveHistoricalSnapshots) : mApp(app) - , mCurrentSnapshot(std::move(snapshot)) - , mCurrentHotArchiveSnapshot(std::move(hotArchiveSnapshot)) - , mHistoricalSnapshots() - // TODO: ARchival snapshots - , mNumHistoricalSnapshots(numHistoricalSnapshots) + , mCurrLiveSnapshot(std::move(snapshot)) + , mCurrHotArchiveSnapshot(std::move(hotArchiveSnapshot)) + , mLiveHistoricalSnapshots() + , mHotArchiveHistoricalSnapshots() + , mNumHistoricalSnapshots(numLiveHistoricalSnapshots) , mBulkLoadMeter(app.getMetrics().NewMeter( {"bucketlistDB", "query", "loads"}, "query")) , mBloomMisses(app.getMetrics().NewMeter( @@ -34,12 +36,12 @@ BucketSnapshotManager::BucketSnapshotManager( {"bucketlistDB", "bloom", "lookups"}, "bloom")) { releaseAssert(threadIsMain()); - releaseAssert(mCurrentSnapshot); - releaseAssert(mCurrentHotArchiveSnapshot); + releaseAssert(mCurrLiveSnapshot); + releaseAssert(mCurrHotArchiveSnapshot); } std::shared_ptr -BucketSnapshotManager::copySearchableBucketListSnapshot() const +BucketSnapshotManager::copySearchableLiveBucketListSnapshot() const { // Can't use std::make_shared due to private constructor return std::shared_ptr( @@ -47,9 +49,9 @@ BucketSnapshotManager::copySearchableBucketListSnapshot() const } std::shared_ptr -BucketSnapshotManager::getSearchableHotArchiveBucketListSnapshot() const +BucketSnapshotManager::copySearchableHotArchiveBucketListSnapshot() const { - releaseAssert(mCurrentHotArchiveSnapshot); + releaseAssert(mCurrHotArchiveSnapshot); // Can't use std::make_shared due to private constructor return std::shared_ptr( new SearchableHotArchiveBucketListSnapshot(*this)); @@ -79,12 +81,43 @@ BucketSnapshotManager::recordBulkLoadMetrics(std::string const& label, return iter->second; } +template void BucketSnapshotManager::maybeUpdateSnapshot( - std::unique_ptr const>& snapshot, - std::map const>>& - historicalSnapshots) const + std::unique_ptr& snapshot, + std::map>& historicalSnapshots) + const { + static_assert( + std::is_same_v> || + std::is_same_v>); + + auto const& managerSnapshot = [&]() -> auto const& + { + if constexpr (std::is_same_v>) + { + return mCurrLiveSnapshot; + } + else + { + return mCurrHotArchiveSnapshot; + } + } + (); + + auto const& managerHistoricalSnapshots = [&]() -> auto const& + { + if constexpr (std::is_same_v>) + { + return mLiveHistoricalSnapshots; + } + else + { + return mHotArchiveHistoricalSnapshots; + } + } + (); + // The canonical snapshot held by the BucketSnapshotManager is not being // modified. Rather, a thread is checking it's copy against the canonical // snapshot, so use a shared lock. @@ -92,95 +125,74 @@ BucketSnapshotManager::maybeUpdateSnapshot( // First update current snapshot if (!snapshot || - snapshot->getLedgerSeq() != mCurrentSnapshot->getLedgerSeq()) + snapshot->getLedgerSeq() != managerSnapshot->getLedgerSeq()) { // Should only update with a newer snapshot releaseAssert(!snapshot || snapshot->getLedgerSeq() < - mCurrentSnapshot->getLedgerSeq()); - snapshot = - std::make_unique>(*mCurrentSnapshot); + managerSnapshot->getLedgerSeq()); + snapshot = std::make_unique(*managerSnapshot); } // Then update historical snapshots (if any exist) - if (mHistoricalSnapshots.empty()) + if (managerHistoricalSnapshots.empty()) { return; } // If size of manager's history map is different, or if the oldest snapshot // ledger seq is different, we need to update. - if (mHistoricalSnapshots.size() != historicalSnapshots.size() || - mHistoricalSnapshots.begin()->first != + if (managerHistoricalSnapshots.size() != historicalSnapshots.size() || + managerHistoricalSnapshots.begin()->first != historicalSnapshots.begin()->first) { // Copy current snapshot map into historicalSnapshots historicalSnapshots.clear(); - for (auto const& [ledgerSeq, snap] : mHistoricalSnapshots) + for (auto const& [ledgerSeq, snap] : managerHistoricalSnapshots) { - historicalSnapshots.emplace( - ledgerSeq, std::make_unique(*snap)); + historicalSnapshots.emplace(ledgerSeq, + std::make_unique(*snap)); } } } -void -BucketSnapshotManager::maybeUpdateHotArchiveSnapshot( - std::unique_ptr const>& snapshot) const -{ - std::lock_guard lock(mSnapshotMutex); - if (!snapshot || - snapshot->getLedgerSeq() != mCurrentHotArchiveSnapshot->getLedgerSeq()) - { - // Should only update with a newer snapshot - releaseAssert(!snapshot || - snapshot->getLedgerSeq() < - mCurrentHotArchiveSnapshot->getLedgerSeq()); - snapshot = std::make_unique>( - *mCurrentHotArchiveSnapshot); - } -} - void BucketSnapshotManager::updateCurrentSnapshot( - std::unique_ptr const>&& newSnapshot) + std::unique_ptr const>&& liveSnapshot, + std::unique_ptr const>&& + hotArchiveSnapshot) { - releaseAssert(newSnapshot); releaseAssert(threadIsMain()); - // Updating the BucketSnapshotManager canonical snapshot, must lock - // exclusively for write access. - std::unique_lock lock(mSnapshotMutex); - releaseAssert(!mCurrentSnapshot || newSnapshot->getLedgerSeq() >= - mCurrentSnapshot->getLedgerSeq()); + auto updateSnapshot = [numHistoricalSnapshots = mNumHistoricalSnapshots]( + auto& currentSnapshot, auto& historicalSnapshots, + auto&& newSnapshot) { + releaseAssert(newSnapshot); + releaseAssert(!currentSnapshot || newSnapshot->getLedgerSeq() >= + currentSnapshot->getLedgerSeq()); - // First update historical snapshots - if (mNumHistoricalSnapshots != 0) - { - // If historical snapshots are full, delete the oldest one - if (mHistoricalSnapshots.size() == mNumHistoricalSnapshots) + // First update historical snapshots + if (numHistoricalSnapshots != 0) { - mHistoricalSnapshots.erase(mHistoricalSnapshots.begin()); + // If historical snapshots are full, delete the oldest one + if (historicalSnapshots.size() == numHistoricalSnapshots) + { + historicalSnapshots.erase(historicalSnapshots.begin()); + } + + historicalSnapshots.emplace(currentSnapshot->getLedgerSeq(), + std::move(currentSnapshot)); + currentSnapshot = nullptr; } - mHistoricalSnapshots.emplace(mCurrentSnapshot->getLedgerSeq(), - std::move(mCurrentSnapshot)); - mCurrentSnapshot = nullptr; - } + currentSnapshot.swap(newSnapshot); + }; - mCurrentSnapshot.swap(newSnapshot); -} - -void -BucketSnapshotManager::updateCurrentHotArchiveSnapshot( - std::unique_ptr const>&& newSnapshot) -{ - releaseAssert(newSnapshot); - releaseAssert(threadIsMain()); - std::lock_guard lock(mSnapshotMutex); - releaseAssert(!mCurrentHotArchiveSnapshot || - newSnapshot->getLedgerSeq() >= - mCurrentHotArchiveSnapshot->getLedgerSeq()); - mCurrentHotArchiveSnapshot.swap(newSnapshot); + // Updating the BucketSnapshotManager canonical snapshot, must lock + // exclusively for write access. + std::unique_lock lock(mSnapshotMutex); + updateSnapshot(mCurrLiveSnapshot, mLiveHistoricalSnapshots, liveSnapshot); + updateSnapshot(mCurrHotArchiveSnapshot, mHotArchiveHistoricalSnapshots, + hotArchiveSnapshot); } void @@ -217,4 +229,16 @@ BucketSnapshotManager::endPointLoadTimer(LedgerEntryType t, iter->second.Update(duration); } } + +template void +BucketSnapshotManager::maybeUpdateSnapshot>( + std::unique_ptr const>& snapshot, + std::map const>>& + historicalSnapshots) const; +template void BucketSnapshotManager::maybeUpdateSnapshot< + BucketListSnapshot>( + std::unique_ptr const>& snapshot, + std::map const>>& + historicalSnapshots) const; } \ No newline at end of file diff --git a/src/bucket/BucketSnapshotManager.h b/src/bucket/BucketSnapshotManager.h index 2d1dd6264e..de44f6f165 100644 --- a/src/bucket/BucketSnapshotManager.h +++ b/src/bucket/BucketSnapshotManager.h @@ -41,18 +41,20 @@ class BucketSnapshotManager : NonMovableOrCopyable // Snapshot that is maintained and periodically updated by BucketManager on // the main thread. When background threads need to generate or refresh a // snapshot, they will copy this snapshot. - std::unique_ptr const> mCurrentSnapshot{}; + std::unique_ptr const> mCurrLiveSnapshot{}; std::unique_ptr const> - mCurrentHotArchiveSnapshot{}; + mCurrHotArchiveSnapshot{}; // ledgerSeq that the snapshot is based on -> snapshot std::map const>> - mHistoricalSnapshots; + mLiveHistoricalSnapshots; + std::map const>> + mHotArchiveHistoricalSnapshots; uint32_t const mNumHistoricalSnapshots; - // Lock must be held when accessing mCurrentSnapshot and - // mHistoricalSnapshots + // Lock must be held when accessing any snapshot mutable std::shared_mutex mSnapshotMutex; mutable UnorderedMap mPointTimers{}; @@ -65,38 +67,35 @@ class BucketSnapshotManager : NonMovableOrCopyable mutable std::optional mTimerStart; public: - // Called by main thread to update mCurrentSnapshot whenever the BucketList + // Called by main thread to update snapshots whenever the BucketList // is updated void updateCurrentSnapshot( - std::unique_ptr const>&& newSnapshot); + std::unique_ptr const>&& liveSnapshot, + std::unique_ptr const>&& + hotArchiveSnapshot); // numHistoricalLedgers is the number of historical snapshots that the // snapshot manager will maintain. If numHistoricalLedgers is 5, snapshots // will be capable of querying state from ledger [lcl, lcl - 5]. - BucketSnapshotManager(Application& app, - std::unique_ptr const>&& snapshot, std::unique_ptr const>&& - hotArchiveSnapshot, - uint32_t numHistoricalLedgers); - void updateCurrentHotArchiveSnapshot( + BucketSnapshotManager( + Application& app, + std::unique_ptr const>&& snapshot, std::unique_ptr const>&& - newSnapshot); + hotArchiveSnapshot, + uint32_t numHistoricalLedgers); std::shared_ptr - copySearchableBucketListSnapshot() const; + copySearchableLiveBucketListSnapshot() const; std::shared_ptr - getSearchableHotArchiveBucketListSnapshot() const; - - // Checks if snapshot is out of date with mCurrentSnapshot and updates - // it accordingly - void maybeUpdateSnapshot( - std::unique_ptr const>& snapshot, - std::map const>>& - historicalSnapshots) const; - - void maybeUpdateHotArchiveSnapshot( - std::unique_ptr const>& snapshot) - const; + copySearchableHotArchiveBucketListSnapshot() const; + + // Checks if snapshot is out of date and updates it accordingly + template + void + maybeUpdateSnapshot(std::unique_ptr& snapshot, + std::map>& + historicalSnapshots) const; // All metric recording functions must only be called by the main thread void startPointLoadTimer() const; diff --git a/src/bucket/FutureBucket.cpp b/src/bucket/FutureBucket.cpp index f6de0d530e..bc2dec6d16 100644 --- a/src/bucket/FutureBucket.cpp +++ b/src/bucket/FutureBucket.cpp @@ -62,7 +62,8 @@ FutureBucket::FutureBucket( if constexpr (!std::is_same_v) { - if (protocolVersionIsBefore( + if (!snap->isEmpty() && + protocolVersionIsBefore( snap->getBucketVersion(), Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION)) { diff --git a/src/bucket/LedgerCmp.h b/src/bucket/LedgerCmp.h index 06f24eee7a..cc550a4e96 100644 --- a/src/bucket/LedgerCmp.h +++ b/src/bucket/LedgerCmp.h @@ -172,7 +172,7 @@ template struct BucketEntryIdCmp if (aty != HOT_ARCHIVE_DELETED && aty != HOT_ARCHIVE_LIVE) { throw std::runtime_error( - "Malformed bucket: expected DELETED/RESTORED key."); + "Malformed bucket: expected DELETED/LIVE key."); } if (bty == HOT_ARCHIVE_ARCHIVED) diff --git a/src/bucket/test/BucketIndexTests.cpp b/src/bucket/test/BucketIndexTests.cpp index 8c3e0a0242..ae3af67dc6 100644 --- a/src/bucket/test/BucketIndexTests.cpp +++ b/src/bucket/test/BucketIndexTests.cpp @@ -135,7 +135,7 @@ class BucketIndexTest auto searchableBL = getBM() .getBucketSnapshotManager() - .copySearchableBucketListSnapshot(); + .copySearchableLiveBucketListSnapshot(); auto lk = LedgerEntryKey(canonicalEntry); auto currentLoadedEntry = searchableBL->load(lk); @@ -149,20 +149,18 @@ class BucketIndexTest for (uint32_t currLedger = ledger; currLedger > 0; --currLedger) { - auto [loadRes, snapshotExists] = - searchableBL->loadKeysFromLedger({lk}, currLedger); + auto loadRes = searchableBL->loadKeysFromLedger({lk}, currLedger); // If we query an older snapshot, should return if (currLedger < ledger - mApp->getConfig().QUERY_SNAPSHOT_LEDGERS) { - REQUIRE(!snapshotExists); - REQUIRE(loadRes.empty()); + REQUIRE(!loadRes); } else { - REQUIRE(snapshotExists); - REQUIRE(loadRes.size() == 1); - REQUIRE(loadRes[0].lastModifiedLedgerSeq == currLedger - 1); + REQUIRE(loadRes); + REQUIRE(loadRes->size() == 1); + REQUIRE(loadRes->at(0).lastModifiedLedgerSeq == currLedger - 1); } } } @@ -256,7 +254,7 @@ class BucketIndexTest { auto searchableBL = getBM() .getBucketSnapshotManager() - .copySearchableBucketListSnapshot(); + .copySearchableLiveBucketListSnapshot(); // Test bulk load lookup auto loadResult = @@ -283,7 +281,7 @@ class BucketIndexTest { auto searchableBL = getBM() .getBucketSnapshotManager() - .copySearchableBucketListSnapshot(); + .copySearchableLiveBucketListSnapshot(); for (size_t i = 0; i < n; ++i) { LedgerKeySet searchSubset; @@ -323,7 +321,7 @@ class BucketIndexTest { auto searchableBL = getBM() .getBucketSnapshotManager() - .copySearchableBucketListSnapshot(); + .copySearchableLiveBucketListSnapshot(); // Load should return empty vector for keys not in bucket list auto keysNotInBL = @@ -500,7 +498,7 @@ class BucketIndexPoolShareTest : public BucketIndexTest { auto searchableBL = getBM() .getBucketSnapshotManager() - .copySearchableBucketListSnapshot(); + .copySearchableLiveBucketListSnapshot(); auto loadResult = searchableBL->loadPoolShareTrustLinesByAccountAndAsset( mAccountToSearch.accountID, mAssetToSearch); @@ -708,7 +706,7 @@ TEST_CASE("hot archive bucket lookups", "[bucket][bucketindex][archive]") // Use snapshot across ledger to test update behavior auto searchableBL = app->getBucketManager() .getBucketSnapshotManager() - .getSearchableHotArchiveBucketListSnapshot(); + .copySearchableHotArchiveBucketListSnapshot(); auto checkLoad = [&](LedgerKey const& k, std::shared_ptr entryPtr) { @@ -746,7 +744,7 @@ TEST_CASE("hot archive bucket lookups", "[bucket][bucketindex][archive]") LedgerKeySet bulkLoadKeys; for (auto const& k : keysToSearch) { - auto entryPtr = searchableBL->getArchiveEntry(k); + auto entryPtr = searchableBL->load(k); checkLoad(k, entryPtr); bulkLoadKeys.emplace(k); } @@ -804,37 +802,34 @@ TEST_CASE("hot archive bucket lookups", "[bucket][bucketindex][archive]") expectedRestoredEntries.emplace(k); } - app->getBucketManager().addHotArchiveBatch( - *app, ledger++, - static_cast( - Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION), - archivedEntries, restoredEntries, deletedEntries); + auto header = + app->getLedgerManager().getLastClosedLedgerHeader().header; + header.ledgerSeq += 1; + header.ledgerVersion = static_cast( + Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION); + addHotArchiveBatchAndUpdateSnapshot(*app, header, archivedEntries, + restoredEntries, deletedEntries); checkResult(); // Add a few batches so that entries are no longer in the top bucket for (auto i = 0; i < 100; ++i) { - app->getBucketManager().addHotArchiveBatch( - *app, 1, - static_cast( - Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION), - {}, {}, {}); + header.ledgerSeq += 1; + addHotArchiveBatchAndUpdateSnapshot(*app, header, {}, {}, {}); } // Shadow entries via liveEntry auto liveShadow1 = LedgerEntryKey(archivedEntries[0]); auto liveShadow2 = deletedEntries[1]; - app->getBucketManager().addHotArchiveBatch( - *app, ledger++, - static_cast( - Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION), - {}, {liveShadow1, liveShadow2}, {}); + header.ledgerSeq += 1; + addHotArchiveBatchAndUpdateSnapshot(*app, header, {}, + {liveShadow1, liveShadow2}, {}); // Point load for (auto const& k : {liveShadow1, liveShadow2}) { - auto entryPtr = searchableBL->getArchiveEntry(k); + auto entryPtr = searchableBL->load(k); REQUIRE(!entryPtr); } @@ -845,14 +840,13 @@ TEST_CASE("hot archive bucket lookups", "[bucket][bucketindex][archive]") // Shadow via deletedEntry auto deletedShadow = LedgerEntryKey(archivedEntries[1]); - app->getBucketManager().addHotArchiveBatch( - *app, ledger++, - static_cast( - Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION), - {}, {}, {deletedShadow}); + + header.ledgerSeq += 1; + addHotArchiveBatchAndUpdateSnapshot(*app, header, {}, {}, + {deletedShadow}); // Point load - auto entryPtr = searchableBL->getArchiveEntry(deletedShadow); + auto entryPtr = searchableBL->load(deletedShadow); REQUIRE(entryPtr); REQUIRE(entryPtr->type() == HotArchiveBucketEntryType::HOT_ARCHIVE_DELETED); @@ -868,15 +862,12 @@ TEST_CASE("hot archive bucket lookups", "[bucket][bucketindex][archive]") auto archivedShadow = archivedEntries[3]; archivedShadow.lastModifiedLedgerSeq = ledger; - app->getBucketManager().addHotArchiveBatch( - *app, ledger++, - static_cast( - Bucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION), - {archivedShadow}, {}, {}); + header.ledgerSeq += 1; + addHotArchiveBatchAndUpdateSnapshot(*app, header, {archivedShadow}, {}, + {}); // Point load - entryPtr = - searchableBL->getArchiveEntry(LedgerEntryKey(archivedShadow)); + entryPtr = searchableBL->load(LedgerEntryKey(archivedShadow)); REQUIRE(entryPtr); REQUIRE(entryPtr->type() == HotArchiveBucketEntryType::HOT_ARCHIVE_ARCHIVED); diff --git a/src/bucket/test/BucketListTests.cpp b/src/bucket/test/BucketListTests.cpp index 3bfbcf3b23..d3bb8a2a9b 100644 --- a/src/bucket/test/BucketListTests.cpp +++ b/src/bucket/test/BucketListTests.cpp @@ -150,20 +150,8 @@ basicBucketListTest() !app->getClock().getIOContext().stopped() && i < 130; ++i) { app->getClock().crank(false); - auto lh = - app->getLedgerManager().getLastClosedLedgerHeader().header; - lh.ledgerSeq = i; - // addBatchAndUpdateSnapshot( - // bl, *app, lh, {}, - // LedgerTestUtils:: - // generateValidUniqueLedgerEntriesWithExclusions( - // {CONFIG_SETTING}, 8), - // LedgerTestUtils::generateValidLedgerEntryKeysWithExclusions( - // {CONFIG_SETTING}, 5)); if constexpr (std::is_same_v) { - - // TODO: addBatchAndUpdateSnapshot bl.addBatch( *app, i, getAppLedgerVersion(app), {}, LedgerTestUtils::generateValidUniqueLedgerEntries(8), @@ -173,7 +161,6 @@ basicBucketListTest() } else { - // TODO: addBatchAndUpdateSnapshot bl.addBatch( *app, i, getAppLedgerVersion(app), {}, LedgerTestUtils::generateValidUniqueLedgerKeysWithTypes( @@ -209,7 +196,7 @@ basicBucketListTest() } else { - for_versions_from(22, cfg, test); + for_versions_from(23, cfg, test); } } @@ -324,8 +311,7 @@ TEST_CASE_VERSIONS("bucket list shadowing pre/post proto 12", { app->getClock().crank(false); auto liveBatch = - LedgerTestUtils::generateValidUniqueLedgerEntriesWithExclusions( - {CONFIG_SETTING}, 5); + LedgerTestUtils::generateValidUniqueLedgerEntries(5); BucketEntry BucketEntryAlice, BucketEntryBob; alice.balance++; @@ -340,11 +326,8 @@ TEST_CASE_VERSIONS("bucket list shadowing pre/post proto 12", BucketEntryBob.liveEntry().data.account() = bob; liveBatch.push_back(BucketEntryBob.liveEntry()); - auto lh = - app->getLedgerManager().getLastClosedLedgerHeader().header; - lh.ledgerSeq = i; - addBatchAndUpdateSnapshot( - bl, *app, lh, {}, liveBatch, + bl.addBatch( + *app, i, getAppLedgerVersion(app), {}, liveBatch, LedgerTestUtils::generateValidLedgerEntryKeysWithExclusions( {CONFIG_SETTING}, 5)); if (i % 100 == 0) @@ -411,7 +394,7 @@ TEST_CASE_VERSIONS("hot archive bucket tombstones expire at bottom level", testutil::BucketListDepthModifier bldm(5); auto app = createTestApplication(clock, cfg); - for_versions_from(22, *app, [&] { + for_versions_from(23, *app, [&] { HotArchiveBucketList bl; auto lastSnapSize = [&] { @@ -487,16 +470,14 @@ TEST_CASE_VERSIONS("live bucket tombstones expire at bottom level", auto& level = bl.getLevel(i); level.setCurr(LiveBucket::fresh( bm, getAppLedgerVersion(app), {}, - LedgerTestUtils::generateValidUniqueLedgerEntriesWithExclusions( - {CONFIG_SETTING}, 8), + LedgerTestUtils::generateValidUniqueLedgerEntries(8), LedgerTestUtils::generateValidLedgerEntryKeysWithExclusions( {CONFIG_SETTING}, 5), /*countMergeEvents=*/true, clock.getIOContext(), /*doFsync=*/true)); level.setSnap(LiveBucket::fresh( bm, getAppLedgerVersion(app), {}, - LedgerTestUtils::generateValidUniqueLedgerEntriesWithExclusions( - {CONFIG_SETTING}, 8), + LedgerTestUtils::generateValidUniqueLedgerEntries(8), LedgerTestUtils::generateValidLedgerEntryKeysWithExclusions( {CONFIG_SETTING}, 5), /*countMergeEvents=*/true, clock.getIOContext(), @@ -510,14 +491,9 @@ TEST_CASE_VERSIONS("live bucket tombstones expire at bottom level", for (auto j : ledgers) { auto n = mergeTimer.count(); - auto lh = - app->getLedgerManager().getLastClosedLedgerHeader().header; - lh.ledgerSeq = j; - addBatchAndUpdateSnapshot( - bl, *app, lh, {}, - LedgerTestUtils:: - generateValidUniqueLedgerEntriesWithExclusions( - {CONFIG_SETTING}, 8), + bl.addBatch( + *app, j, getAppLedgerVersion(app), {}, + LedgerTestUtils::generateValidUniqueLedgerEntries(8), LedgerTestUtils::generateValidLedgerEntryKeysWithExclusions( {CONFIG_SETTING}, 5)); app->getClock().crank(false); @@ -556,6 +532,7 @@ TEST_CASE_VERSIONS("bucket tombstones mutually-annihilate init entries", for_versions_with_differing_bucket_logic(cfg, [&](Config const& cfg) { Application::pointer app = createTestApplication(clock, cfg); LiveBucketList bl; + auto vers = getAppLedgerVersion(app); autocheck::generator flip; std::deque entriesToModify; for (uint32_t i = 1; i < 512; ++i) @@ -591,11 +568,7 @@ TEST_CASE_VERSIONS("bucket tombstones mutually-annihilate init entries", deadEntries.push_back(LedgerEntryKey(e)); } } - auto lh = - app->getLedgerManager().getLastClosedLedgerHeader().header; - lh.ledgerSeq = i; - addBatchAndUpdateSnapshot(bl, *app, lh, initEntries, liveEntries, - deadEntries); + bl.addBatch(*app, i, vers, initEntries, liveEntries, deadEntries); app->getClock().crank(false); for (uint32_t k = 0u; k < LiveBucketList::kNumLevels; ++k) { @@ -640,23 +613,17 @@ TEST_CASE_VERSIONS("single entry bubbling up", std::vector emptySetEntry; CLOG_DEBUG(Bucket, "Adding single entry in lowest level"); - addBatchAndUpdateSnapshot( - bl, *app, - app->getLedgerManager().getLastClosedLedgerHeader().header, {}, - LedgerTestUtils::generateValidLedgerEntriesWithExclusions( - {CONFIG_SETTING}, 1), - emptySet); + bl.addBatch(*app, 1, getAppLedgerVersion(app), {}, + LedgerTestUtils::generateValidLedgerEntries(1), + emptySet); CLOG_DEBUG(Bucket, "Adding empty batches to bucket list"); for (uint32_t i = 2; !app->getClock().getIOContext().stopped() && i < 300; ++i) { app->getClock().crank(false); - auto lh = - app->getLedgerManager().getLastClosedLedgerHeader().header; - lh.ledgerSeq = i; - addBatchAndUpdateSnapshot(bl, *app, lh, {}, emptySetEntry, - emptySet); + bl.addBatch(*app, i, getAppLedgerVersion(app), {}, + emptySetEntry, emptySet); if (i % 10 == 0) CLOG_DEBUG(Bucket, "Added batch {}, hash={}", i, binToHex(bl.getHash())); @@ -879,8 +846,8 @@ TEST_CASE("BucketList check bucket sizes", "[bucket][bucketlist][count]") auto lh = app->getLedgerManager().getLastClosedLedgerHeader().header; lh.ledgerSeq = ledgerSeq; - addBatchAndUpdateSnapshot(bl, *app, lh, {}, - {ledgers[ledgerSeq - 1]}, emptySet); + addLiveBatchAndUpdateSnapshot(*app, lh, {}, + {ledgers[ledgerSeq - 1]}, emptySet); } for (uint32_t level = 0; level < LiveBucketList::kNumLevels; ++level) { @@ -1471,7 +1438,7 @@ TEST_CASE_VERSIONS("Searchable BucketListDB snapshots", "[bucketlist]") entry.data.claimableBalance().amount = 0; auto searchableBL = - bm.getBucketSnapshotManager().copySearchableBucketListSnapshot(); + bm.getBucketSnapshotManager().copySearchableLiveBucketListSnapshot(); // Update entry every 5 ledgers so we can see bucket merge events for (auto ledgerSeq = 1; ledgerSeq < 101; ++ledgerSeq) diff --git a/src/bucket/test/BucketManagerTests.cpp b/src/bucket/test/BucketManagerTests.cpp index 70ba290f2f..f2a500f6e2 100644 --- a/src/bucket/test/BucketManagerTests.cpp +++ b/src/bucket/test/BucketManagerTests.cpp @@ -365,8 +365,8 @@ TEST_CASE_VERSIONS("bucketmanager reattach to finished merge", auto lh = app->getLedgerManager().getLastClosedLedgerHeader().header; lh.ledgerSeq = ledger; - addBatchAndUpdateSnapshot( - bl, *app, lh, {}, + addLiveBatchAndUpdateSnapshot( + *app, lh, {}, LedgerTestUtils::generateValidLedgerEntriesWithExclusions( {CONFIG_SETTING}, 10), {}); @@ -454,8 +454,8 @@ TEST_CASE_VERSIONS("bucketmanager reattach to running merge", auto lh = app->getLedgerManager().getLastClosedLedgerHeader().header; lh.ledgerSeq = ledger; - addBatchAndUpdateSnapshot( - bl, *app, lh, {}, + addLiveBatchAndUpdateSnapshot( + *app, lh, {}, LedgerTestUtils::generateValidUniqueLedgerEntriesWithExclusions( {CONFIG_SETTING}, 100), {}); @@ -580,7 +580,6 @@ TEST_CASE_VERSIONS( auto vers = getAppLedgerVersion(app); auto& hm = app->getHistoryManager(); auto& bm = app->getBucketManager(); - auto& bl = bm.getLiveBucketList(); hm.setPublicationEnabled(false); app->getHistoryArchiveManager().initializeHistoryArchive( tcfg.getArchiveDirName()); @@ -596,8 +595,8 @@ TEST_CASE_VERSIONS( auto lh = app->getLedgerManager().getLastClosedLedgerHeader().header; lh.ledgerSeq++; - addBatchAndUpdateSnapshot( - bl, *app, lh, {}, + addLiveBatchAndUpdateSnapshot( + *app, lh, {}, LedgerTestUtils::generateValidUniqueLedgerEntriesWithExclusions( {CONFIG_SETTING}, 100), {}); diff --git a/src/bucket/test/BucketTestUtils.cpp b/src/bucket/test/BucketTestUtils.cpp index 9d72c65ab6..d39c5384b2 100644 --- a/src/bucket/test/BucketTestUtils.cpp +++ b/src/bucket/test/BucketTestUtils.cpp @@ -12,6 +12,7 @@ #include "main/Application.h" #include "test/test.h" #include "xdr/Stellar-ledger.h" +#include namespace stellar { @@ -32,18 +33,47 @@ getAppLedgerVersion(Application::pointer app) } void -addBatchAndUpdateSnapshot(LiveBucketList& bl, Application& app, - LedgerHeader header, - std::vector const& initEntries, - std::vector const& liveEntries, - std::vector const& deadEntries) +addLiveBatchAndUpdateSnapshot(Application& app, LedgerHeader header, + std::vector const& initEntries, + std::vector const& liveEntries, + std::vector const& deadEntries) { - bl.addBatch(app, header.ledgerSeq, header.ledgerVersion, initEntries, - liveEntries, deadEntries); + auto& liveBl = app.getBucketManager().getLiveBucketList(); + liveBl.addBatch(app, header.ledgerSeq, header.ledgerVersion, initEntries, + liveEntries, deadEntries); if (app.getConfig().isUsingBucketListDB()) { + auto liveSnapshot = + std::make_unique>(liveBl, header); + auto hotArchiveSnapshot = + std::make_unique>( + app.getBucketManager().getHotArchiveBucketList(), header); + + app.getBucketManager().getBucketSnapshotManager().updateCurrentSnapshot( + std::move(liveSnapshot), std::move(hotArchiveSnapshot)); + } +} + +void +addHotArchiveBatchAndUpdateSnapshot( + Application& app, LedgerHeader header, + std::vector const& archiveEntries, + std::vector const& restoredEntries, + std::vector const& deletedEntries) +{ + auto& hotArchiveBl = app.getBucketManager().getHotArchiveBucketList(); + hotArchiveBl.addBatch(app, header.ledgerSeq, header.ledgerVersion, + archiveEntries, restoredEntries, deletedEntries); + if (app.getConfig().isUsingBucketListDB()) + { + auto liveSnapshot = std::make_unique>( + app.getBucketManager().getLiveBucketList(), header); + auto hotArchiveSnapshot = + std::make_unique>(hotArchiveBl, + header); + app.getBucketManager().getBucketSnapshotManager().updateCurrentSnapshot( - std::make_unique(bl, header)); + std::move(liveSnapshot), std::move(hotArchiveSnapshot)); } } diff --git a/src/bucket/test/BucketTestUtils.h b/src/bucket/test/BucketTestUtils.h index 14f567625e..c77794a80a 100644 --- a/src/bucket/test/BucketTestUtils.h +++ b/src/bucket/test/BucketTestUtils.h @@ -12,11 +12,16 @@ namespace stellar namespace BucketTestUtils { -void addBatchAndUpdateSnapshot(LiveBucketList& bl, Application& app, - LedgerHeader header, - std::vector const& initEntries, - std::vector const& liveEntries, - std::vector const& deadEntries); +void addLiveBatchAndUpdateSnapshot(Application& app, LedgerHeader header, + std::vector const& initEntries, + std::vector const& liveEntries, + std::vector const& deadEntries); + +void addHotArchiveBatchAndUpdateSnapshot( + Application& app, LedgerHeader header, + std::vector const& archiveEntries, + std::vector const& restoredEntries, + std::vector const& deletedEntries); uint32_t getAppLedgerVersion(Application& app); diff --git a/src/bucket/test/BucketTests.cpp b/src/bucket/test/BucketTests.cpp index 1155a981bb..841d0ca6e6 100644 --- a/src/bucket/test/BucketTests.cpp +++ b/src/bucket/test/BucketTests.cpp @@ -279,7 +279,7 @@ TEST_CASE_VERSIONS("merging hot archive bucket entries", "[bucket][archival]") Config const& cfg = getTestConfig(); auto app = createTestApplication(clock, cfg); - for_versions_from(22, *app, [&] { + for_versions_from(23, *app, [&] { auto& bm = app->getBucketManager(); auto vers = getAppLedgerVersion(app); diff --git a/src/history/test/HistoryTests.cpp b/src/history/test/HistoryTests.cpp index 68bc81e4a5..bb4d4c92be 100644 --- a/src/history/test/HistoryTests.cpp +++ b/src/history/test/HistoryTests.cpp @@ -1318,8 +1318,8 @@ TEST_CASE_VERSIONS( { auto lcl = lm.getLastClosedLedgerHeader(); lcl.header.ledgerSeq += 1; - BucketTestUtils::addBatchAndUpdateSnapshot( - bl, *app, lcl.header, {}, + BucketTestUtils::addLiveBatchAndUpdateSnapshot( + *app, lcl.header, {}, LedgerTestUtils::generateValidUniqueLedgerEntries(8), {}); clock.crank(true); } diff --git a/src/invariant/test/BucketListIsConsistentWithDatabaseTests.cpp b/src/invariant/test/BucketListIsConsistentWithDatabaseTests.cpp index 654418fcd7..27a7ae7f89 100644 --- a/src/invariant/test/BucketListIsConsistentWithDatabaseTests.cpp +++ b/src/invariant/test/BucketListIsConsistentWithDatabaseTests.cpp @@ -146,8 +146,7 @@ struct BucketListGenerator auto header = ltx.loadHeader().current(); ltx.getAllEntries(initEntries, liveEntries, deadEntries); BucketTestUtils::addLiveBatchAndUpdateSnapshot( - *app, header, initEntries, - liveEntries, deadEntries); + *app, header, initEntries, liveEntries, deadEntries); ltx.commit(); } diff --git a/src/ledger/LedgerManagerImpl.cpp b/src/ledger/LedgerManagerImpl.cpp index 80d551eee3..c5436f841e 100644 --- a/src/ledger/LedgerManagerImpl.cpp +++ b/src/ledger/LedgerManagerImpl.cpp @@ -1298,10 +1298,14 @@ LedgerManagerImpl::advanceLedgerPointers(LedgerHeader const& header, if (mApp.getConfig().isUsingBucketListDB() && header.ledgerSeq != prevLedgerSeq) { - mApp.getBucketManager() - .getBucketSnapshotManager() - .updateCurrentSnapshot(std::make_unique( - mApp.getBucketManager().getBucketList(), header)); + auto& bm = mApp.getBucketManager(); + auto liveSnapshot = std::make_unique>( + bm.getLiveBucketList(), header); + auto hotArchiveSnapshot = + std::make_unique>( + bm.getHotArchiveBucketList(), header); + bm.getBucketSnapshotManager().updateCurrentSnapshot( + std::move(liveSnapshot), std::move(hotArchiveSnapshot)); } } diff --git a/src/ledger/LedgerStateSnapshot.cpp b/src/ledger/LedgerStateSnapshot.cpp index 6f0228884e..e97864e0d1 100644 --- a/src/ledger/LedgerStateSnapshot.cpp +++ b/src/ledger/LedgerStateSnapshot.cpp @@ -164,7 +164,7 @@ LedgerTxnReadOnly::executeWithMaybeInnerSnapshot( } BucketSnapshotState::BucketSnapshotState(BucketManager& bm) - : mSnapshot(bm.getSearchableBucketListSnapshot()) + : mSnapshot(bm.getSearchableLiveBucketListSnapshot()) , mLedgerHeader(LedgerHeaderWrapper( std::make_shared(mSnapshot->getLedgerHeader()))) { diff --git a/src/ledger/LedgerStateSnapshot.h b/src/ledger/LedgerStateSnapshot.h index 7a57b1c771..dc4f6f76f9 100644 --- a/src/ledger/LedgerStateSnapshot.h +++ b/src/ledger/LedgerStateSnapshot.h @@ -105,7 +105,7 @@ class LedgerTxnReadOnly : public AbstractLedgerStateSnapshot // A concrete implementation of read-only BucketList snapshot wrapper class BucketSnapshotState : public AbstractLedgerStateSnapshot { - std::shared_ptr mSnapshot; + std::shared_ptr mSnapshot; // Store a copy of the header from mSnapshot. This is needed for // validation flow where for certain validation scenarios the header needs // to be modified diff --git a/src/ledger/LedgerTxn.cpp b/src/ledger/LedgerTxn.cpp index 8cf1e52261..348cb6fc68 100644 --- a/src/ledger/LedgerTxn.cpp +++ b/src/ledger/LedgerTxn.cpp @@ -3103,7 +3103,7 @@ LedgerTxnRoot::Impl::prefetchInternal(UnorderedSet const& keys, { insertIfNotLoaded(keysToSearch, key); } - auto blLoad = getSearchableBucketListSnapshot().loadKeysWithLimits( + auto blLoad = getSearchableLiveBucketListSnapshot().loadKeysWithLimits( keysToSearch, lkMeter); cacheResult(populateLoadedEntries(keysToSearch, blLoad, lkMeter)); } @@ -3487,14 +3487,15 @@ LedgerTxnRoot::Impl::areEntriesMissingInCacheForOffer(OfferEntry const& oe) } SearchableLiveBucketListSnapshot& -LedgerTxnRoot::Impl::getSearchableBucketListSnapshot() const +LedgerTxnRoot::Impl::getSearchableLiveBucketListSnapshot() const { releaseAssert(mApp.getConfig().isUsingBucketListDB()); if (!mSearchableBucketListSnapshot) { - mSearchableBucketListSnapshot = mApp.getBucketManager() - .getBucketSnapshotManager() - .copySearchableBucketListSnapshot(); + mSearchableBucketListSnapshot = + mApp.getBucketManager() + .getBucketSnapshotManager() + .copySearchableLiveBucketListSnapshot(); } return *mSearchableBucketListSnapshot; @@ -3635,7 +3636,7 @@ LedgerTxnRoot::Impl::getPoolShareTrustLinesByAccountAndAsset( if (mApp.getConfig().isUsingBucketListDB()) { trustLines = - getSearchableBucketListSnapshot() + getSearchableLiveBucketListSnapshot() .loadPoolShareTrustLinesByAccountAndAsset(account, asset); } else @@ -3698,7 +3699,7 @@ LedgerTxnRoot::Impl::getInflationWinners(size_t maxWinners, int64_t minVotes) { if (mApp.getConfig().isUsingBucketListDB()) { - return getSearchableBucketListSnapshot().loadInflationWinners( + return getSearchableLiveBucketListSnapshot().loadInflationWinners( maxWinners, minVotes); } else @@ -3754,7 +3755,7 @@ LedgerTxnRoot::Impl::getNewestVersion(InternalLedgerKey const& gkey) const { if (mApp.getConfig().isUsingBucketListDB() && key.type() != OFFER) { - entry = getSearchableBucketListSnapshot().load(key); + entry = getSearchableLiveBucketListSnapshot().load(key); } else { diff --git a/src/ledger/LedgerTxnImpl.h b/src/ledger/LedgerTxnImpl.h index 42b4571514..5b6d9299e7 100644 --- a/src/ledger/LedgerTxnImpl.h +++ b/src/ledger/LedgerTxnImpl.h @@ -871,7 +871,8 @@ class LedgerTxnRoot::Impl bool areEntriesMissingInCacheForOffer(OfferEntry const& oe); - SearchableLiveBucketListSnapshot& getSearchableBucketListSnapshot() const; + SearchableLiveBucketListSnapshot& + getSearchableLiveBucketListSnapshot() const; uint32_t prefetchInternal(UnorderedSet const& keys, LedgerKeyMeter* lkMeter = nullptr); diff --git a/src/ledger/NetworkConfig.cpp b/src/ledger/NetworkConfig.cpp index c9347088b3..cef11cb916 100644 --- a/src/ledger/NetworkConfig.cpp +++ b/src/ledger/NetworkConfig.cpp @@ -1865,9 +1865,8 @@ SorobanNetworkConfig::writeAllSettings(AbstractLedgerTxn& ltx, { auto lcl = app.getLedgerManager().getLastClosedLedgerHeader(); lcl.header.ledgerSeq += 1; - BucketTestUtils::addBatchAndUpdateSnapshot( - app.getBucketManager().getLiveBucketList(), app, lcl.header, {}, - entries, {}); + BucketTestUtils::addLiveBatchAndUpdateSnapshot(app, lcl.header, {}, + entries, {}); } } #endif diff --git a/src/ledger/test/LedgerTxnTests.cpp b/src/ledger/test/LedgerTxnTests.cpp index e98ad0acce..10200eea2d 100644 --- a/src/ledger/test/LedgerTxnTests.cpp +++ b/src/ledger/test/LedgerTxnTests.cpp @@ -2747,9 +2747,8 @@ TEST_CASE("LedgerTxnRoot prefetch classic entries", "[ledgertxn]") .getLastClosedLedgerHeader() .header.ledgerVersion; lh.ledgerSeq = 2; - BucketTestUtils::addBatchAndUpdateSnapshot( - app->getBucketManager().getLiveBucketList(), *app, lh, {}, - ledgerVect, {}); + BucketTestUtils::addLiveBatchAndUpdateSnapshot(*app, lh, {}, + ledgerVect, {}); } ltx.commit(); @@ -2980,9 +2979,8 @@ TEST_CASE("LedgerTxnRoot prefetch soroban entries", "[ledgertxn]") .getLastClosedLedgerHeader() .header.ledgerVersion; lh.ledgerSeq = 2; - BucketTestUtils::addBatchAndUpdateSnapshot( - app->getBucketManager().getLiveBucketList(), *app, lh, {}, ledgerVect, - deadKeyVect); + BucketTestUtils::addLiveBatchAndUpdateSnapshot(*app, lh, {}, ledgerVect, + deadKeyVect); ltx.commit(); auto addTxn = [&](bool enoughQuota, std::vector entries, diff --git a/src/main/QueryServer.cpp b/src/main/QueryServer.cpp index 97657105a1..95f1d80a44 100644 --- a/src/main/QueryServer.cpp +++ b/src/main/QueryServer.cpp @@ -66,8 +66,8 @@ QueryServer::QueryServer(const std::string& address, unsigned short port, auto workerPids = mServer.start(); for (auto pid : workerPids) { - mBucketListSnapshots[pid] = - std::move(bucketSnapshotManager.copySearchableBucketListSnapshot()); + mBucketListSnapshots[pid] = std::move( + bucketSnapshotManager.copySearchableLiveBucketListSnapshot()); } } @@ -149,16 +149,17 @@ QueryServer::getLedgerEntryRaw(std::string const& params, { root["ledgerSeq"] = *snapshotLedger; - bool snapshotExists; - std::tie(loadedKeys, snapshotExists) = + auto loadedKeysOp = bl.loadKeysFromLedger(orderedKeys, *snapshotLedger); // Return 404 if ledgerSeq not found - if (!snapshotExists) + if (!loadedKeysOp) { retStr = "LedgerSeq not found"; return false; } + + loadedKeys = std::move(*loadedKeysOp); } // Otherwise default to current ledger else diff --git a/src/main/QueryServer.h b/src/main/QueryServer.h index f16a79c945..53ee434087 100644 --- a/src/main/QueryServer.h +++ b/src/main/QueryServer.h @@ -14,7 +14,7 @@ namespace stellar { -class SearchableBucketListSnapshot; +class SearchableLiveBucketListSnapshot; class BucketSnapshotManager; class QueryServer @@ -26,7 +26,7 @@ class QueryServer httpThreaded::server::server mServer; std::unordered_map> + std::shared_ptr> mBucketListSnapshots; bool safeRouter(HandlerRoute route, std::string const& params, diff --git a/src/overlay/test/FloodTests.cpp b/src/overlay/test/FloodTests.cpp index e224530931..e2f9b50ac1 100644 --- a/src/overlay/test/FloodTests.cpp +++ b/src/overlay/test/FloodTests.cpp @@ -73,9 +73,8 @@ TEST_CASE("Flooding", "[flood][overlay][acceptance]") auto const& header = n->getLedgerManager() .getLastClosedLedgerHeader() .header; - BucketTestUtils::addBatchAndUpdateSnapshot( - n->getBucketManager().getBucketList(), *n, header, {}, - {gen}, {}); + BucketTestUtils::addLiveBatchAndUpdateSnapshot( + *n, header, {}, {gen}, {}); } } } diff --git a/src/simulation/CoreTests.cpp b/src/simulation/CoreTests.cpp index b3304dcdfd..423b77e211 100644 --- a/src/simulation/CoreTests.cpp +++ b/src/simulation/CoreTests.cpp @@ -687,8 +687,7 @@ TEST_CASE("Bucket list entries vs write throughput", "[scalability][!hide]") lh.ledgerVersion = Config::CURRENT_LEDGER_PROTOCOL_VERSION; lh.ledgerSeq = i; BucketTestUtils::addLiveBatchAndUpdateSnapshot( - *app, lh, - LedgerTestUtils::generateValidLedgerEntries(100), + *app, lh, LedgerTestUtils::generateValidLedgerEntries(100), LedgerTestUtils::generateValidLedgerEntries(20), LedgerTestUtils::generateValidLedgerEntryKeysWithExclusions( {CONFIG_SETTING}, 5));