Test fixes and code hardening in preparation for parallel features (#5022)

As we're planning to enable various parallel features by default, this
PR cleans up some test paths and hardens the code. Concrete changes are:
- Crank in tests to support async apply flow
- Harden shutdown paths (avoid calling LM in App destructor - if BM
initialization throws, LM is null)
- Don't parallel apply in standalone mode (as it expects immediate
result)
- Test fixes to properly work with Postgres (e.g. make timestamps valid
to avoid pg complaining)

Author: marta-lokhova

src/bucket/test/BucketTestUtils.cpp

@@ -102,10 +102,10 @@ closeLedger(Application& app, std::optional<SecretKey> skToSignValue,
     app.getHerder().externalizeValue(TxSetXDRFrame::makeEmpty(lcl), ledgerNum,
                                      lcl.header.scpValue.closeTime, upgrades,
                                      skToSignValue);
-    // NB: this assert will probably stop being true when background apply is
-    // turned on by default: externalize will have handed the ledger off to
-    // apply but not yet received the results of apply or updated LCL. The fix
-    // should be just to crank here until LCL advances to ledgerSeq.
+    while (lm.getLastClosedLedgerNum() < ledgerNum)
+    {
+        app.getClock().crank(true);
+    }
     releaseAssert(lm.getLastClosedLedgerNum() == ledgerNum);
     return lm.getLastClosedLedgerHeader().hash;
 }
@@ -235,11 +235,13 @@ LedgerManagerForBucketTests::finalizeLedgerTxnChanges(
 
                 LedgerTxn ltxEvictions(ltx);
 
+                auto sorobanConfig =
+                    SorobanNetworkConfig::loadFromLedger(ltxEvictions);
+
                 auto evictedState =
                     mApp.getBucketManager().resolveBackgroundEvictionScan(
                         ltxEvictions, lh.ledgerSeq, keys, initialLedgerVers,
-                        mApp.getLedgerManager()
-                            .getLastClosedSorobanNetworkConfig());
+                        sorobanConfig);
                 if (protocolVersionStartsFrom(
                         initialLedgerVers,
                         LiveBucket::

src/herder/HerderImpl.cpp

@@ -936,6 +936,10 @@ HerderImpl::externalizeValue(TxSetXDRFrameConstPtr txSet, uint32_t ledgerSeq,
     StellarValue sv =
         makeStellarValue(txSet->getContentsHash(), closeTime, upgrades, sk);
     getHerderSCPDriver().valueExternalized(ledgerSeq, xdr::xdr_to_opaque(sv));
+    while (mApp.getLedgerManager().getLastClosedLedgerNum() < ledgerSeq)
+    {
+        mApp.getClock().crank(true);
+    }
 }
 
 bool

src/ledger/LedgerManagerImpl.cpp

@@ -267,7 +267,7 @@ LedgerManagerImpl::ApplyState::getSorobanInMemoryStateSizeForTesting() const
 void
 LedgerManagerImpl::ApplyState::threadInvariant() const
 {
-    if (mAppConnector.getConfig().EXPERIMENTAL_PARALLEL_LEDGER_APPLY)
+    if (mAppConnector.getConfig().parallelLedgerClose())
     {
         releaseAssert(threadIsMain() || mAppConnector.threadIsType(
                                             Application::ThreadType::APPLY));

src/main/ApplicationImpl.cpp

@@ -633,7 +633,7 @@ ApplicationImpl::~ApplicationImpl()
     mStopping = true;
     try
     {
-        idempotentShutdown();
+        idempotentShutdown(false);
     }
     catch (std::exception const& e)
     {
@@ -821,7 +821,7 @@ ApplicationImpl::start()
 }
 
 void
-ApplicationImpl::idempotentShutdown()
+ApplicationImpl::idempotentShutdown(bool forgetBuckets)
 {
     // Graceful shutdown sequence:
     // Perform a graceful shutdown by first signaling all managers to stop
@@ -850,11 +850,14 @@ ApplicationImpl::idempotentShutdown()
     }
     if (mBucketManager)
     {
-        // This call happens in shutdown -- before destruction -- so that we can
-        // be sure other subsystems (ledger etc.) are still alive and we can
-        // call into them to figure out which buckets _are_ referenced.
-        mBucketManager->forgetUnreferencedBuckets(
-            mLedgerManager->getLastClosedLedgerHAS());
+        if (forgetBuckets)
+        {
+            // This call happens in shutdown -- before destruction -- so that we
+            // can be sure other subsystems (ledger etc.) are still alive and we
+            // can call into them to figure out which buckets _are_ referenced.
+            mBucketManager->forgetUnreferencedBuckets(
+                mLedgerManager->getLastClosedLedgerHAS());
+        }
         mBucketManager->shutdown();
     }
     if (mHerder)
@@ -875,7 +878,7 @@ ApplicationImpl::gracefulStop()
         return;
     }
     mStopping = true;
-    idempotentShutdown();
+    idempotentShutdown(true);
 
     mStoppingTimer.expires_from_now(
         std::chrono::seconds(SHUTDOWN_DELAY_SECONDS));
@@ -900,7 +903,7 @@ ApplicationImpl::shutdownWorkScheduler()
     }
 }
 
-void
+bool
 ApplicationImpl::shutdownThread(
     std::unique_ptr<std::thread>& threadPtr,
     std::unique_ptr<asio::io_context::work>& workPtr,
@@ -919,23 +922,29 @@ ApplicationImpl::shutdownThread(
         LOG_INFO(DEFAULT_LOG, "Joining {} thread", threadName);
         threadPtr->join();
         threadPtr.reset();
+        return true;
     }
+    return false;
 }
 
 void
 ApplicationImpl::joinAllThreads()
 {
-    uint32_t const THREAD_COUNT = 3 + mWorkerThreads.size();
-    shutdownThread(mLedgerCloseThread, mLedgerCloseWork, "ledger close");
+    uint32_t joined = 0;
+    joined +=
+        shutdownThread(mLedgerCloseThread, mLedgerCloseWork, "ledger close");
     for (auto& w : mWorkerThreads)
     {
-        shutdownThread(w, mWork, "worker");
+        joined += shutdownThread(w, mWork, "worker");
     }
     mWorkerThreads.clear();
 
-    shutdownThread(mOverlayThread, mOverlayWork, "overlay");
-    shutdownThread(mEvictionThread, mEvictionWork, "eviction");
-    LOG_INFO(DEFAULT_LOG, "Joined all {} threads", THREAD_COUNT);
+    joined += shutdownThread(mOverlayThread, mOverlayWork, "overlay");
+    joined += shutdownThread(mEvictionThread, mEvictionWork, "eviction");
+    if (joined)
+    {
+        LOG_INFO(DEFAULT_LOG, "Joined all {} threads", joined);
+    }
 }
 
 std::string

src/main/ApplicationImpl.h

@@ -292,8 +292,10 @@ class ApplicationImpl : public Application
     void upgradeToCurrentSchemaAndMaybeRebuildLedger(bool applyBuckets,
                                                      bool forceRebuild);
 
-    void idempotentShutdown();
-    void shutdownThread(std::unique_ptr<std::thread>& threadPtr,
+    // Set `forgetBuckets` to true to clean up unreferenced buckets
+    // Note: this flag requires LM and BM to be fully constructed
+    void idempotentShutdown(bool forgetBuckets);
+    bool shutdownThread(std::unique_ptr<std::thread>& threadPtr,
                         std::unique_ptr<asio::io_context::work>& workPtr,
                         std::string const& threadName);
 };

src/main/Config.cpp

@@ -1916,7 +1916,8 @@ Config::processConfig(std::shared_ptr<cpptoml::table> t)
         {
             std::string msg =
                 "Invalid configuration: EXPERIMENTAL_PARALLEL_LEDGER_APPLY "
-                "does not support SQLite. Either switch to Postgres or set "
+                "does not support SQLite or RUN_STANDALONE mode. Either switch "
+                "to Postgres or set "
                 "EXPERIMENTAL_PARALLEL_LEDGER_APPLY=false";
             throw std::runtime_error(msg);
         }
@@ -2479,8 +2480,10 @@ Config::allBucketsInMemory() const
 bool
 Config::parallelLedgerClose() const
 {
+    // Standalone mode expects synchronous ledger application
     return EXPERIMENTAL_PARALLEL_LEDGER_APPLY &&
-           !(DATABASE.value.find("sqlite3://") != std::string::npos);
+           !(DATABASE.value.find("sqlite3://") != std::string::npos) &&
+           !RUN_STANDALONE;
 }
 
 void

src/main/test/CommandHandlerTests.cpp

@@ -198,6 +198,7 @@ TEST_CASE("manualclose", "[commandhandler]")
         auto& commandHandler = app->getCommandHandler();
         std::string retStr;
         issue(commandHandler, retStr);
+        app->gracefulStop();
     };
 
     SECTION("'manualclose' is forbidden if MANUAL_CLOSE is not configured")

