Added state snapshot invariant tests (#5009)

# Description

Resolves #5002

This adds the `stateSnapshotInvariant`. This is intended for expensive
checks, like scanning the entire BucketList to reconcile checks.
Periodically, we take a snapshot of ledger state at the end of
applyLedger, spin up a background thread, and pass the snapshot over to
the background thread to run checks. This is expensive from a memory
perspective, so it is hidden behind the `INVARIANT_EXTRA_CHECKS` flag.
The flag itself is also now only allowed for watchers. I've replaced the
`start` invariant with this snapshot invariant and set a special case on
startup where the snapshot invariant will run. I also rewrote the
`ArchivedStateConsistency::start` invariant to use this new snapshot
interface and not rely on HAS files.

There's two sort of messy places I'm not super sure on. First, the
failure behavior of the invariant. Currently, if a strict invariant
fails, I post the failure handling back to the main thread. The function
on the main thread will then throw, killing the process. This is
basically what the other invariants do and I think kills the process in
the way we want, but I'm not sure if it's the best way to do this. The
other place I'm unsure of is `getInMemorySorobanStateForInvariantCheck`.
I want to maintain the apply time phase invariants we have around
`getInMemorySorobanState`, as we shouldn't be reading from it outside of
the apply phase. However, we do need to copy it during the commit phase
for the invariant. For now I've just made another getter with different
asserts, with naming to suggest that it should only be used for
invariants, but it feels like there's a better way to harden this.

# Checklist
- [x] Reviewed the
[contributing](https://github.com/stellar/stellar-core/blob/master/CONTRIBUTING.md#submitting-changes)
document
- [x] Rebased on top of master (no merge commits)
- [x] Ran `clang-format` v8.0.0 (via `make format` or the Visual Studio
extension)
- [x] Compiles
- [x] Ran all tests
- [ ] If change impacts performance, include supporting evidence per the
[performance
document](https://github.com/stellar/stellar-core/blob/master/performance-eval/performance-eval.md)

Author: marta-lokhova

docs/metrics.md

@@ -87,6 +87,7 @@ ledger.apply-soroban.max-clusters         | counter   | maximum number of cluste
 ledger.apply-soroban.stages               | counter   | number of stages used for parallel apply in a ledger
 ledger.catchup.duration                   | timer     | time between entering LM_CATCHING_UP_STATE and entering LM_SYNCED_STATE
 ledger.invariant.failure                  | counter   | number of times invariants failed
+ledger.invariant.state-snapshot-skipped   | counter   | number of times state snapshot invariant was skipped due to previous scan still running
 ledger.ledger.close                       | timer     | time to close a ledger (excluding consensus)
 ledger.memory.queued-ledgers              | counter   | number of ledgers queued in memory for replay
 ledger.metastream.bytes                   | meter     | number of bytes written per ledger into meta-stream

docs/stellar-core_example.cfg

@@ -516,6 +516,13 @@ INVARIANT_CHECKS = [ "AccountSubEntriesCountIsValid",
 "SponsorshipCountIsValid" ]
 
 
+# STATE_SNAPSHOT_INVARIANT_LEDGER_FREQUENCY (integer) defaults to 300
+# Frequency (in seconds) at which expensive state snapshot invariants should run.
+# State snapshot invariants perform comprehensive checks that scan the entire
+# bucket list, which are too expensive to perform on every ledger.
+STATE_SNAPSHOT_INVARIANT_LEDGER_FREQUENCY=300
+
+
 # MANUAL_CLOSE (true or false) defaults to false
 # Mode for testing. Ledger will only close when stellar-core gets
 #  the `manualclose` command

src/invariant/ArchivedStateConsistency.cpp

@@ -3,9 +3,9 @@
 // of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
 
 #include "invariant/ArchivedStateConsistency.h"
-#include "bucket/BucketManager.h"
 #include "bucket/BucketSnapshot.h"
 #include "bucket/BucketSnapshotManager.h"
+#include "bucket/HotArchiveBucket.h"
 #include "bucket/LedgerCmp.h"
 #include "invariant/InvariantManager.h"
 #include "ledger/LedgerManager.h"
@@ -17,100 +17,89 @@
 #include "util/XDRCereal.h"
 #include "util/types.h"
 #include <fmt/format.h>
+#include <vector>
 
 namespace stellar
 {
 ArchivedStateConsistency::ArchivedStateConsistency() : Invariant(true)
 {
 }
 
-std::string
-ArchivedStateConsistency::start(Application& app)
+bool
+ArchivedStateConsistency::usesStateSnapshotInvariant() const
 {
-    releaseAssert(threadIsMain());
-    LogSlowExecution logSlow("ArchivedStateConsistency startup");
+    return true;
+}
 
-    if (!app.getConfig().INVARIANT_EXTRA_CHECKS)
-    {
-        CLOG_INFO(Invariant,
-                  "Skipping ArchivedStateConsistency startup check - "
-                  "INVARIANT_EXTRA_CHECKS is disabled");
-        return std::string{};
-    }
+// This test iterates through both the live and archived bucket lists and checks
+// that no entry is live in both BucketLists simultaneously.
+std::string
+ArchivedStateConsistency::stateSnapshotInvariant(
+    CompleteConstLedgerStatePtr ledgerState,
+    InMemorySorobanState const& inMemorySnapshot)
+{
+    LogSlowExecution logSlow("ArchivedStateConsistency::stateSnapshotInvariant",
+                             LogSlowExecution::Mode::AUTOMATIC_RAII, "took",
+                             std::chrono::seconds(30));
 
-    auto protocolVersion =
-        app.getLedgerManager().getLastClosedLedgerHeader().header.ledgerVersion;
+    auto liveSnapshot = ledgerState->getBucketSnapshot();
+    auto hotArchiveSnapshot = ledgerState->getHotArchiveSnapshot();
+    auto const& header = liveSnapshot->getLedgerHeader();
     if (protocolVersionIsBefore(
-            protocolVersion,
+            header.ledgerVersion,
             LiveBucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION))
     {
-        CLOG_INFO(Invariant,
-                  "Skipping ArchivedStateConsistency invariant for "
-                  "protocol version {}",
-                  protocolVersion);
         return std::string{};
     }
 
-    CLOG_INFO(Invariant, "Starting ArchivedStateConsistency invariant");
-    auto has = app.getLedgerManager().getLastClosedLedgerHAS();
+    auto const ARCHIVAL_ENTRY_TYPES = std::set{CONTRACT_CODE, CONTRACT_DATA};
 
-    std::map<LedgerKey, LedgerEntry> archived =
-        app.getBucketManager().loadCompleteHotArchiveState(has);
-
-    // 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())
-        {
-            BucketEntry const& e = *it;
-            if (e.type() == LIVEENTRY || e.type() == INITENTRY)
+    std::string errorMsg;
+
+    // For each entry in the Live BucketList, check if we have seen it in a
+    // previous level. If not, this entry is the newest version, so check if it
+    // exists in the Hot Archive.
+    auto checkIfLiveEntryInArchive =
+        [&seenKeys, &errorMsg, &hotArchiveSnapshot](BucketEntry const& be) {
+            if (be.type() == LIVEENTRY || be.type() == INITENTRY)
             {
-                auto key = LedgerEntryKey(e.liveEntry());
+                auto lk = LedgerEntryKey(be.liveEntry());
+                auto [_, wasInserted] = seenKeys.emplace(lk);
 
-                // Skip if we've already seen this key in a more recent
-                // bucket
-                if (seenKeys.find(key) == seenKeys.end())
+                // If this BucketEntry is not shadowed, and the key exists in
+                // the Hot Archive, we have an error.
+                if (wasInserted && hotArchiveSnapshot->load(lk))
                 {
-                    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"));
-                    }
+                    errorMsg = fmt::format(
+                        FMT_STRING("ArchivedStateConsistency invariant failed: "
+                                   "Live entry is present in both live and "
+                                   "archived state: {}"),
+                        xdrToCerealString(lk, "entry_key"));
+                    return Loop::COMPLETE;
                 }
             }
-            else
+            // Mark DEADENTRY as seen, but the key does not exist wrt ledger
+            // state, so we don't need to check the Hot Archive.
+            else if (be.type() == DEADENTRY)
             {
-                seenKeys.insert(e.deadEntry());
+                seenKeys.emplace(be.deadEntry());
             }
-            ++it;
-        }
-        return result.empty() ? Loop::INCOMPLETE : Loop::COMPLETE;
-    });
+            return Loop::INCOMPLETE;
+        };
 
-    if (!result.empty())
+    // We just need to check for Soroban types that are stored in the Hot
+    // Archive
+    for (auto const& type : ARCHIVAL_ENTRY_TYPES)
     {
-        return result;
+        liveSnapshot->scanForEntriesOfType(type, checkIfLiveEntryInArchive);
+        if (!errorMsg.empty())
+        {
+            return errorMsg;
+        }
+        seenKeys.clear();
     }
 
-    CLOG_INFO(Invariant, "ArchivedStateConsistency invariant passed");
     return std::string{};
 }
 
@@ -540,4 +529,4 @@ ArchivedStateConsistency::checkRestoreInvariants(
 
     return std::string{};
 }
-};
+};

