Merge pull request #4166 from marta-lokhova/startup_path

Rewrite state loading path on startup

Reviewed-by: dmkozh

Author: latobarita

src/bucket/BucketManager.h

@@ -270,9 +270,10 @@ class BucketManager : NonMovableOrCopyable
     checkForMissingBucketsFiles(HistoryArchiveState const& has) = 0;
 
     // Assume state from `has` in BucketList: find and attach all buckets in
-    // `has`, set current BL. Note: Does not restart merging
+    // `has`, set current BL.
     virtual void assumeState(HistoryArchiveState const& has,
-                             uint32_t maxProtocolVersion) = 0;
+                             uint32_t maxProtocolVersion,
+                             bool restartMerges) = 0;
 
     virtual void shutdown() = 0;
 

src/bucket/BucketManagerImpl.cpp

@@ -1048,7 +1048,7 @@ BucketManagerImpl::checkForMissingBucketsFiles(HistoryArchiveState const& has)
 
 void
 BucketManagerImpl::assumeState(HistoryArchiveState const& has,
-                               uint32_t maxProtocolVersion)
+                               uint32_t maxProtocolVersion, bool restartMerges)
 {
     ZoneScoped;
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
@@ -1093,7 +1093,10 @@ BucketManagerImpl::assumeState(HistoryArchiveState const& has,
         mBucketList->getLevel(i).setNext(nextFuture);
     }
 
-    mBucketList->restartMerges(mApp, maxProtocolVersion, has.currentLedger);
+    if (restartMerges)
+    {
+        mBucketList->restartMerges(mApp, maxProtocolVersion, has.currentLedger);
+    }
     cleanupStaleFiles();
 }
 

src/bucket/BucketManagerImpl.h

@@ -166,7 +166,7 @@ class BucketManagerImpl : public BucketManager
     std::vector<std::string>
     checkForMissingBucketsFiles(HistoryArchiveState const& has) override;
     void assumeState(HistoryArchiveState const& has,
-                     uint32_t maxProtocolVersion) override;
+                     uint32_t maxProtocolVersion, bool restartMerges) override;
     void shutdown() override;
 
     bool isShutdown() const override;

src/catchup/ApplyBucketsWork.cpp

@@ -279,7 +279,8 @@ ApplyBucketsWork::doWork()
         CLOG_INFO(History, "ApplyBuckets : done, assuming state");
 
         // After all buckets applied, spawn assumeState work
-        addWork<AssumeStateWork>(mApplyState, mMaxProtocolVersion);
+        addWork<AssumeStateWork>(mApplyState, mMaxProtocolVersion,
+                                 /* restartMerges */ true);
         mSpawnedAssumeStateWork = true;
     }
 

src/catchup/AssumeStateWork.cpp

@@ -14,10 +14,12 @@ namespace stellar
 {
 AssumeStateWork::AssumeStateWork(Application& app,
                                  HistoryArchiveState const& has,
-                                 uint32_t maxProtocolVersion)
+                                 uint32_t maxProtocolVersion,
+                                 bool restartMerges)
     : Work(app, "assume-state", BasicWork::RETRY_NEVER)
     , mHas(has)
     , mMaxProtocolVersion(maxProtocolVersion)
