Make parallel apply compatible with RUN_STANDALONE (#5146)

# Description

This change makes RUN_STANDALONE compatible with parallel ledger apply.
There are a few motivations for this. First, many tests enable
`RUN_STANDALONE`, so we silently disable parallel apply in many unit
tests. This change increases test coverage (and we'll need to do it
eventually anyway if we make parallel apply the unconditional default).
2nd, this allows us to use parallel apply in apply load tests, which
significantly reduces variance of tests like max-sac-tps test.

This touches a few tests I'm not very familiar with (and was AI
assisted), but it looks reasonable to me and CI passes, but I'd like for
someone more familiar with these unit tests to take a look.

# 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: SirTyson

docs/apply-load-max-sac-tps.cfg

@@ -39,7 +39,6 @@ APPLY_LOAD_BL_LAST_BATCH_LEDGERS = 0
 # Minimal core config boilerplate
 
 RUN_STANDALONE=true
-PARALLEL_LEDGER_APPLY=false
 NODE_IS_VALIDATOR=true
 UNSAFE_QUORUM=true
 NETWORK_PASSPHRASE="Standalone Network ; February 2017"

src/bucket/test/BucketManagerTests.cpp

@@ -692,6 +692,10 @@ TEST_CASE_VERSIONS(
     cfg.MAX_CONCURRENT_SUBPROCESSES = 1;
     cfg.ARTIFICIALLY_ACCELERATE_TIME_FOR_TESTING = true;
     cfg.ARTIFICIALLY_PESSIMIZE_MERGES_FOR_TESTING = true;
+    // Test loop calls forgetUnreferencedBuckets and
+    // setNextLedgerEntryBatchForBucketTesting while ledgers close
+    // automatically, which races with background apply.
+    cfg.PARALLEL_LEDGER_APPLY = false;
     stellar::historytestutils::TmpDirHistoryConfigurator tcfg;
     cfg = tcfg.configure(cfg, true);
 

src/herder/test/HerderTests.cpp

@@ -5282,6 +5282,11 @@ externalize(SecretKey const& sk, LedgerManager& lm, HerderImpl& herder,
                                 xdr::xvector<UpgradeType, 6>{}, sk);
     herder.getHerderSCPDriver().valueExternalized(ledgerSeq,
                                                   xdr::xdr_to_opaque(sv));
+    // With background apply, crank until the ledger is fully applied
+    while (lm.getLastClosedLedgerNum() < ledgerSeq)
+    {
+        app.getClock().crank(true);
+    }
 }
 
 TEST_CASE("do not flood invalid transactions", "[herder]")

src/herder/test/TransactionQueueTests.cpp

@@ -3069,6 +3069,12 @@ TEST_CASE("remove applied", "[herder][transactionqueue]")
             app->getConfig().NODE_SEED);
         herder.getHerderSCPDriver().valueExternalized(ledgerSeq,
                                                       xdr::xdr_to_opaque(sv));
+
+        // With background apply, crank until the ledger is fully applied
+        while (lm.getLastClosedLedgerNum() < ledgerSeq)
+        {
+            clock.crank(true);
+        }
     }
 
     REQUIRE(tq.getTransactions({}).size() == 1);

src/history/test/HistoryTests.cpp

