diff --git a/src/bucket/BucketManagerImpl.cpp b/src/bucket/BucketManagerImpl.cpp
index 83ce4adb53..3ca87545d1 100644
--- a/src/bucket/BucketManagerImpl.cpp
+++ b/src/bucket/BucketManagerImpl.cpp
@@ -107,7 +107,10 @@ BucketManagerImpl::BucketManagerImpl(Application& app)
     , mBucketSnapMerge(app.getMetrics().NewTimer({"bucket", "snap", "merge"}))
     , mSharedBucketsSize(
           app.getMetrics().NewCounter({"bucket", "memory", "shared"}))
-    , mDeleteEntireBucketDirInDtor(app.getConfig().isInMemoryMode())
+    // Minimal DB is stored in the buckets dir, so delete it only when
+    // mode does not use minimal DB
+    , mDeleteEntireBucketDirInDtor(
+          app.getConfig().isInMemoryModeWithoutMinimalDB())
 {
 }
 
diff --git a/src/herder/HerderImpl.cpp b/src/herder/HerderImpl.cpp
index 756524d645..8acb21de98 100644
--- a/src/herder/HerderImpl.cpp
+++ b/src/herder/HerderImpl.cpp
@@ -173,7 +173,7 @@ HerderImpl::processExternalized(uint64 slotIndex, StellarValue const& value)
     TxSetFramePtr externalizedSet = mPendingEnvelopes.getTxSet(value.txSetHash);
 
     // save the SCP messages in the database