src/invariant/ArchivedStateConsistency.h

@@ -42,6 +42,10 @@ class ArchivedStateConsistency : public Invariant
         UnorderedMap<LedgerKey, LedgerEntry> const& restoredFromLiveState)
         override;
 
-    virtual std::string start(Application& app) override;
+    bool usesStateSnapshotInvariant() const override;
+
+    std::string stateSnapshotInvariant(
+        CompleteConstLedgerStatePtr ledgerState,
+        InMemorySorobanState const& inMemorySnapshot) override;
 };
 }

src/invariant/Invariant.h

@@ -6,9 +6,9 @@
 
 #include "bucket/BucketSnapshotManager.h"
 #include "bucket/BucketUtils.h"
+#include "ledger/LedgerStateSnapshot.h"
 #include "xdr/Stellar-ledger.h"
 #include <cstdint>
-#include <functional>
 #include <memory>
 #include <string>
 #include <unordered_set>
@@ -84,8 +84,15 @@ class Invariant
         return std::string{};
     }
 
+    virtual bool
+    usesStateSnapshotInvariant() const
+    {
+        return false;
+    }
+
     virtual std::string
-    start(Application& app)
+    stateSnapshotInvariant(CompleteConstLedgerStatePtr ledgerState,
+                           InMemorySorobanState const& inMemorySnapshot)
     {
         return std::string{};
     }