+    , mRestartMerges(restartMerges)
 {
     // Maintain reference to all Buckets in HAS to avoid garbage collection,
     // including future buckets that have already finished merging
@@ -68,8 +70,10 @@ AssumeStateWork::doWork()
         // Add bucket files to BucketList and restart merges
         auto assumeStateCB = [&has = mHas,
                               maxProtocolVersion = mMaxProtocolVersion,
+                              restartMerges = mRestartMerges,
                               &buckets = mBuckets](Application& app) {
-            app.getBucketManager().assumeState(has, maxProtocolVersion);
+            app.getBucketManager().assumeState(has, maxProtocolVersion,
+                                               restartMerges);
 
             // Drop bucket references once assume state complete since buckets
             // now referenced by BucketList

src/catchup/AssumeStateWork.h

@@ -17,14 +17,15 @@ class AssumeStateWork : public Work
     HistoryArchiveState const& mHas;
     uint32_t const mMaxProtocolVersion;
     bool mWorkSpawned{false};
+    bool const mRestartMerges;
 
     // Keep strong reference to buckets in HAS so they are not garbage
     // collected during indexing
     std::vector<std::shared_ptr<Bucket>> mBuckets{};
 
   public:
     AssumeStateWork(Application& app, HistoryArchiveState const& has,
-                    uint32_t maxProtocolVersion);
+                    uint32_t maxProtocolVersion, bool restartMerges);
 
   protected:
     State doWork() override;

src/ledger/LedgerManager.h

@@ -147,9 +147,14 @@ class LedgerManager
     // Called by application lifecycle events, system startup.
     virtual void startNewLedger() = 0;
 
-    // loads the last ledger information from the database
-    // if handler is set, also loads bucket information and invokes handler.
-    virtual void loadLastKnownLedger(std::function<void()> handler) = 0;
+    // loads the last ledger information from the database with the following
+    // parameters:
+    //  * restoreBucketlist indicates whether to restore the bucket list fully,
+    //  and restart merges
+    //  * isLedgerStateReady indicates whether the ledger state is ready or is
+    //  still being rebuilt (in which case we can't yet load ledger entries)
+    virtual void loadLastKnownLedger(bool restoreBucketlist,
+                                     bool isLedgerStateReady) = 0;
 
     // Return true if core is currently rebuilding in-memory state via local
     // catchup

src/ledger/LedgerManagerImpl.cpp

@@ -333,11 +333,12 @@ setLedgerTxnHeader(LedgerHeader const& lh, Application& app)
 }
 
 void
-LedgerManagerImpl::loadLastKnownLedger(function<void()> handler)
+LedgerManagerImpl::loadLastKnownLedger(bool restoreBucketlist,
+                                       bool isLedgerStateReady)
 {
     ZoneScoped;
-    auto ledgerTime = mLedgerClose.TimeScope();
 
+    // Step 1. Load LCL state from the DB and extract latest ledger hash
     string lastLedger =
         mApp.getPersistentState().getState(PersistentState::kLastClosedLedger);
 
@@ -346,103 +347,119 @@ LedgerManagerImpl::loadLastKnownLedger(function<void()> handler)
         throw std::runtime_error(
             "No reference in DB to any last closed ledger");
     }
