State Archival Consistency Invariant (#4968)

(Overview by @claude )

Introduce new invariant ArchivedStateConsistency that is invoked on
every commit.

### Eviction Checks (checkEvictionInvariants)
- Verifies that entries being archived don't already exist in the
archive
- Confirms that evicted entries actually exist in the current live state
- Validates that associated TTL (Time-To-Live) entries are properly
cleaned up when entries are archived
- Ensures archived entries contain the correct data from the live state
before eviction

### Restoration Checks (checkRestoreInvariants)
- Validates entries being restored from archive are correctly retrieved
  - Ensures restored entries don't conflict with current live state
- Verifies TTL entries for restored persistent entries are handled
correctly

### Startup Validation
- On node startup, scans the complete hot archive and live bucket lists
- Ensures no entries exist in both live and archived state
simultaneously (critical consistency check)

Author: marta-lokhova

docs/stellar-core_example.cfg

@@ -497,11 +497,17 @@ RUN_STANDALONE=false
 #     of the network, caution is advised when using this.
 # - "EventsAreConsistentWithEntryDiffs"
 #     Setting this will cause additional work on each operation apply - it
-#     checks that all asset movements and events provide equivalent 
+#     checks that all asset movements and events provide equivalent
 #     balance changes. The invariant required both EMIT_CLASSIC_EVENTS and
 #     BACKFILL_STELLAR_ASSET_EVENTS config flags to be enabled. This is also
 #     a "strict" invariant, which means your node will shutdown if the invariant
 #     triggers.
+# - "ArchivedStateConsistency"
+#     Setting this will cause additional work during ledger close - it
+#     checks that archived entries evicted from the database have the correct value.
+#     This check runs when entries are evicted during ledger close, adding
+#     overhead to ledger close time. This is also a "strict" invariant, which
+#     means your node will shutdown if the invariant triggers.
 INVARIANT_CHECKS = [ "AccountSubEntriesCountIsValid",
 "ConservationOfLumens",
 "ConstantProductInvariant",

src/bucket/BucketManager.cpp

@@ -16,6 +16,7 @@
 #include "crypto/Hex.h"
 #include "history/HistoryManager.h"
 #include "historywork/VerifyBucketWork.h"
+#include "invariant/InvariantManager.h"
 #include "ledger/LedgerManager.h"
 #include "ledger/LedgerTxn.h"
 #include "ledger/LedgerTypeUtils.h"
@@ -28,6 +29,7 @@
 #include "util/Logging.h"
 #include "util/ProtocolVersion.h"
 #include "util/TmpDir.h"
+#include "util/UnorderedMap.h"
 #include "util/types.h"
 #include "xdr/Stellar-ledger.h"
 #include <filesystem>
@@ -1144,6 +1146,9 @@ BucketManager::startBackgroundEvictionScan(uint32_t ledgerSeq,
 
     auto searchableBL =
         mSnapshotManager->copySearchableLiveBucketListSnapshot();
+
+    // Snapshot should be based on the lcl ledger
+    releaseAssertOrThrow(searchableBL->getLedgerSeq() == ledgerSeq - 1);
     auto const& sas = cfg.stateArchivalSettings();
 
     using task_t =
@@ -1429,17 +1434,65 @@ loadEntriesFromBucket(std::shared_ptr<LiveBucket> b, std::string const& name,
               b->getSize(), name, ms, formatSize(bytesPerSec));
 }
 
+// Loads a single bucket worth of entries into `map`, deleting tombstone entries
+// and inserting archived entries. Should be called in a loop over a BL, from
+// old to new.
+static void
+loadEntriesFromHotArchiveBucket(std::shared_ptr<HotArchiveBucket> b,
+                                std::string const& name,
+                                std::map<LedgerKey, LedgerEntry>& map)
+{
+    ZoneScoped;
+
+    using namespace std::chrono;
+    medida::Timer timer;
+    HotArchiveBucketInputIterator in(b);
+    timer.Time([&]() {
+        while (in)
+        {
+            HotArchiveBucketEntry const& e = *in;
+            if (e.type() == HOT_ARCHIVE_ARCHIVED)
+            {
+                map[LedgerEntryKey(e.archivedEntry())] = e.archivedEntry();
+            }
+            else
+            {
+                if (e.type() != HOT_ARCHIVE_LIVE)
+                {
+                    std::string err =
+                        "Malformed hot archive bucket: unexpected "
+                        "non-HOT_ARCHIVE_LIVE entry.";
+                    CLOG_ERROR(Bucket, "{}", err);
+                    throw std::runtime_error(err);
+                }
+                map.erase(e.key());
+            }
+            ++in;
+        }
+    });
+    nanoseconds ns =
+        timer.duration_unit() * static_cast<nanoseconds::rep>(timer.max());
+    milliseconds ms = duration_cast<milliseconds>(ns);
+    size_t bytesPerSec = (b->getSize() * 1000 / (1 + ms.count()));
+    CLOG_INFO(Bucket, "Read {}-byte bucket file '{}' in {} ({}/s)",
+              b->getSize(), name, ms, formatSize(bytesPerSec));
+}
+
+template <typename BucketT>
 std::map<LedgerKey, LedgerEntry>
-BucketManager::loadCompleteLedgerState(HistoryArchiveState const& has)
+BucketManager::loadCompleteBucketListStateHelper(
+    std::vector<HistoryStateBucket<BucketT>> const& buckets,
+    std::function<void(std::shared_ptr<BucketT>, std::string const&,
+                       std::map<LedgerKey, LedgerEntry>&)>
+        loadFunc)
 {
     ZoneScoped;
 
     std::map<LedgerKey, LedgerEntry> ledgerMap;
     std::vector<std::pair<Hash, std::string>> hashes;
-    for (uint32_t i = LiveBucketList::kNumLevels; i > 0; --i)
+    for (uint32_t i = BucketListBase<BucketT>::kNumLevels; i > 0; --i)
     {
-        HistoryStateBucket<LiveBucket> const& hsb =
-            has.currentBuckets.at(i - 1);
+        HistoryStateBucket<BucketT> const& hsb = buckets.at(i - 1);
         hashes.emplace_back(hexToBin256(hsb.snap),
                             fmt::format(FMT_STRING("snap {:d}"), i - 1));
         hashes.emplace_back(hexToBin256(hsb.curr),
@@ -1451,17 +1504,36 @@ BucketManager::loadCompleteLedgerState(HistoryArchiveState const& has)
         {
             continue;
         }
-        auto b = getBucketByHashInternal(pair.first, mSharedLiveBuckets);
+        auto b = getBucketByHash<BucketT>(pair.first);
         if (!b)
         {
             throw std::runtime_error(std::string("missing bucket: ") +
                                      binToHex(pair.first));
         }
-        loadEntriesFromBucket(b, pair.second, ledgerMap);
+
+        loadFunc(b, pair.second, ledgerMap);
     }
     return ledgerMap;
 }
 
+// Loads the complete state of the live BucketList into a map
+std::map<LedgerKey, LedgerEntry>
+BucketManager::loadCompleteLedgerState(HistoryArchiveState const& has)
+{
+    CLOG_INFO(Bucket, "Loading complete live ledger state");
+    return loadCompleteBucketListStateHelper<LiveBucket>(has.currentBuckets,
+                                                         loadEntriesFromBucket);
+}
+
+// Loads the complete state of the hot archive BucketList into a map
+std::map<LedgerKey, LedgerEntry>
+BucketManager::loadCompleteHotArchiveState(HistoryArchiveState const& has)
+{
+    CLOG_INFO(Bucket, "Loading complete hot archive state");
+    return loadCompleteBucketListStateHelper<HotArchiveBucket>(
+        has.hotArchiveBuckets, loadEntriesFromHotArchiveBucket);
+}
+
 std::shared_ptr<LiveBucket>
 BucketManager::mergeBuckets(asio::io_context& ctx,
                             HistoryArchiveState const& has)

src/bucket/BucketManager.h

@@ -1,9 +1,11 @@
 #pragma once
 
 #include "bucket/BucketMergeMap.h"
+#include "history/HistoryArchive.h"
 #include "ledger/NetworkConfig.h"
 #include "main/Config.h"
 #include "util/TmpDir.h"
+#include "util/UnorderedMap.h"
 #include "util/types.h"
 #include "work/BasicWork.h"
 #include "xdr/Stellar-ledger.h"
@@ -171,6 +173,13 @@ class BucketManager : NonMovableOrCopyable
 
     void reportLiveBucketIndexCacheMetrics();
 
+    template <class BucketT>
+    std::map<LedgerKey, LedgerEntry> loadCompleteBucketListStateHelper(
+        std::vector<HistoryStateBucket<BucketT>> const& buckets,
+        std::function<void(std::shared_ptr<BucketT>, std::string const&,
+                           std::map<LedgerKey, LedgerEntry>&)>
+            loadFunc);
+
 #ifdef BUILD_TESTS
     bool mUseFakeTestValuesForNextClose{false};
     uint32_t mFakeTestProtocolVersion;
@@ -378,6 +387,9 @@ class BucketManager : NonMovableOrCopyable
     std::map<LedgerKey, LedgerEntry>
     loadCompleteLedgerState(HistoryArchiveState const& has);
 
+    std::map<LedgerKey, LedgerEntry>
+    loadCompleteHotArchiveState(HistoryArchiveState const& has);
+
     // Merge the bucket list of the provided HAS into a single "super bucket"
     // consisting of only live entries, and return it.
     std::shared_ptr<LiveBucket> mergeBuckets(asio::io_context& ctx,

src/bucket/test/BucketTestUtils.cpp

@@ -187,6 +187,8 @@ template size_t countEntries(std::shared_ptr<HotArchiveBucket> bucket);
 
 void
 LedgerManagerForBucketTests::finalizeLedgerTxnChanges(
+    SearchableSnapshotConstPtr lclSnapshot,
+    SearchableHotArchiveSnapshotConstPtr lclHotArchiveSnapshot,
     AbstractLedgerTxn& ltx,
     std::unique_ptr<LedgerCloseMetaFrame> const& ledgerCloseMeta,
     LedgerHeader lh, uint32_t initialLedgerVers)
@@ -342,8 +344,9 @@ LedgerManagerForBucketTests::finalizeLedgerTxnChanges(
     }
     else
     {
-        LedgerManagerImpl::finalizeLedgerTxnChanges(ltx, ledgerCloseMeta, lh,
-                                                    initialLedgerVers);
+        LedgerManagerImpl::finalizeLedgerTxnChanges(
+            lclSnapshot, lclHotArchiveSnapshot, ltx, ledgerCloseMeta, lh,
+            initialLedgerVers);
     }
 }
 

src/bucket/test/BucketTestUtils.h

@@ -73,6 +73,8 @@ class LedgerManagerForBucketTests : public LedgerManagerImpl
 
   protected:
     void finalizeLedgerTxnChanges(
+        SearchableSnapshotConstPtr lclSnapshot,
+        SearchableHotArchiveSnapshotConstPtr lclHotArchiveSnapshot,
         AbstractLedgerTxn& ltx,
         std::unique_ptr<LedgerCloseMetaFrame> const& ledgerCloseMeta,
         LedgerHeader lh, uint32_t initialLedgerVers) override;