@@ -1599,6 +1599,14 @@ TEST_CASE_VERSIONS(
 
             while (hm.getPublishQueueCount() != 1)
             {
+                // With background apply, wait for any in-progress
+                // ledger close to finish before writing to the shared
+                // test-entry vectors that finalizeLedgerTxnChanges
+                // reads on the apply thread.
+                while (lm.isApplying())
+                {
+                    clock.crank(true);
+                }
                 auto lcl = lm.getLastClosedLedgerHeader();
                 lcl.header.ledgerSeq += 1;
                 // Generate entries excluding soroban types to avoid worrying
@@ -1969,6 +1977,12 @@ TEST_CASE("Introduce and fix gap without starting catchup",
     // Fill in the second gap. All buffered ledgers should be applied, but we
     // wait for another ledger to close to get in sync
     catchupSimulation.externalizeLedger(herder, nextLedger + 4);
+
+    // With background apply, crank until all queued ledgers are applied
+    while (lm.getLastClosedLedgerNum() < nextLedger + 5)
+    {
+        app->getClock().crank(true);
+    }
     REQUIRE(lm.isSynced());
     REQUIRE(lam.getLargestLedgerSeqHeard() == lm.getLastClosedLedgerNum());
     REQUIRE(!lam.isCatchupInitialized());

src/invariant/test/ConservationOfLumensTests.cpp

@@ -315,6 +315,11 @@ TEST_CASE(
 {
     auto cfg = getTestConfig();
     cfg.INVARIANT_CHECKS = {"ConservationOfLumens"};
+    // This test directly modifies LedgerTxnRoot header (totalCoins), which
+    // creates a hash mismatch between the DB header and what SCP externalized.
+    // This is incompatible with background apply where the cross-check runs on
+    // a thread that doesn't have access to the cached LCL header.
+    cfg.PARALLEL_LEDGER_APPLY = false;
 
     SorobanTest test(cfg);
 
@@ -378,6 +383,9 @@ TEST_CASE("ConservationOfLumens snapshot invariant detects bucket corruption",
     auto cfg = getTestConfig();
     cfg.INVARIANT_CHECKS = {}; // Disable automatic invariant checks because we
                                // will invoke it manually
+    // This test directly modifies LedgerTxnRoot header (totalCoins), which is
+    // incompatible with background apply (see comment in the test above).
+    cfg.PARALLEL_LEDGER_APPLY = false;
 
     VirtualClock clock;
     auto app = createTestApplication<BucketTestUtils::BucketTestApplication>(

src/ledger/LedgerManagerImpl.cpp

@@ -1519,6 +1519,17 @@ LedgerManagerImpl::applyLedger(LedgerCloseData const& ledgerData,
         CLOG_ERROR(Ledger, "{}", xdrToCerealString(prevHeader, "Full LCL"));
         CLOG_ERROR(Ledger, "{}", POSSIBLY_CORRUPTED_LOCAL_DATA);
 
+#ifdef BUILD_TESTS
+        if (!threadIsMain())
+        {
+            throw std::runtime_error(
+                "txset mismatch on background apply thread. This usually means "
+                "a test directly modified the LedgerTxnRoot header (e.g. "
+                "totalCoins). Set cfg.PARALLEL_LEDGER_APPLY = false for such "
+                "tests.");
+        }
+#endif
+
         throw std::runtime_error("txset mismatch");
     }
 

src/ledger/test/LedgerCloseMetaStreamTests.cpp

@@ -248,6 +248,13 @@ TEST_CASE("METADATA_DEBUG_LEDGERS works", "[metadebug]")
     {
         // Generate just enough meta to not triggers garbage collection
         closeLedgers(cfg.METADATA_DEBUG_LEDGERS);
+
+        // Drain any remaining background apply before stopping, so the
+        // debug tx set file and LCL are consistent when we read them.
+        while (lm.isApplying())
+        {
+            clock.crank(true);
+        }
         app->gracefulStop();
 
         // Verify presence of the latest debug tx set

src/main/ApplicationImpl.cpp

@@ -987,6 +987,15 @@ ApplicationImpl::manualClose(std::optional<uint32_t> const& manualLedgerSeq,
 
         if (mConfig.RUN_STANDALONE)
         {
+            // With background apply, triggerNextLedger posts work to the
+            // apply thread. Crank until the ledger is fully applied and
+            // LCL has advanced.
+            while (getLedgerManager().getLastClosedLedgerNum() <
+                   targetLedgerSeq)
+            {
+                getClock().crank(true);
+            }
+
             auto const newLedgerSeq =
                 getLedgerManager().getLastClosedLedgerNum();
             if (newLedgerSeq != targetLedgerSeq)
@@ -1563,7 +1572,15 @@ ApplicationImpl::postOnLedgerCloseThread(std::function<void()>&& f,
     asio::post(*mLedgerCloseIOContext, [this, f = std::move(f), isSlow]() {
         JITTER_INJECT_DELAY();
         mPostOnLedgerCloseThreadDelay.Update(isSlow.checkElapsedTime());
-        f();
+        try
+        {
+            f();
+        }
+        catch (...)
+        {
+            getClock().finishedBackgroundWork();
+            throw;
+        }
         getClock().finishedBackgroundWork();
     });
 }

src/main/CommandLine.cpp

@@ -1921,6 +1921,9 @@ runApplyLoad(CommandLineArgs const& args)
 
                 // Apply Load may exceed TX_SET byte size limits, so ignore them
                 config.IGNORE_MESSAGE_LIMITS_FOR_TESTING = true;
+
+                // Always use background ledger close for max-sac-tps
+                config.PARALLEL_LEDGER_APPLY = true;
             }
 
             VirtualClock clock(VirtualClock::REAL_TIME);