Make state archival startup check opt-in, optimize memory consumption (#4985)

This is useful for boxes with low-end memory config - they can still run
the invariant, but skip the expensive startup check that requires
loading all keys in-memory. The change also improve memory consumption
of the startup invariant by traversing the bucketlist iteratively.

Author: graydon

src/bucket/BucketListSnapshotBase.h

@@ -103,12 +103,6 @@ class SearchableBucketListSnapshotBase : public NonMovableOrCopyable
     medida::Meter& mBloomMisses;
     medida::Meter& mBloomLookups;
 
-    // Loops through all buckets, starting with curr at level 0, then snap at
-    // level 0, etc. Calls f on each bucket. Exits early if function
-    // returns Loop::COMPLETE.
-    void loopAllBuckets(std::function<Loop(BucketSnapshotT const&)> f,
-                        BucketListSnapshot<BucketT> const& snapshot) const;
-
     SearchableBucketListSnapshotBase(
         BucketSnapshotManager const& snapshotManager,
         AppConnector const& appConnector, SnapshotPtrT<BucketT>&& snapshot,
@@ -122,6 +116,19 @@ class SearchableBucketListSnapshotBase : public NonMovableOrCopyable
                                     size_t numEntries) const;
 
   public:
+    // Loops through all buckets, starting with curr at level 0, then snap at
+    // level 0, etc. Calls f on each bucket. Exits early if function
+    // returns Loop::COMPLETE.
+    void loopAllBuckets(std::function<Loop(BucketSnapshotT const&)> f,
+                        BucketListSnapshot<BucketT> const& snapshot) const;
+
+    // Overload that uses the internal snapshot
+    void
+    loopAllBuckets(std::function<Loop(BucketSnapshotT const&)> f) const
+    {
+        loopAllBuckets(f, *mSnapshot);
+    }
+
     uint32_t
     getLedgerSeq() const
     {

src/invariant/ArchivedStateConsistency.cpp

@@ -27,7 +27,16 @@ ArchivedStateConsistency::ArchivedStateConsistency() : Invariant(true)
 std::string
 ArchivedStateConsistency::start(Application& app)
 {
-    LogSlowExecution logSlow("ArchivedStateConsistency::start");
+    releaseAssert(threadIsMain());
+    LogSlowExecution logSlow("ArchivedStateConsistency startup");
+
+    if (!app.getConfig().INVARIANT_EXTRA_CHECKS)
+    {
+        CLOG_INFO(Invariant,
+                  "Skipping ArchivedStateConsistency startup check - "
+                  "INVARIANT_EXTRA_CHECKS is disabled");
+        return std::string{};
+    }
 
     auto protocolVersion =
         app.getLedgerManager().getLastClosedLedgerHeader().header.ledgerVersion;
@@ -47,31 +56,58 @@ ArchivedStateConsistency::start(Application& app)
 
     std::map<LedgerKey, LedgerEntry> archived =
         app.getBucketManager().loadCompleteHotArchiveState(has);
-    std::map<LedgerKey, LedgerEntry> live =
-        app.getBucketManager().loadCompleteLedgerState(has);
-    auto archivedIt = archived.begin();
-    auto liveIt = live.begin();
 
-    while (archivedIt != archived.end() && liveIt != live.end())
-    {
-        if (archivedIt->first < liveIt->first)
+    // Get live snapshot for iterating through buckets
+    auto liveSnapshot = app.getBucketManager()
+                            .getBucketSnapshotManager()
+                            .copySearchableLiveBucketListSnapshot();
+
+    // Track which keys we've already seen in live buckets (from level 0 upward)
+    // to avoid checking duplicates
+    UnorderedSet<LedgerKey> seenKeys;
+
+    // Iterate through live buckets from level 0 upward and check if any key
+    // also exists in archived state
+    std::string result;
+    liveSnapshot->loopAllBuckets([&](LiveBucketSnapshot const& bucketSnapshot) {
+        LiveBucketInputIterator it(bucketSnapshot.getRawBucket());
+        while (it && result.empty())
         {
-            archivedIt++;
-            continue;
-        }
-        else if (liveIt->first < archivedIt->first)
-        {
-            liveIt++;
-            continue;
-        }
-        else
-        {
-            return fmt::format(
-                FMT_STRING(
-                    "ArchivedStateConsistency:: Entry with the same key is "
-                    "present in both live and archived state. Key: {}"),
-                xdrToCerealString(archivedIt->first, "entry_key"));
+            BucketEntry const& e = *it;
+            if (e.type() == LIVEENTRY || e.type() == INITENTRY)
+            {
+                auto key = LedgerEntryKey(e.liveEntry());
+
+                // Skip if we've already seen this key in a more recent
+                // bucket
+                if (seenKeys.find(key) == seenKeys.end())
+                {
+                    seenKeys.insert(key);
+
+                    // Check if this key also exists in archived state
+                    if (archived.find(key) != archived.end())
+                    {
+                        result = fmt::format(
+                            FMT_STRING(
+                                "ArchivedStateConsistency: Entry with the "
+                                "same key is present in both live and "
+                                "archived state. Key: {}"),
+                            xdrToCerealString(key, "entry_key"));
+                    }
+                }
+            }
+            else
+            {
+                seenKeys.insert(e.deadEntry());
+            }
+            ++it;
         }
+        return result.empty() ? Loop::INCOMPLETE : Loop::COMPLETE;
+    });
+
+    if (!result.empty())
+    {
+        return result;
     }
 
     CLOG_INFO(Invariant, "ArchivedStateConsistency invariant passed");
@@ -134,6 +170,10 @@ ArchivedStateConsistency::checkOnLedgerCommit(
     // Keys for restored entries
     for (auto const& [key, entry] : restoredFromArchive)
     {
+        if (key.type() != TTL)
+        {
+            releaseAssertOrThrow(isPersistentEntry(key));
+        }
         allKeys.insert(key);
     }
     for (auto const& [key, entry] : restoredFromLiveState)

src/main/Config.cpp

@@ -355,6 +355,7 @@ Config::Config() : NODE_SEED(SecretKey::random())
     COMMANDS = {};
     REPORT_METRICS = {};
     INVARIANT_CHECKS = {};
+    INVARIANT_EXTRA_CHECKS = false;
 
 #ifdef BUILD_TESTS
     TEST_CASES_ENABLED = false;
@@ -1464,6 +1465,8 @@ Config::processConfig(std::shared_ptr<cpptoml::table> t)
                  [&]() { NETWORK_PASSPHRASE = readString(item); }},
                 {"INVARIANT_CHECKS",
                  [&]() { INVARIANT_CHECKS = readArray<std::string>(item); }},
+                {"INVARIANT_EXTRA_CHECKS",
+                 [&]() { INVARIANT_EXTRA_CHECKS = readBool(item); }},
                 {"ENTRY_CACHE_SIZE",
                  [&]() { ENTRY_CACHE_SIZE = readInt<uint32_t>(item); }},
                 {"PREFETCH_BATCH_SIZE",

src/main/Config.h

@@ -751,6 +751,11 @@ class Config : public std::enable_shared_from_this<Config>
     // Invariants
     std::vector<std::string> INVARIANT_CHECKS;
 
+    // Enable extra invariant checks that provide additional consistency
+    // verification but may be slow or resource-intensive. The default value is
+    // false.
+    bool INVARIANT_EXTRA_CHECKS;
+
     std::map<std::string, std::string> VALIDATOR_NAMES;
 
     // Information necessary to compute the weight of a validator for leader