-    else
-    {
-        CLOG_INFO(Ledger, "Last closed ledger (LCL) hash is {}", lastLedger);
-        Hash lastLedgerHash = hexToBin256(lastLedger);
 
-        if (mApp.getConfig().MODE_STORES_HISTORY_LEDGERHEADERS)
+    CLOG_INFO(Ledger, "Last closed ledger (LCL) hash is {}", lastLedger);
+    Hash lastLedgerHash = hexToBin256(lastLedger);
+
+    // Step 2. Restore LedgerHeader from DB based on the ledger hash derived
+    // earlier, or verify we're at genesis if in no-history mode
+    std::optional<LedgerHeader> latestLedgerHeader;
+    if (mApp.getConfig().MODE_STORES_HISTORY_LEDGERHEADERS)
+    {
+        if (mRebuildInMemoryState)
         {
-            if (mRebuildInMemoryState)
+            LedgerHeader lh;
+            CLOG_INFO(Ledger,
+                      "Setting empty ledger while core rebuilds state: {}",
+                      ledgerAbbrev(lh));
+            setLedgerTxnHeader(lh, mApp);
+            latestLedgerHeader = lh;
+        }
+        else
+        {
+            auto currentLedger =
+                LedgerHeaderUtils::loadByHash(getDatabase(), lastLedgerHash);
+            if (!currentLedger)
             {
-                LedgerHeader lh;
-                CLOG_INFO(Ledger,
-                          "Setting empty ledger while core rebuilds state: {}",
-                          ledgerAbbrev(lh));
-                setLedgerTxnHeader(lh, mApp);
+                throw std::runtime_error("Could not load ledger from database");
             }
-            else
+            HistoryArchiveState has = getLastClosedLedgerHAS();
+            if (currentLedger->ledgerSeq != has.currentLedger)
             {
-                auto currentLedger = LedgerHeaderUtils::loadByHash(
-                    getDatabase(), lastLedgerHash);
-                if (!currentLedger)
-                {
-                    throw std::runtime_error(
-                        "Could not load ledger from database");
-                }
-                HistoryArchiveState has = getLastClosedLedgerHAS();
-                if (currentLedger->ledgerSeq != has.currentLedger)
-                {
-                    throw std::runtime_error(
-                        "Invalid database state: last known "
-                        "ledger does not agree with HAS");
-                }
-
-                CLOG_INFO(Ledger, "Loaded LCL header from database: {}",
-                          ledgerAbbrev(*currentLedger));
-                setLedgerTxnHeader(*currentLedger, mApp);
+                throw std::runtime_error("Invalid database state: last known "
+                                         "ledger does not agree with HAS");
             }
+
+            CLOG_INFO(Ledger, "Loaded LCL header from database: {}",
+                      ledgerAbbrev(*currentLedger));
+            setLedgerTxnHeader(*currentLedger, mApp);
+            latestLedgerHeader = *currentLedger;
         }
-        else
+    }
+    else
+    {
+        // In no-history mode, this method should only be called when
+        // the LCL is genesis.
+        releaseAssertOrThrow(mLastClosedLedger.hash == lastLedgerHash);
+        releaseAssertOrThrow(mLastClosedLedger.header.ledgerSeq ==
+                             GENESIS_LEDGER_SEQ);
+        CLOG_INFO(Ledger, "LCL is genesis: {}",
+                  ledgerAbbrev(mLastClosedLedger));
+        latestLedgerHeader = mLastClosedLedger.header;
+    }
+
+    releaseAssert(latestLedgerHeader.has_value());
+
+    // Step 3. Restore BucketList if we're doing a full core startup
+    // (startServices=true), OR when using BucketListDB
+    if (restoreBucketlist || mApp.getConfig().isUsingBucketListDB())
+    {
+        HistoryArchiveState has = getLastClosedLedgerHAS();
+        auto missing = mApp.getBucketManager().checkForMissingBucketsFiles(has);
+        auto pubmissing = mApp.getHistoryManager()
+                              .getMissingBucketsReferencedByPublishQueue();
+        missing.insert(missing.end(), pubmissing.begin(), pubmissing.end());
+        if (!missing.empty())
         {
-            // In no-history mode, this method should only be called when
-            // the LCL is genesis.
-            releaseAssertOrThrow(mLastClosedLedger.hash == lastLedgerHash);
-            releaseAssertOrThrow(mLastClosedLedger.header.ledgerSeq ==
-                                 GENESIS_LEDGER_SEQ);
-            CLOG_INFO(Ledger, "LCL is genesis: {}",
-                      ledgerAbbrev(mLastClosedLedger));
+            CLOG_ERROR(Ledger,
+                       "{} buckets are missing from bucket directory '{}'",
+                       missing.size(), mApp.getBucketManager().getBucketDir());
+            throw std::runtime_error("Bucket directory is corrupt");
         }
 
-        if (handler)
+        if (mApp.getConfig().MODE_ENABLES_BUCKETLIST)
         {
-            HistoryArchiveState has = getLastClosedLedgerHAS();
-            auto missing =
-                mApp.getBucketManager().checkForMissingBucketsFiles(has);
-            auto pubmissing = mApp.getHistoryManager()
-                                  .getMissingBucketsReferencedByPublishQueue();
-            missing.insert(missing.end(), pubmissing.begin(), pubmissing.end());
-            if (!missing.empty())
+            // Only restart merges in full startup mode. Many modes in core
+            // (standalone offline commands, in-memory setup) do not need to
+            // spin up expensive merge processes.
+            auto assumeStateWork =
+                mApp.getWorkScheduler().executeWork<AssumeStateWork>(
+                    has, latestLedgerHeader->ledgerVersion, restoreBucketlist);
+            if (assumeStateWork->getState() == BasicWork::State::WORK_SUCCESS)
             {
-                CLOG_ERROR(
-                    Ledger, "{} buckets are missing from bucket directory '{}'",
-                    missing.size(), mApp.getBucketManager().getBucketDir());
-                throw std::runtime_error("Bucket directory is corrupt");
+                CLOG_INFO(Ledger, "Assumed bucket-state for LCL: {}",
+                          ledgerAbbrev(*latestLedgerHeader));
             }
             else
             {
-                {
-                    LedgerTxn ltx(mApp.getLedgerTxnRoot());
-                    auto header = ltx.loadHeader();
-                    uint32_t ledgerVersion = header.current().ledgerVersion;
-                    if (mApp.getConfig().MODE_ENABLES_BUCKETLIST)
-                    {
-                        auto assumeStateWork =
-                            mApp.getWorkScheduler()
-                                .executeWork<AssumeStateWork>(has,
-                                                              ledgerVersion);
-                        if (assumeStateWork->getState() ==
-                            BasicWork::State::WORK_SUCCESS)
-                        {
-                            CLOG_INFO(Ledger,
-                                      "Assumed bucket-state for LCL: {}",
-                                      ledgerAbbrev(header.current()));
-                        }
-                        else
-                        {
-                            // Work should only fail during graceful shutdown
-                            releaseAssert(mApp.isStopping());
-                        }
-                    }
-                    advanceLedgerPointers(header.current());
-                }
-                handler();
+                // Work should only fail during graceful shutdown
+                releaseAssertOrThrow(mApp.isStopping());
             }
         }
-        else
+    }
+
+    // Step 4. Restore LedgerManager's internal state
+    advanceLedgerPointers(*latestLedgerHeader);
+
+    if (protocolVersionStartsFrom(latestLedgerHeader->ledgerVersion,
+                                  SOROBAN_PROTOCOL_VERSION))
+    {
+        if (isLedgerStateReady)
         {
+            // Step 5. If ledger state is ready and core is in v20, load network
+            // configs right away
             LedgerTxn ltx(mApp.getLedgerTxnRoot());
-            advanceLedgerPointers(ltx.loadHeader().current());
+            updateNetworkConfig(ltx);
+        }
+        else
+        {
+            // In some modes, e.g. in-memory, core's state is rebuilt
+            // asynchronously via catchup. In this case, we're not able to load
+            // the network config at this time, and instead must let catchup do
+            // it when ready.
+            CLOG_INFO(Ledger,
+                      "Ledger state is being rebuilt, network config will "
+                      "be loaded once the rebuild is done");
         }
     }
 }