src/invariant/InvariantManager.h

@@ -15,6 +15,7 @@ class AppConnector;
 class Application;
 class Bucket;
 class Invariant;
+class LedgerManager;
 struct EvictedStateVectors;
 struct LedgerTxnDelta;
 struct Operation;
@@ -60,13 +61,36 @@ class InvariantManager
         UnorderedMap<LedgerKey, LedgerEntry> const& restoredFromArchive,
         UnorderedMap<LedgerKey, LedgerEntry> const& restoredFromLiveState) = 0;
 
+    // This is used for expensive invariants that can't run in a blocking
+    // fashion, such as invariants that require scanning the entire BucketList.
+    // The invariant will periodically run on a background thread against the
+    // given ledger state snapshot. These invariants will only run if
+    // INVARIANT_EXTRA_CHECKS is enabled.
+    virtual void
+    runStateSnapshotInvariant(CompleteConstLedgerStatePtr ledgerState,
+                              InMemorySorobanState const& inMemorySnapshot) = 0;
+
     virtual void registerInvariant(std::shared_ptr<Invariant> invariant) = 0;
 
     virtual void enableInvariant(std::string const& name) = 0;
 
-    virtual void start(Application& app) = 0;
+    virtual void start(LedgerManager const& ledgerManager) = 0;
+
+    virtual bool shouldRunInvariantSnapshot() const = 0;
+
+    // Copy InMemorySorobanState for invariant checking. This is the only
+    // method that can access the private copy constructor of
+    // InMemorySorobanState.
+    virtual std::shared_ptr<InMemorySorobanState const>
+    copyInMemorySorobanStateForInvariant(
+        InMemorySorobanState const& state) const = 0;
 
 #ifdef BUILD_TESTS
+    // Blocks until any running snapshot invariant scan completes. Used in
+    // testing mode when ALWAYS_RUN_SNAPSHOT_FOR_TESTING is true to ensure
+    // scans complete before proceeding to the next ledger.
+    virtual void waitForScanToCompleteForTesting() const = 0;
+
     virtual void snapshotForFuzzer() = 0;
     virtual void resetForFuzzer() = 0;
 #endif // BUILD_TESTS