src/overlay/test/OverlayTests.cpp

@@ -2253,8 +2253,10 @@ TEST_CASE("database is purged at overlay start", "[overlay]")
     auto app = createTestApplication(clock, cfg, true, false);
     auto& om = app->getOverlayManager();
     auto& peerManager = om.getPeerManager();
-    auto record = [](size_t numFailures) {
-        return PeerRecord{{}, numFailures, static_cast<int>(PeerType::INBOUND)};
+    auto record = [app](size_t numFailures) {
+        return PeerRecord{
+            VirtualClock::systemPointToTm(app->getClock().system_now()),
+            numFailures, static_cast<int>(PeerType::INBOUND)};
     };
 
     // Need to set max tx size on tests that start OverlayManager without

src/overlay/test/PeerManagerTests.cpp

@@ -513,8 +513,10 @@ TEST_CASE("purge peer table", "[overlay][PeerManager]")
     VirtualClock clock;
     auto app = createTestApplication(clock, getTestConfig());
     auto& peerManager = app->getOverlayManager().getPeerManager();
-    auto record = [](size_t numFailures) {
-        return PeerRecord{{}, numFailures, static_cast<int>(PeerType::INBOUND)};
+    auto record = [&app](size_t numFailures) {
+        return PeerRecord{
+            VirtualClock::systemPointToTm(app->getClock().system_now()),
+            numFailures, static_cast<int>(PeerType::INBOUND)};
     };
 
     peerManager.store(localhost(1), record(1), false);

src/test/TxTests.cpp

@@ -594,10 +594,10 @@ closeLedgerOn(Application& app, uint32 ledgerSeq, TimePoint closeTime,
     }
     app.getHerder().externalizeValue(txSet.first, ledgerSeq, closeTime,
                                      upgrades);
-    // NB: this assert will probably stop being true when background apply is
-    // turned on by default: externalize will have handed the ledger off to
-    // apply but not yet received the results of apply or updated LCL. The fix
-    // should be just to crank here until LCL advances to ledgerSeq.
+    while (app.getLedgerManager().getLastClosedLedgerNum() < ledgerSeq)
+    {
+        app.getClock().crank(true);
+    }
     releaseAssert(app.getLedgerManager().getLastClosedLedgerNum() == ledgerSeq);
     auto& lm = static_cast<LedgerManagerImpl&>(app.getLedgerManager());
     return lm.mLatestTxResultSet;