src/ledger/LedgerManagerImpl.h

@@ -182,7 +182,8 @@ class LedgerManagerImpl : public LedgerManager
 
     void startNewLedger(LedgerHeader const& genesisLedger);
     void startNewLedger() override;
-    void loadLastKnownLedger(std::function<void()> handler) override;
+    void loadLastKnownLedger(bool restoreBucketlist,
+                             bool isLedgerStateReady) override;
     virtual bool rebuildingInMemoryState() override;
     virtual void setupInMemoryStateRebuild() override;
 

src/main/ApplicationImpl.cpp

@@ -550,15 +550,8 @@ ApplicationImpl::getJsonInfo(bool verbose)
 void
 ApplicationImpl::reportInfo(bool verbose)
 {
-    auto loadConfig = [this]() {
-        LedgerTxn ltx(getLedgerTxnRoot());
-        if (protocolVersionStartsFrom(ltx.loadHeader().current().ledgerVersion,
-                                      SOROBAN_PROTOCOL_VERSION))
-        {
-            getLedgerManager().updateNetworkConfig(ltx);
-        }
-    };
-    mLedgerManager->loadLastKnownLedger(loadConfig);
+    mLedgerManager->loadLastKnownLedger(/* restoreBucketlist */ false,
+                                        /* isLedgerStateReady */ true);
     LOG_INFO(DEFAULT_LOG, "Reporting application info");
     std::cout << getJsonInfo(verbose).toStyledString() << std::endl;
 }
@@ -793,67 +786,58 @@ ApplicationImpl::validateAndLogConfig()
 }
 
 void