-    if (mApp.getConfig().MODE_STORES_HISTORY)
+    if (mApp.getConfig().MODE_STORES_HISTORY_MISC)
     {
         mApp.getHerderPersistence().saveSCPHistory(
             static_cast<uint32>(slotIndex),
diff --git a/src/herder/test/HerderTests.cpp b/src/herder/test/HerderTests.cpp
index b37c2c0437..1aa277c640 100644
--- a/src/herder/test/HerderTests.cpp
+++ b/src/herder/test/HerderTests.cpp
@@ -2586,7 +2586,7 @@ TEST_CASE("slot herder policy", "[herder]")
     Config cfg(getTestConfig());
 
     // start in sync
-    cfg.FORCE_SCP = true;
+    cfg.FORCE_SCP = false;
     cfg.MANUAL_CLOSE = false;
     cfg.NODE_SEED = v0SecretKey;
     cfg.MAX_SLOTS_TO_REMEMBER = 5;
@@ -2599,6 +2599,7 @@ TEST_CASE("slot herder policy", "[herder]")
 
     VirtualClock clock;
     Application::pointer app = createTestApplication(clock, cfg);
+    app->start();
 
     auto& herder = static_cast<HerderImpl&>(app->getHerder());
 
diff --git a/src/ledger/LedgerManagerImpl.cpp b/src/ledger/LedgerManagerImpl.cpp
index b91c2fc372..672fa8ac2e 100644
--- a/src/ledger/LedgerManagerImpl.cpp
+++ b/src/ledger/LedgerManagerImpl.cpp
@@ -260,7 +260,7 @@ LedgerManagerImpl::loadLastKnownLedger(
         CLOG_INFO(Ledger, "Last closed ledger (LCL) hash is {}", lastLedger);
         Hash lastLedgerHash = hexToBin256(lastLedger);
 
-        if (mApp.getConfig().MODE_STORES_HISTORY)
+        if (mApp.getConfig().MODE_STORES_HISTORY_LEDGERHEADERS)
         {
             auto currentLedger =
                 LedgerHeaderUtils::loadByHash(getDatabase(), lastLedgerHash);
@@ -663,7 +663,7 @@ LedgerManagerImpl::closeLedger(LedgerCloseData const& ledgerData)
             }
             // Note: Index from 1 rather than 0 to match the behavior of
             // storeTransaction and storeTransactionFee.
-            if (mApp.getConfig().MODE_STORES_HISTORY)
+            if (mApp.getConfig().MODE_STORES_HISTORY_MISC)
             {
                 Upgrades::storeUpgradeHistory(getDatabase(), ledgerSeq,
                                               lupgrade, changes,
@@ -896,7 +896,7 @@ LedgerManagerImpl::processFeesSeqNums(
             // txs counting from 1, not 0. We preserve this for the time being
             // in case anyone depends on it.
             ++index;
-            if (mApp.getConfig().MODE_STORES_HISTORY)
+            if (mApp.getConfig().MODE_STORES_HISTORY_MISC)
             {
                 storeTransactionFee(mApp.getDatabase(), ledgerSeq, tx, changes,
                                     index);
@@ -1014,7 +1014,7 @@ LedgerManagerImpl::applyTransactions(
         // txs counting from 1, not 0. We preserve this for the time being
         // in case anyone depends on it.
         ++index;
-        if (mApp.getConfig().MODE_STORES_HISTORY)
+        if (mApp.getConfig().MODE_STORES_HISTORY_MISC)
         {
             auto ledgerSeq = ltx.loadHeader().current().ledgerSeq;
             storeTransaction(mApp.getDatabase(), ledgerSeq, tx, tm,
@@ -1044,7 +1044,7 @@ void
 LedgerManagerImpl::storeCurrentLedger(LedgerHeader const& header)
 {
     ZoneScoped;
-    if (mApp.getConfig().MODE_STORES_HISTORY)
+    if (mApp.getConfig().MODE_STORES_HISTORY_LEDGERHEADERS)
     {
         LedgerHeaderUtils::storeInDatabase(mApp.getDatabase(), header);
     }
diff --git a/src/main/Application.h b/src/main/Application.h
index a0234299b6..ea4ffafdea 100644
--- a/src/main/Application.h
+++ b/src/main/Application.h
@@ -285,6 +285,8 @@ class Application
 
     virtual AbstractLedgerTxnParent& getLedgerTxnRoot() = 0;
 
+    virtual void validateAndLogConfig() = 0;
+
     // Factory: create a new Application object bound to `clock`, with a local
     // copy made of `cfg`
     static pointer create(VirtualClock& clock, Config const& cfg,
@@ -297,10 +299,16 @@ class Application
         auto ret = std::make_shared<T>(clock, cfg, std::forward<Args>(args)...);
         ret->initialize(newDB);
         validateNetworkPassphrase(ret);
+        ret->validateAndLogConfig();
 
         return ret;
     }
 
+    // This method is used in in-memory mode: when rebuilding state from buckets
+    // is not possible, this method resets the database state back to genesis
+    // (while preserving the overlay data).
+    virtual void resetDBForInMemoryMode() = 0;
+
   protected:
     Application()
     {
diff --git a/src/main/ApplicationImpl.cpp b/src/main/ApplicationImpl.cpp
index fa6957758f..20e5c95810 100644
--- a/src/main/ApplicationImpl.cpp
+++ b/src/main/ApplicationImpl.cpp
@@ -40,6 +40,7 @@
 #include "medida/timer.h"
 #include "overlay/BanManager.h"
 #include "overlay/OverlayManager.h"
+#include "overlay/OverlayManagerImpl.h"
 #include "process/ProcessManager.h"
 #include "scp/LocalNode.h"
 #include "scp/QuorumSetUtils.h"
@@ -135,13 +136,24 @@ ApplicationImpl::ApplicationImpl(VirtualClock& clock, Config const& cfg)
 void
 ApplicationImpl::initialize(bool createNewDB)
 {
+    // Subtle: initialize the bucket manager first before initializing the
+    // database. This is needed as some modes in core (such as in-memory) use a
+    // small database inside the bucket directory.
+    mBucketManager = BucketManager::create(*this);
+
+    bool initNewDB =
+        createNewDB || mConfig.DATABASE.value == "sqlite3://:memory:";
+    if (initNewDB)
+    {
+        mBucketManager->dropAll();
+    }
+
     mDatabase = createDatabase();
     mPersistentState = std::make_unique<PersistentState>(*this);
     mOverlayManager = createOverlayManager();
     mLedgerManager = createLedgerManager();
     mHerder = createHerder();
     mHerderPersistence = HerderPersistence::create(*this);
-    mBucketManager = BucketManager::create(*this);
     mCatchupManager = CatchupManager::create(*this);
     mHistoryArchiveManager = std::make_unique<HistoryArchiveManager>(*this);
     mHistoryManager = HistoryManager::create(*this);
@@ -192,13 +204,13 @@ ApplicationImpl::initialize(bool createNewDB)
     SponsorshipCountIsValid::registerInvariant(*this);
     enableInvariantsFromConfig();
 
-    if (createNewDB || mConfig.DATABASE.value == "sqlite3://:memory:")
+    if (initNewDB)
     {
         newDB();
     }
     else
     {
-        upgradeDB();
+        mDatabase->upgradeToCurrentSchema();
     }
 
     // Subtle: process manager should come to existence _after_ BucketManager
@@ -213,16 +225,9 @@ ApplicationImpl::newDB()
 {
     mDatabase->initialize();
     mDatabase->upgradeToCurrentSchema();
-    mBucketManager->dropAll();
     mLedgerManager->startNewLedger();
 }
 
-void
-ApplicationImpl::upgradeDB()
-{
-    mDatabase->upgradeToCurrentSchema();
-}
-
 void
 ApplicationImpl::reportCfgMetrics()
 {
@@ -394,6 +399,18 @@ ApplicationImpl::~ApplicationImpl()
     LOG_INFO(DEFAULT_LOG, "Application destroyed");
 }
 
+void
+ApplicationImpl::resetDBForInMemoryMode()
+{
+    // Load the peer information and reinitialize the DB
+    auto& pm = getOverlayManager().getPeerManager();
+    auto peerData = pm.loadAllPeers();
+    newDB();
+    pm.storePeers(peerData);
+
+    LOG_INFO(DEFAULT_LOG, "In-memory state is reset back to genesis");
+}
+
 uint64_t
 ApplicationImpl::timeNow()
 {
@@ -401,21 +418,8 @@ ApplicationImpl::timeNow()
 }
 
 void
-ApplicationImpl::start()
+ApplicationImpl::validateAndLogConfig()
 {
-    if (mStarted)
-    {
-        CLOG_INFO(Ledger, "Skipping application start up");
-        return;
-    }
-    CLOG_INFO(Ledger, "Starting up application");
-    mStarted = true;
-
-    if (mConfig.TESTING_UPGRADE_DATETIME.time_since_epoch().count() != 0)
-    {
-        mHerder->setUpgrades(mConfig);
-    }
-
     if (mConfig.FORCE_SCP && !mConfig.NODE_IS_VALIDATOR)
     {
         throw std::invalid_argument(
@@ -454,10 +458,11 @@ ApplicationImpl::start()
 
     if (getHistoryArchiveManager().hasAnyWritableHistoryArchive())
     {
-        if (!mConfig.MODE_STORES_HISTORY)
+        if (!mConfig.modeStoresAllHistory())
         {
-            throw std::invalid_argument("MODE_STORES_HISTORY is not set, but "
-                                        "some history archives are writable");
+            throw std::invalid_argument(
+                "Core is not configured to store history, but "
+                "some history archives are writable");
         }
     }
 
@@ -467,6 +472,23 @@ ApplicationImpl::start()
     }
 
     mConfig.logBasicInfo();
+}
+
+void
+ApplicationImpl::start()
+{
+    if (mStarted)
+    {
+        CLOG_INFO(Ledger, "Skipping application start up");
+        return;
+    }
+    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,
diff --git a/src/main/ApplicationImpl.h b/src/main/ApplicationImpl.h
index ec5defe9cd..eba5f58ffc 100644
--- a/src/main/ApplicationImpl.h
+++ b/src/main/ApplicationImpl.h
@@ -94,6 +94,8 @@ class ApplicationImpl : public Application
     // returns.
     virtual void joinAllThreads() override;
 
+    virtual void validateAndLogConfig() override;
+
     virtual std::string
     manualClose(std::optional<uint32_t> const& manualLedgerSeq,
                 std::optional<TimePoint> const& manualCloseTime) override;
@@ -120,6 +122,8 @@ class ApplicationImpl : public Application
 
     virtual AbstractLedgerTxnParent& getLedgerTxnRoot() override;
 
+    virtual void resetDBForInMemoryMode() override;
+
   protected:
     std::unique_ptr<LedgerManager>
         mLedgerManager;              // allow to change that for tests
@@ -143,9 +147,9 @@ class ApplicationImpl : public Application
     asio::io_context mWorkerIOContext;
     std::unique_ptr<asio::io_context::work> mWork;
 
+    std::unique_ptr<BucketManager> mBucketManager;
     std::unique_ptr<Database> mDatabase;
     std::unique_ptr<OverlayManager> mOverlayManager;
-    std::unique_ptr<BucketManager> mBucketManager;
     std::unique_ptr<CatchupManager> mCatchupManager;
     std::unique_ptr<HerderPersistence> mHerderPersistence;
     std::unique_ptr<HistoryArchiveManager> mHistoryArchiveManager;
@@ -195,7 +199,6 @@ class ApplicationImpl : public Application
     Hash mNetworkID;
 
     void newDB();
-    void upgradeDB();
 
     void shutdownMainIOContext();
     void shutdownWorkScheduler();
diff --git a/src/main/ApplicationUtils.cpp b/src/main/ApplicationUtils.cpp
index 952cc437b8..7281eac402 100644
--- a/src/main/ApplicationUtils.cpp
+++ b/src/main/ApplicationUtils.cpp
@@ -7,12 +7,14 @@
 #include "bucket/BucketManager.h"
 #include "catchup/ApplyBucketsWork.h"
 #include "catchup/CatchupConfiguration.h"
+#include "crypto/Hex.h"
 #include "database/Database.h"
 #include "herder/Herder.h"
 #include "history/HistoryArchive.h"
 #include "history/HistoryArchiveManager.h"
 #include "history/HistoryArchiveReportWork.h"
 #include "historywork/GetHistoryArchiveStateWork.h"
+#include "ledger/LedgerHeaderUtils.h"
 #include "ledger/LedgerManager.h"
 #include "main/ErrorMessages.h"
 #include "main/ExternalQueue.h"
@@ -20,6 +22,7 @@
 #include "main/PersistentState.h"
 #include "main/StellarCoreVersion.h"
 #include "overlay/OverlayManager.h"
+#include "util/GlobalChecks.h"
 #include "util/Logging.h"
 #include "work/WorkScheduler.h"
 
@@ -30,75 +33,137 @@
 namespace stellar
 {
 
-int
-runWithConfig(Config cfg, std::optional<CatchupConfiguration> cc)
+bool
+canRebuildInMemoryLedgerFromBuckets(uint32_t startAtLedger, uint32_t lcl)
+{
+    // Number of streaming ledgers ahead of LCL. Core will
+    // rebuild the existing state if the difference between the start
+    // ledger and LCL is within this window.
+    uint32_t const RESTORE_STATE_LEDGER_WINDOW = 10;
+    // Do not rebuild genesis state
+    bool isGenesis = lcl == LedgerManager::GENESIS_LEDGER_SEQ;
+    return !isGenesis && startAtLedger >= lcl &&
+           startAtLedger - lcl <= RESTORE_STATE_LEDGER_WINDOW;
+}
+
+void
+setupMinimalDBForInMemoryMode(Config const& cfg, uint32_t startAtLedger)
 {
-    VirtualClock::Mode clockMode = VirtualClock::REAL_TIME;
+    releaseAssertOrThrow(cfg.isInMemoryMode());
+
+    VirtualClock clock;
+    Application::pointer app;
 
-    if (cfg.MANUAL_CLOSE)
+    // Look for an existing minimal database, and see if it's possible to
+    // restore ledger state from buckets. If it is not possible, reset the
+    // existing database back to genesis. If the minimal database does not
+    // exist, create a new one.
+    bool found = false;
+    try
     {
-        if (!cfg.NODE_IS_VALIDATOR)
-        {
-            LOG_ERROR(DEFAULT_LOG,
-                      "Starting stellar-core in MANUAL_CLOSE mode requires "
-                      "NODE_IS_VALIDATOR to be set");
-            return 1;
-        }
-        if (cfg.RUN_STANDALONE)
+        app = Application::create(clock, cfg, /* newDB */ false);
+        found = true;
+    }
+    catch (std::runtime_error const&)
+    {
+        LOG_INFO(DEFAULT_LOG, "Minimal database not found, creating one...");
+        app = Application::create(clock, cfg, /* newDB */ true);
+    }
+
+    // Rebuild the state from scratch if:
+    //  - --start-at-ledger was not provided
+    //  - target catchup ledger is before LCL
+    //  - target catchup ledger is too far ahead of LCL
+    // In all other cases, attempt restoring the ledger states via
+    // local bucket application
+    if (found)
+    {
+        LOG_INFO(DEFAULT_LOG, "Found the existing minimal database");
+        app->getLedgerManager().loadLastKnownLedger(nullptr);
+        auto lcl = app->getLedgerManager().getLastClosedLedgerNum();
+        LOG_INFO(DEFAULT_LOG, "Current in-memory state, got LCL: {}", lcl);
+
+        if (!canRebuildInMemoryLedgerFromBuckets(startAtLedger, lcl))
         {
-            clockMode = VirtualClock::VIRTUAL_TIME;
-            if (cfg.AUTOMATIC_MAINTENANCE_COUNT != 0 ||
-                cfg.AUTOMATIC_MAINTENANCE_PERIOD.count() != 0)
-            {
-                LOG_WARNING(
-                    DEFAULT_LOG,
-                    "Using MANUAL_CLOSE and RUN_STANDALONE together "
-                    "induces virtual time, which requires automatic "
-                    "maintenance to be disabled.  "
-                    "AUTOMATIC_MAINTENANCE_COUNT and "
-                    "AUTOMATIC_MAINTENANCE_PERIOD are being overridden to "
-                    "0.");
-                cfg.AUTOMATIC_MAINTENANCE_COUNT = 0;
-                cfg.AUTOMATIC_MAINTENANCE_PERIOD = std::chrono::seconds{0};
-            }
+            LOG_INFO(DEFAULT_LOG, "Cannot restore the in-memory state, "
+                                  "rebuilding the state from scratch");
+            app->resetDBForInMemoryMode();
         }
     }
+}
+
+Application::pointer
+setupApp(Config& cfg, VirtualClock& clock, uint32_t startAtLedger,
+         std::string const& startAtHash)
+{
+    if (cfg.isInMemoryMode())
+    {
+        setupMinimalDBForInMemoryMode(cfg, startAtLedger);
+    }
 
     LOG_INFO(DEFAULT_LOG, "Starting stellar-core {}", STELLAR_CORE_VERSION);
-    VirtualClock clock(clockMode);
     Application::pointer app;
-    try
+    app = Application::create(clock, cfg, false);
+    if (!app->getHistoryArchiveManager().checkSensibleConfig())
     {
-        cfg.COMMANDS.push_back("self-check");
-        app = Application::create(clock, cfg, false);
+        return nullptr;
+    }
 
-        if (!app->getHistoryArchiveManager().checkSensibleConfig())
-        {
-            return 1;
-        }
-        if (cfg.ARTIFICIALLY_ACCELERATE_TIME_FOR_TESTING)
+    bool doCatchupForInMemoryMode =
+        cfg.isInMemoryMode() && startAtLedger != 0 && !startAtHash.empty();
+    if (doCatchupForInMemoryMode)
+    {
+        app->getLedgerManager().loadLastKnownLedger(nullptr);
+        auto lcl = app->getLedgerManager().getLastClosedLedgerHeader();
+        if (canRebuildInMemoryLedgerFromBuckets(startAtLedger,
+                                                lcl.header.ledgerSeq))
         {
-            LOG_WARNING(
-                DEFAULT_LOG,
-                "Artificial acceleration of time enabled (for testing only)");
+            auto lclHashStr = binToHex(lcl.hash);
+            if (lcl.header.ledgerSeq == startAtLedger &&
+                lclHashStr != startAtHash)
+            {
+                LOG_ERROR(DEFAULT_LOG,
+                          "Provided hash {} does not agree with stored hash {}",
+                          startAtHash, lclHashStr);
+                return nullptr;
+            }
+            if (!applyBucketsForLCL(*app))
+            {
+                return nullptr;
+            }
         }
-
-        if (cc)
+        else
         {
+            LedgerNumHashPair pair;
+            pair.first = startAtLedger;
+            pair.second = std::optional<Hash>(hexToBin256(startAtHash));
+            auto mode = CatchupConfiguration::Mode::OFFLINE_COMPLETE;
             Json::Value catchupInfo;
-            // Allow catchup from any readable archive
-            int res = catchup(app, *cc, catchupInfo, /* archive */ nullptr);
+            int res =
+                catchup(app, CatchupConfiguration{pair, 0, mode}, catchupInfo,
+                        /* archive */ nullptr);
             if (res != 0)
             {
-                return res;
+                return nullptr;
             }
         }
-        else
-        {
-            app->start();
-        }
+    }
 
-        if (!cfg.MODE_AUTO_STARTS_OVERLAY)
+    return app;
+}
+
+int
+runApp(Application::pointer app)
+{
+    // Certain in-memory modes in core may start the app before reaching this
+    // point, but since start is idempotent, second call will just no-op
+    app->start();
+
+    // Perform additional startup procedures (must be done after the app is
+    // setup) and run the app
+    try
+    {
+        if (!app->getConfig().MODE_AUTO_STARTS_OVERLAY)
         {
             app->getHerder().restoreState();
             app->getOverlayManager().start();
@@ -115,11 +180,11 @@ runWithConfig(Config cfg, std::optional<CatchupConfiguration> cc)
 
     try
     {
-        auto& io = clock.getIOContext();
+        auto& io = app->getClock().getIOContext();
         asio::io_context::work mainWork(io);
         while (!io.stopped())
         {
-            clock.crank();
+            app->getClock().crank();
         }
     }
     catch (std::exception const& e)
@@ -130,6 +195,35 @@ runWithConfig(Config cfg, std::optional<CatchupConfiguration> cc)
     return 0;
 }
 
+bool
+applyBucketsForLCL(Application& app)
+{
+    auto has = app.getLedgerManager().getLastClosedLedgerHAS();
+    auto lclHash =
+        app.getPersistentState().getState(PersistentState::kLastClosedLedger);
+
+    // As the local HAS might have merges in progress, let
+    // `prepareForPublish` convert it into a "valid historical HAS".
+    has.prepareForPublish(app);
+
+    auto maxProtocolVersion = Config::CURRENT_LEDGER_PROTOCOL_VERSION;
+    auto currentLedger =
+        LedgerHeaderUtils::loadByHash(app.getDatabase(), hexToBin256(lclHash));
+    if (currentLedger)
+    {
+        maxProtocolVersion = currentLedger->ledgerVersion;
+    }
+
+    std::map<std::string, std::shared_ptr<Bucket>> buckets;
+    auto work = app.getWorkScheduler().scheduleWork<ApplyBucketsWork>(
+        buckets, has, maxProtocolVersion);
+
+    while (app.getClock().crank(true) && !work->isDone())
+        ;
+
+    return work->getState() == BasicWork::State::WORK_SUCCESS;
+}
+
 void
 httpCommand(std::string const& command, unsigned short port)
 {
@@ -272,16 +366,8 @@ rebuildLedgerFromBuckets(Config cfg)
     ps.setState(PersistentState::kNetworkPassphrase, pass);
 
     LOG_INFO(DEFAULT_LOG, "Applying buckets from LCL bucket list.");
-    std::map<std::string, std::shared_ptr<Bucket>> localBuckets;
-    auto& ws = app->getWorkScheduler();
-
-    HistoryArchiveState has;
-    has.fromString(hasStr);
-    has.prepareForPublish(*app);
+    auto ok = applyBucketsForLCL(*app);
 
-    auto applyBucketsWork = ws.executeWork<ApplyBucketsWork>(
-        localBuckets, has, Config::CURRENT_LEDGER_PROTOCOL_VERSION);
-    auto ok = applyBucketsWork->getState() == BasicWork::State::WORK_SUCCESS;
     if (ok)
     {
         tx.commit();
diff --git a/src/main/ApplicationUtils.h b/src/main/ApplicationUtils.h
index 075f0dd6bf..3eca06afe1 100644
--- a/src/main/ApplicationUtils.h
+++ b/src/main/ApplicationUtils.h
@@ -13,7 +13,11 @@ namespace stellar
 
 class CatchupConfiguration;
 
-int runWithConfig(Config cfg, std::optional<CatchupConfiguration> cc);
+// Create application and validate its configuration
+Application::pointer setupApp(Config& cfg, VirtualClock& clock,
+                              uint32_t startAtLedger,
+                              std::string const& startAtHash);
+int runApp(Application::pointer app);
 void setForceSCPFlag();
 void initializeDatabase(Config cfg);
 void httpCommand(std::string const& command, unsigned short port);
@@ -31,5 +35,8 @@ void writeCatchupInfo(Json::Value const& catchupInfo,
                       std::string const& outputFile);
 int catchup(Application::pointer app, CatchupConfiguration cc,
             Json::Value& catchupInfo, std::shared_ptr<HistoryArchive> archive);
+// Reduild ledger state based on the buckets. Ensure ledger state is properly
+// reset before calling this function.
+bool applyBucketsForLCL(Application& app);
 int publish(Application::pointer app);
 }
diff --git a/src/main/CommandHandler.cpp b/src/main/CommandHandler.cpp
index 741bdaa667..5e04c5ddb9 100644
--- a/src/main/CommandHandler.cpp
+++ b/src/main/CommandHandler.cpp
@@ -75,7 +75,7 @@ CommandHandler::CommandHandler(Application& app) : mApp(app)
 
     mServer->add404(std::bind(&CommandHandler::fileNotFound, this, _1, _2));
 
-    if (mApp.getConfig().MODE_STORES_HISTORY)
+    if (mApp.getConfig().modeStoresAnyHistory())
     {
         addRoute("dropcursor", &CommandHandler::dropcursor);
         addRoute("getcursor", &CommandHandler::getcursor);
diff --git a/src/main/CommandLine.cpp b/src/main/CommandLine.cpp
index 6df9884e83..521406d525 100644
--- a/src/main/CommandLine.cpp
+++ b/src/main/CommandLine.cpp
@@ -40,6 +40,8 @@
 namespace stellar
 {
 
+static const uint32_t MAINTENANCE_LEDGER_COUNT = 1000000;
+
 void
 writeWithTextFlow(std::ostream& os, std::string const& text)
 {
@@ -282,11 +284,17 @@ maybeSetMetadataOutputStream(Config& cfg, std::string const& stream)
     }
 }
 
-std::optional<CatchupConfiguration>
-maybeEnableInMemoryLedgerMode(Config& config, bool inMemory,
-                              uint32_t startAtLedger,
-                              std::string const& startAtHash)
+std::string
+minimalDBForInMemoryMode(Config const& cfg)
+{
+    return fmt::format("sqlite3://{}/minimal.db", cfg.BUCKET_DIR_PATH);
+}
+
+void
+maybeEnableInMemoryMode(Config& config, bool inMemory, uint32_t startAtLedger,
+                        std::string const& startAtHash, bool persistMinimalData)
 {
+    // First, ensure user parameters are valid
     if (!inMemory)
     {
         if (startAtLedger != 0)
@@ -297,12 +305,8 @@ maybeEnableInMemoryLedgerMode(Config& config, bool inMemory,
         {
             throw std::runtime_error("--start-at-hash requires --in-memory");
         }
-        return std::nullopt;
+        return;
     }
-
-    // Adjust configs for in-memory-replay mode
-    config.setInMemoryMode();
-
     if (startAtLedger != 0 && startAtHash.empty())
     {
         throw std::runtime_error("--start-at-ledger requires --start-at-hash");
@@ -311,17 +315,27 @@ maybeEnableInMemoryLedgerMode(Config& config, bool inMemory,
     {
         throw std::runtime_error("--start-at-hash requires --start-at-ledger");
     }
-    else if (startAtLedger != 0 && !startAtHash.empty())
+
+    // Adjust configs for live in-memory-replay mode
+    config.setInMemoryMode();
+
+    if (startAtLedger != 0 && !startAtHash.empty())
     {
         config.MODE_AUTO_STARTS_OVERLAY = false;
-        LedgerNumHashPair pair;
-        pair.first = startAtLedger;
-        pair.second = std::optional<Hash>(hexToBin256(startAtHash));
-        uint32_t count = 0;
-        auto mode = CatchupConfiguration::Mode::OFFLINE_COMPLETE;
-        return std::make_optional<CatchupConfiguration>(pair, count, mode);
     }
-    return std::nullopt;
+
+    // Set database to a small sqlite database used to store minimal data needed
+    // to restore the ledger state
+    if (persistMinimalData)
+    {
+        config.DATABASE = SecretValue{minimalDBForInMemoryMode(config)};
+        config.MODE_STORES_HISTORY_LEDGERHEADERS = true;
+        // Since this mode stores historical data (needed to restore
+        // ledger state in certain scenarios), set maintenance to run
+        // aggressively so that we only store a few ledgers worth of data
+        config.AUTOMATIC_MAINTENANCE_PERIOD = std::chrono::seconds(30);
+        config.AUTOMATIC_MAINTENANCE_COUNT = MAINTENANCE_LEDGER_COUNT;
+    }
 }
 
 clara::Opt
@@ -676,11 +690,12 @@ runCatchup(CommandLineArgs const& args)
                 // bulk catchup, otherwise the DB is likely to overflow with
                 // unwanted history.
                 config.AUTOMATIC_MAINTENANCE_PERIOD = std::chrono::seconds{30};
-                config.AUTOMATIC_MAINTENANCE_COUNT = 1000000;
+                config.AUTOMATIC_MAINTENANCE_COUNT = MAINTENANCE_LEDGER_COUNT;
             }
 
-            maybeEnableInMemoryLedgerMode(config, (inMemory || replayInMemory),
-                                          startAtLedger, startAtHash);
+            maybeEnableInMemoryMode(config, (inMemory || replayInMemory),
+                                    startAtLedger, startAtHash,
+                                    /* persistMinimalData */ false);
             maybeSetMetadataOutputStream(config, stream);
 
             VirtualClock clock(VirtualClock::REAL_TIME);
@@ -957,11 +972,28 @@ int
 runNewDB(CommandLineArgs const& args)
 {
     CommandLine::ConfigOption configOption;
+    bool minimalForInMemoryMode = false;
 
-    return runWithHelp(args, {configurationParser(configOption)}, [&] {
-        initializeDatabase(configOption.getConfig());
-        return 0;
-    });
+    auto minimalDBParser = [](bool& minimalForInMemoryMode) {
+        return clara::Opt{
+            minimalForInMemoryMode}["--minimal-for-in-memory-mode"](
+            "Reset the special database used only for in-memory mode (see "
+            "--in-memory flag");
+    };
+
+    return runWithHelp(args,
+                       {configurationParser(configOption),
+                        minimalDBParser(minimalForInMemoryMode)},
+                       [&] {
+                           auto cfg = configOption.getConfig();
+                           if (minimalForInMemoryMode)
+                           {
+                               cfg.DATABASE =
+                                   SecretValue{minimalDBForInMemoryMode(cfg)};
+                           }
+                           initializeDatabase(cfg);
+                           return 0;
+                       });
 }
 
 int
@@ -1068,26 +1100,79 @@ run(CommandLineArgs const& args)
          startAtLedgerParser(startAtLedger), startAtHashParser(startAtHash)},
         [&] {
             Config cfg;
-            std::optional<CatchupConfiguration> cc;
+            std::shared_ptr<VirtualClock> clock;
+            VirtualClock::Mode clockMode = VirtualClock::REAL_TIME;
+            Application::pointer app;
+
             try
             {
+                // First, craft and validate the configuration
                 cfg = configOption.getConfig();
                 cfg.DISABLE_BUCKET_GC = disableBucketGC;
                 if (simulateSleepPerOp > 0)
                 {
                     cfg.DATABASE = SecretValue{"sqlite3://:memory:"};
                     cfg.OP_APPLY_SLEEP_TIME_FOR_TESTING = simulateSleepPerOp;
-                    cfg.MODE_STORES_HISTORY = false;
+                    cfg.MODE_STORES_HISTORY_MISC = false;
                     cfg.MODE_USES_IN_MEMORY_LEDGER = false;
                     cfg.MODE_ENABLES_BUCKETLIST = false;
                     cfg.PREFETCH_BATCH_SIZE = 0;
                 }
 
-                cc = maybeEnableInMemoryLedgerMode(cfg, inMemory, startAtLedger,
-                                                   startAtHash);
+                maybeEnableInMemoryMode(cfg, inMemory, startAtLedger,
+                                        startAtHash,
+                                        /* persistMinimalData */ true);
                 maybeSetMetadataOutputStream(cfg, stream);
                 cfg.FORCE_SCP =
                     cfg.NODE_IS_VALIDATOR ? !waitForConsensus : false;
+                cfg.COMMANDS.push_back("self-check");
+
+                if (cfg.MANUAL_CLOSE)
+                {
+                    if (!cfg.NODE_IS_VALIDATOR)
+                    {
+                        LOG_ERROR(DEFAULT_LOG, "Starting stellar-core in "
+                                               "MANUAL_CLOSE mode requires "
+                                               "NODE_IS_VALIDATOR to be set");
+                        return 1;
+                    }
+                    if (cfg.RUN_STANDALONE)
+                    {
+                        clockMode = VirtualClock::VIRTUAL_TIME;
+                        if (cfg.AUTOMATIC_MAINTENANCE_COUNT != 0 ||
+                            cfg.AUTOMATIC_MAINTENANCE_PERIOD.count() != 0)
+                        {
+                            LOG_WARNING(DEFAULT_LOG,
+                                        "Using MANUAL_CLOSE and RUN_STANDALONE "
+                                        "together induces virtual time, which "
+                                        "requires automatic maintenance to be "
+                                        "disabled. AUTOMATIC_MAINTENANCE_COUNT "
+                                        "and AUTOMATIC_MAINTENANCE_PERIOD are "
+                                        "being overridden to 0.");
+                            cfg.AUTOMATIC_MAINTENANCE_COUNT = 0;
+                            cfg.AUTOMATIC_MAINTENANCE_PERIOD =
+                                std::chrono::seconds{0};
+                        }
+                    }
+                }
+
+                if (cfg.ARTIFICIALLY_ACCELERATE_TIME_FOR_TESTING)
+                {
+                    LOG_WARNING(DEFAULT_LOG, "Artificial acceleration of time "
+                                             "enabled (for testing only)");
+                }
+
+                // Second, setup the app with the final configuration.
+                // Note that when in in-memory mode, additional setup may be
+                // required (such as database reset, catchup, etc)
+                clock = std::make_shared<VirtualClock>(clockMode);
+                app = setupApp(cfg, *clock, startAtLedger, startAtHash);
+                if (!app)
+                {
+                    LOG_ERROR(DEFAULT_LOG,
+                              "Unable to setup the application to run");
+                    return 1;
+                }
             }
             catch (std::exception& e)
             {
@@ -1096,9 +1181,9 @@ run(CommandLineArgs const& args)
                 return 1;
             }
 
-            // run outside of catch block so that we properly
-            // capture crashes
-            return runWithConfig(cfg, cc);
+            // Finally, run the application outside of catch block so that we
+            // properly capture crashes
+            return runApp(app);
         });
 }
 
diff --git a/src/main/Config.cpp b/src/main/Config.cpp
index e96b5a2b29..aefcdc6d72 100644
--- a/src/main/Config.cpp
+++ b/src/main/Config.cpp
@@ -98,7 +98,8 @@ Config::Config() : NODE_SEED(SecretKey::random())
     // non configurable
     MODE_ENABLES_BUCKETLIST = true;
     MODE_USES_IN_MEMORY_LEDGER = false;
-    MODE_STORES_HISTORY = true;
+    MODE_STORES_HISTORY_MISC = true;
+    MODE_STORES_HISTORY_LEDGERHEADERS = true;
     MODE_DOES_CATCHUP = true;
     MODE_AUTO_STARTS_OVERLAY = true;
     OP_APPLY_SLEEP_TIME_FOR_TESTING = 0;
@@ -1560,7 +1561,8 @@ Config::setInMemoryMode()
 {
     MODE_USES_IN_MEMORY_LEDGER = true;
     DATABASE = SecretValue{"sqlite3://:memory:"};
-    MODE_STORES_HISTORY = false;
+    MODE_STORES_HISTORY_MISC = false;
+    MODE_STORES_HISTORY_LEDGERHEADERS = false;
     MODE_ENABLES_BUCKETLIST = true;
 }
 
@@ -1570,6 +1572,24 @@ Config::isInMemoryMode() const
     return MODE_USES_IN_MEMORY_LEDGER;
 }
 
+bool
+Config::isInMemoryModeWithoutMinimalDB() const
+{
+    return MODE_USES_IN_MEMORY_LEDGER && !MODE_STORES_HISTORY_LEDGERHEADERS;
+}
+
+bool
+Config::modeStoresAllHistory() const
+{
+    return MODE_STORES_HISTORY_LEDGERHEADERS && MODE_STORES_HISTORY_MISC;
+}
+
+bool
+Config::modeStoresAnyHistory() const
+{
+    return MODE_STORES_HISTORY_LEDGERHEADERS || MODE_STORES_HISTORY_MISC;
+}
+
 void
 Config::setNoListen()
 {
diff --git a/src/main/Config.h b/src/main/Config.h
index 62ef1f3663..07d0157d51 100644
--- a/src/main/Config.h
+++ b/src/main/Config.h
@@ -210,7 +210,10 @@ class Config : public std::enable_shared_from_this<Config>
 
     // A config parameter that stores historical data, such as transactions,
     // fees, and scp history in the database
-    bool MODE_STORES_HISTORY;
+    bool MODE_STORES_HISTORY_MISC;
+
+    // A config parameter that stores ledger headers in the database
+    bool MODE_STORES_HISTORY_LEDGERHEADERS;
 
     // A config parameter that controls whether core automatically catches up
     // when it has buffered enough input; if false an out-of-sync node will
@@ -419,6 +422,9 @@ class Config : public std::enable_shared_from_this<Config>
 
     void setInMemoryMode();
     bool isInMemoryMode() const;
+    bool isInMemoryModeWithoutMinimalDB() const;
+    bool modeStoresAllHistory() const;
+    bool modeStoresAnyHistory() const;
 
     void logBasicInfo();
     void setNoListen();
diff --git a/src/overlay/PeerManager.cpp b/src/overlay/PeerManager.cpp
index 35d16aa8e6..4654777ae7 100644
--- a/src/overlay/PeerManager.cpp
+++ b/src/overlay/PeerManager.cpp
@@ -570,6 +570,61 @@ PeerManager::dropAll(Database& db)
     db.getSession() << kSQLCreateStatement;
 }
 
+std::vector<std::pair<PeerBareAddress, PeerRecord>>
+PeerManager::loadAllPeers()
+{
+    ZoneScoped;
+    std::vector<std::pair<PeerBareAddress, PeerRecord>> result;
+    std::string sql =
+        "SELECT ip, port, nextattempt, numfailures, type FROM peers";
+
+    try
+    {
+        std::string ip;
+        int port;
+        PeerRecord record;
+
+        auto prep = mApp.getDatabase().getPreparedStatement(sql);
+        auto& st = prep.statement();
+
+        st.exchange(into(ip));
+        st.exchange(into(port));
+        st.exchange(into(record.mNextAttempt));
+        st.exchange(into(record.mNumFailures));
+        st.exchange(into(record.mType));
+
+        st.define_and_bind();
+        {
+            auto timer = mApp.getDatabase().getSelectTimer("peer");
+            st.execute(true);
+        }
+        while (st.got_data())
+        {
+            PeerBareAddress pba{ip, static_cast<unsigned short>(port)};
+            result.emplace_back(std::make_pair(pba, record));
+            st.fetch();
+        }
+    }
+    catch (soci_error& err)
+    {
+        CLOG_ERROR(Overlay, "loadPeers error: {}", err.what());
+    }
+
+    return result;
+}
+
+void
+PeerManager::storePeers(
+    std::vector<std::pair<PeerBareAddress, PeerRecord>> peers)
+{
+    soci::transaction tx(mApp.getDatabase().getSession());
+    for (auto const& peer : peers)
+    {
+        store(peer.first, peer.second, /* inDatabase */ false);
+    }
+    tx.commit();
+}
+
 const char* PeerManager::kSQLCreateStatement =
     "CREATE TABLE peers ("
     "ip            VARCHAR(15) NOT NULL,"
diff --git a/src/overlay/PeerManager.h b/src/overlay/PeerManager.h
index ea2eb9bc66..9290f40d9a 100644
--- a/src/overlay/PeerManager.h
+++ b/src/overlay/PeerManager.h
@@ -139,6 +139,16 @@ class PeerManager
     std::vector<PeerBareAddress> getPeersToSend(int size,
                                                 PeerBareAddress const& address);
 
+    /**
+     * Load all peers from the database.
+     */
+    std::vector<std::pair<PeerBareAddress, PeerRecord>> loadAllPeers();
+
+    /**
+     * Store peers in the database.
+     */
+    void storePeers(std::vector<std::pair<PeerBareAddress, PeerRecord>>);
+
   private:
     static const char* kSQLCreateStatement;