-ApplicationImpl::start()
+ApplicationImpl::startServices()
 {
-    if (mStarted)
+    // restores Herder's state before starting overlay
+    mHerder->start();
+    // set known cursors before starting maintenance job
+    ExternalQueue ps(*this);
+    ps.setInitialCursors(mConfig.KNOWN_CURSORS);
+    mMaintainer->start();
+    if (mConfig.MODE_AUTO_STARTS_OVERLAY)
     {
-        CLOG_INFO(Ledger, "Skipping application start up");
-        return;
+        mOverlayManager->start();
+    }
+    auto npub = mHistoryManager->publishQueuedHistory();
+    if (npub != 0)
+    {
+        CLOG_INFO(Ledger, "Restarted publishing {} queued snapshots", npub);
+    }
+    if (mConfig.FORCE_SCP)
+    {
+        LOG_INFO(DEFAULT_LOG, "* ");
+        LOG_INFO(DEFAULT_LOG,
+                 "* Force-starting scp from the current db state.");
+        LOG_INFO(DEFAULT_LOG, "* ");
+
+        mHerder->bootstrap();
+    }
+    if (mConfig.AUTOMATIC_SELF_CHECK_PERIOD.count() != 0)
+    {
+        scheduleSelfCheck(true);
     }
-    CLOG_INFO(Ledger, "Starting up application");
-    mStarted = true;
 
     if (mConfig.TESTING_UPGRADE_DATETIME.time_since_epoch().count() != 0)
     {
         mHerder->setUpgrades(mConfig);
     }
+}
 
-    bool done = false;
-    mLedgerManager->loadLastKnownLedger([this, &done]() {
-        // Make sure to load Soroban network config before
-        // herder starts (tx queue configuration is based on it).
-        {
-            LedgerTxn ltx(getLedgerTxnRoot());
-            if (protocolVersionStartsFrom(
-                    ltx.loadHeader().current().ledgerVersion,
-                    SOROBAN_PROTOCOL_VERSION))
-            {
-                getLedgerManager().updateNetworkConfig(ltx);
-            }
-        }
-        // restores Herder's state before starting overlay
-        mHerder->start();
-        // set known cursors before starting maintenance job
-        ExternalQueue ps(*this);
-        ps.setInitialCursors(mConfig.KNOWN_CURSORS);
-        mMaintainer->start();
-        if (mConfig.MODE_AUTO_STARTS_OVERLAY)
-        {
-            mOverlayManager->start();
-        }
-        auto npub = mHistoryManager->publishQueuedHistory();
-        if (npub != 0)
-        {
-            CLOG_INFO(Ledger, "Restarted publishing {} queued snapshots", npub);
-        }
-        if (mConfig.FORCE_SCP)
-        {
-            LOG_INFO(DEFAULT_LOG, "* ");
-            LOG_INFO(DEFAULT_LOG,
-                     "* Force-starting scp from the current db state.");
-            LOG_INFO(DEFAULT_LOG, "* ");
+void
+ApplicationImpl::start()
+{
+    if (mStarted)
+    {
+        CLOG_INFO(Ledger, "Skipping application start up");
+        return;
+    }
 
-            mHerder->bootstrap();
-        }
-        if (mConfig.AUTOMATIC_SELF_CHECK_PERIOD.count() != 0)
-        {
-            scheduleSelfCheck(true);
-        }
-        done = true;
-    });
+    CLOG_INFO(Ledger, "Starting up application");
+    mStarted = true;
 
-    while (!done && mVirtualClock.crank(true))
-        ;
+    mLedgerManager->loadLastKnownLedger(/* restoreBucketlist */ true,
+                                        /* isLedgerStateReady */ true);
+    startServices();
 }
 
 void

src/main/ApplicationImpl.h

@@ -84,6 +84,7 @@ class ApplicationImpl : public Application
                                         std::string jobName) override;
 
     virtual void start() override;
+    void startServices();
 
     // Stops the worker io_context, which should cause the threads to exit once
     // they finish running any work-in-progress. If you want a more abrupt exit

src/main/ApplicationUtils.cpp

@@ -162,7 +162,9 @@ setupMinimalDBForInMemoryMode(Config const& cfg, uint32_t startAtLedger)
 
         if (!rebuildDB)
         {
-            app->getLedgerManager().loadLastKnownLedger(nullptr);
+            // Ledger state is not yet ready during this setup step
+            app->getLedgerManager().loadLastKnownLedger(
+                /* restoreBucketlist */ false, /* isLedgerStateReady */ false);
             auto lcl = app->getLedgerManager().getLastClosedLedgerNum();
             LOG_INFO(DEFAULT_LOG, "Current in-memory state, got LCL: {}", lcl);
             rebuildDB =
@@ -195,7 +197,10 @@ setupApp(Config& cfg, VirtualClock& clock, uint32_t startAtLedger,
         return nullptr;
     }
 
-    app->getLedgerManager().loadLastKnownLedger(nullptr);
+    // With in-memory mode, ledger state is not yet ready during this setup step
+    app->getLedgerManager().loadLastKnownLedger(
+        /* restoreBucketlist */ false,
+        /* isLedgerStateReady */ !cfg.isInMemoryMode());
     auto lcl = app->getLedgerManager().getLastClosedLedgerHeader();
 
     if (cfg.isInMemoryMode() &&
@@ -447,15 +452,8 @@ selfCheck(Config cfg)
 
     // We run self-checks from a "loaded but dormant" state where the
     // application is not started, but the LM has loaded the LCL.
-    app->getLedgerManager().loadLastKnownLedger(nullptr);
-    {
-        LedgerTxn ltx(app->getLedgerTxnRoot());
-        if (protocolVersionStartsFrom(ltx.loadHeader().current().ledgerVersion,
-                                      SOROBAN_PROTOCOL_VERSION))
-        {
-            app->getLedgerManager().updateNetworkConfig(ltx);
-        }
-    }
+    app->getLedgerManager().loadLastKnownLedger(/* restoreBucketlist */ false,
+                                                /* isLedgerStateReady */ true);
 
     // First we schedule the cheap, asynchronous "online" checks that get run by
     // the HTTP "self-check" endpoint, and crank until they're done.
@@ -533,17 +531,11 @@ mergeBucketList(Config cfg, std::string const& outputDir)
     VirtualClock clock;
     cfg.setNoListen();
     Application::pointer app = Application::create(clock, cfg, false);
-    app->getLedgerManager().loadLastKnownLedger(nullptr);
     auto& lm = app->getLedgerManager();
-    {
-        LedgerTxn ltx(app->getLedgerTxnRoot());
-        if (protocolVersionStartsFrom(ltx.loadHeader().current().ledgerVersion,
-                                      SOROBAN_PROTOCOL_VERSION))
-        {
-            lm.updateNetworkConfig(ltx);
-        }
-    }
     auto& bm = app->getBucketManager();
+
+    lm.loadLastKnownLedger(/* restoreBucketlist */ false,
+                           /* isLedgerStateReady */ true);
     HistoryArchiveState has = lm.getLastClosedLedgerHAS();
     auto bucket = bm.mergeBuckets(has);
 
@@ -578,16 +570,10 @@ dumpLedger(Config cfg, std::string const& outputFile,
     VirtualClock clock;
     cfg.setNoListen();
     Application::pointer app = Application::create(clock, cfg, false);
-    app->getLedgerManager().loadLastKnownLedger(nullptr);
     auto& lm = app->getLedgerManager();
-    {
-        LedgerTxn ltx(app->getLedgerTxnRoot());
-        if (protocolVersionStartsFrom(ltx.loadHeader().current().ledgerVersion,
-                                      SOROBAN_PROTOCOL_VERSION))
-        {
-            lm.updateNetworkConfig(ltx);
-        }
-    }
+
+    lm.loadLastKnownLedger(/* restoreBucketlist */ false,
+                           /* isLedgerStateReady */ true);
     HistoryArchiveState has = lm.getLastClosedLedgerHAS();
     std::optional<uint32_t> minLedger;
     if (lastModifiedLedgerCount)