// Copyright 2014 Stellar Development Foundation and contributors. Licensed

// under the Apache License, Version 2.0. See the COPYING file at the root

// of this distribution or at http://www.apache.org/licenses/LICENSE-2.0

#include "ApplicationImpl.h"

#include "util/Fs.h"

#include "work/ConditionalWork.h"

#include "work/WorkWithCallback.h"

#include "xdr/Stellar-ledger-entries.h"

#include <limits>

// ASIO is somewhat particular about when it gets included -- it wants to be the

// first to include <windows.h> -- so we try to include it before everything

// else.

#include "util/asio.h"

#include "bucket/BucketManager.h"

#include "catchup/ApplyBucketsWork.h"

#include "crypto/Hex.h"

#include "crypto/SHA.h"

#include "crypto/SecretKey.h"

#include "database/Database.h"

#include "herder/Herder.h"

#include "herder/HerderPersistence.h"

#include "history/HistoryArchiveManager.h"

#include "history/HistoryManager.h"

#include "invariant/AccountSubEntriesCountIsValid.h"

#include "invariant/ArchivedStateConsistency.h"

#include "invariant/BucketListIsConsistentWithDatabase.h"

#include "invariant/BucketListStateConsistency.h"

#include "invariant/ConservationOfLumens.h"

#include "invariant/ConstantProductInvariant.h"

#include "invariant/EventsAreConsistentWithEntryDiffs.h"

#include "invariant/InvariantManager.h"

#include "invariant/LedgerEntryIsValid.h"

#include "invariant/LiabilitiesMatchOffers.h"

#include "invariant/SponsorshipCountIsValid.h"

#include "ledger/LedgerHeaderUtils.h"

#include "ledger/LedgerManager.h"

#include "ledger/LedgerManagerImpl.h"

#include "ledger/LedgerTxn.h"

#include "ledger/P23HotArchiveBug.h"

#include "main/AppConnector.h"

#include "main/ApplicationUtils.h"

#include "main/CommandHandler.h"

#include "main/StellarCoreVersion.h"

#include "medida/counter.h"

#include "medida/meter.h"

#include "medida/reporting/console_reporter.h"

#include "medida/timer.h"

#include "overlay/BanManager.h"

#include "overlay/OverlayManager.h"

#include "overlay/OverlayManagerImpl.h"

#include "process/ProcessManager.h"

#include "transactions/SignatureChecker.h"

#include "util/GlobalChecks.h"

#include "util/JitterInjection.h"

#include "util/LogSlowExecution.h"

#include "util/Logging.h"

#include "util/MetricsRegistry.h"

#include "util/ProtocolVersion.h"

#include "util/StatusManager.h"

#include "util/Thread.h"

#include "util/TmpDir.h"

#include "work/BasicWork.h"

#include "work/WorkScheduler.h"

#ifdef BUILD_TESTS

#include "ledger/test/InMemoryLedgerTxn.h"

#include "ledger/test/InMemoryLedgerTxnRoot.h"

#include "simulation/LoadGenerator.h"

#include "test/TestAccount.h"

#endif

#include <Tracy.hpp>

#include <fmt/format.h>

#include <optional>

#include <set>

#include <string>

static int const SHUTDOWN_DELAY_SECONDS = 1;

namespace stellar

{

ApplicationImpl::ApplicationImpl(VirtualClock& clock, Config const& cfg)

: mVirtualClock(clock)

, mConfig(cfg)

// Allocate one worker to eviction when background eviction enabled

, mWorkerIOContext(mConfig.WORKER_THREADS - 1)

, mEvictionIOContext(std::make_unique<asio::io_context>(1))

, mWork(std::make_unique<asio::io_context::work>(mWorkerIOContext))

, mEvictionWork(

mEvictionIOContext

? std::make_unique<asio::io_context::work>(*mEvictionIOContext)

: nullptr)

, mOverlayIOContext(mConfig.BACKGROUND_OVERLAY_PROCESSING

? std::make_unique<asio::io_context>(1)

: nullptr)

, mOverlayWork(mOverlayIOContext ? std::make_unique<asio::io_context::work>(

*mOverlayIOContext)

: nullptr)

, mLedgerCloseIOContext(mConfig.parallelLedgerClose()

? std::make_unique<asio::io_context>(1)

: nullptr)

, mLedgerCloseWork(

mLedgerCloseIOContext

? std::make_unique<asio::io_context::work>(*mLedgerCloseIOContext)

: nullptr)

, mWorkerThreads()

, mEvictionThread()

, mStopSignals(clock.getIOContext(), SIGINT)

, mStarted(false)

, mStopping(false)

#ifdef BUILD_TESTS

, mRunInOverlayOnlyMode(false)

#endif

, mStoppingTimer(*this)

, mSelfCheckTimer(*this)

, mMetrics(std::make_unique<MetricsRegistry>(cfg.HISTOGRAM_WINDOW_SIZE))

, mPostOnMainThreadDelay(

mMetrics->NewTimer({"app", "post-on-main-thread", "delay"}))

, mPostOnBackgroundThreadDelay(

mMetrics->NewTimer({"app", "post-on-background-thread", "delay"}))

, mPostOnOverlayThreadDelay(

mMetrics->NewTimer({"app", "post-on-overlay-thread", "delay"}))

, mPostOnLedgerCloseThreadDelay(

mMetrics->NewTimer({"app", "post-on-ledger-close-thread", "delay"}))

, mStartedOn(clock.system_now())

{

#ifdef SIGQUIT

mStopSignals.add(SIGQUIT);

#endif

#ifdef SIGTERM

mStopSignals.add(SIGTERM);

#endif

std::srand(static_cast<uint32>(clock.now().time_since_epoch().count()));

mNetworkID = sha256(mConfig.NETWORK_PASSPHRASE);

TracyAppInfo(STELLAR_CORE_VERSION.c_str(), STELLAR_CORE_VERSION.size());

TracyAppInfo(mConfig.NETWORK_PASSPHRASE.c_str(),

mConfig.NETWORK_PASSPHRASE.size());

std::string nodeStr("Node: ");

nodeStr += mConfig.NODE_SEED.getStrKeyPublic();

TracyAppInfo(nodeStr.c_str(), nodeStr.size());

std::string homeStr("HomeDomain: ");

homeStr += mConfig.NODE_HOME_DOMAIN;

TracyAppInfo(homeStr.c_str(), homeStr.size());

mStopSignals.async_wait([this](asio::error_code const& ec, int sig) {

if (!ec)

{

LOG_INFO(DEFAULT_LOG, "got signal {}, shutting down", sig);

#ifdef BUILD_TESTS

if (mConfig.TEST_CASES_ENABLED)

{

exit(1);

}

#endif

this->gracefulStop();

}

});

auto t = mConfig.WORKER_THREADS;

LOG_DEBUG(DEFAULT_LOG, "Application constructing (worker threads: {})", t);

releaseAssert(mConfig.WORKER_THREADS > 0);

releaseAssert(mEvictionIOContext);

mThreadTypes[std::this_thread::get_id()] = ThreadType::MAIN;

// Allocate one thread for Eviction scan

mEvictionThread = std::make_unique<std::thread>([this]() {

runCurrentThreadWithMediumPriority();

mEvictionIOContext->run();

});

mThreadTypes[mEvictionThread->get_id()] = ThreadType::EVICTION;

--t;

while (t--)

{

auto thread = std::make_unique<std::thread>([this]() {

runCurrentThreadWithLowPriority();

mWorkerIOContext.run();

});

mThreadTypes[thread->get_id()] = ThreadType::WORKER;

mWorkerThreads.emplace_back(std::move(thread));

}

if (mConfig.BACKGROUND_OVERLAY_PROCESSING)

{

// Keep priority unchanged as overlay processes time-sensitive tasks

mOverlayThread = std::make_unique<std::thread>(

[this]() { mOverlayIOContext->run(); });

mThreadTypes[mOverlayThread->get_id()] = ThreadType::OVERLAY;

}

if (mConfig.parallelLedgerClose())

{

mLedgerCloseThread = std::make_unique<std::thread>(

[this]() { mLedgerCloseIOContext->run(); });

mThreadTypes[mLedgerCloseThread->get_id()] = ThreadType::APPLY;

}

}

static void

maybeRebuildLedger(Application& app, bool applyBuckets)

{

auto& ps = app.getPersistentState();

if (ps.shouldRebuildForOfferTable())

{

soci::transaction tx(app.getDatabase().getRawSession());

LOG_INFO(DEFAULT_LOG, "Dropping offers");

app.getLedgerTxnRoot().dropOffers();

tx.commit();

// No transaction is needed. ApplyBucketsWork breaks the apply into many

// small chunks, each of which has its own transaction. If it fails at

// some point in the middle, then rebuildledger will not be cleared so

// this will run again on next start up.

if (applyBuckets)

{

LOG_INFO(DEFAULT_LOG,

"Rebuilding ledger tables by applying buckets");

if (!applyBucketsForLCL(app))

{

throw std::runtime_error("Could not rebuild ledger tables");

}

}

LOG_INFO(DEFAULT_LOG, "Successfully rebuilt ledger tables");

}

ps.clearRebuildForOfferTable();

}

void

ApplicationImpl::initialize(bool createNewDB, bool forceRebuild)

{

// 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.

mAppConnector = std::make_unique<AppConnector>(*this);

mBucketManager = BucketManager::create(getAppConnector());

bool initNewDB =

createNewDB || mConfig.DATABASE.value == "sqlite3://:memory:";

if (initNewDB)

{

mBucketManager->maybeDropAndCreateNew();

}

mDatabase = createDatabase();

mPersistentState = std::make_unique<PersistentState>(*this);

mOverlayManager = createOverlayManager();

mLedgerManager = createLedgerManager();

mHerder = createHerder();

mHerderPersistence = HerderPersistence::create(*this);

mLedgerApplyManager = LedgerApplyManager::create(*this);

mHistoryArchiveManager = std::make_unique<HistoryArchiveManager>(*this);

mHistoryManager = HistoryManager::create(*this);

mInvariantManager = createInvariantManager();

mWorkScheduler = WorkScheduler::create(*this);

mBanManager = BanManager::create(*this);

mStatusManager = std::make_unique<StatusManager>();

// Load Protocol 23 corruption data if path to CSV is specified

if (!mConfig.PATH_TO_PROTOCOL_23_CORRUPTION_FILE.empty())

{

mProtocol23CorruptionDataVerifier = std::make_unique<

p23_hot_archive_bug::Protocol23CorruptionDataVerifier>();

if (!mProtocol23CorruptionDataVerifier->loadFromFile(

mConfig.PATH_TO_PROTOCOL_23_CORRUPTION_FILE))

{

throw std::invalid_argument(

"Failed to load Protocol 23 corruption file: " +

mConfig.PATH_TO_PROTOCOL_23_CORRUPTION_FILE);

}

}

if (mConfig.BACKFILL_STELLAR_ASSET_EVENTS)

{

mProtocol23CorruptionEventReconciler = std::make_unique<

p23_hot_archive_bug::Protocol23CorruptionEventReconciler>(

getNetworkID());

}

if (mConfig.ENTRY_CACHE_SIZE < 20000)

{

LOG_WARNING(DEFAULT_LOG,

"ENTRY_CACHE_SIZE({}) is below the recommended minimum "

"of 20000",

mConfig.ENTRY_CACHE_SIZE);

}

mLedgerTxnRoot = getLedgerManager().createLedgerTxnRoot(

*this, mConfig.ENTRY_CACHE_SIZE, mConfig.PREFETCH_BATCH_SIZE

#ifdef BEST_OFFER_DEBUGGING

,

mConfig.BEST_OFFER_DEBUGGING_ENABLED

#endif

);

#ifdef BUILD_TESTS

if (getConfig().MODE_USES_IN_MEMORY_LEDGER)

{

resetLedgerState();

}

#endif

BucketListIsConsistentWithDatabase::registerInvariant(*this);

AccountSubEntriesCountIsValid::registerInvariant(*this);

ConservationOfLumens::registerInvariant(*this);

LedgerEntryIsValid::registerInvariant(*this);

LiabilitiesMatchOffers::registerInvariant(*this);

SponsorshipCountIsValid::registerInvariant(*this);

ConstantProductInvariant::registerInvariant(*this);

EventsAreConsistentWithEntryDiffs::registerInvariant(*this);

ArchivedStateConsistency::registerInvariant(*this);

BucketListStateConsistency::registerInvariant(*this);

enableInvariantsFromConfig();

if (initNewDB)

{

newDB();

}

else

{

upgradeToCurrentSchemaAndMaybeRebuildLedger(true, forceRebuild);

}

// Subtle: process manager should come to existence _after_ BucketManager

// initialization and newDB run, as it relies on tmp dir created in the

// constructor

mProcessManager = ProcessManager::create(*this);

// After everything is initialized, start accepting HTTP commands

mCommandHandler = std::make_unique<CommandHandler>(*this);

LOG_DEBUG(DEFAULT_LOG, "Application constructed");

}

void

ApplicationImpl::resetLedgerState()

{

#ifdef BUILD_TESTS

mRootAccount.reset();

if (getConfig().MODE_USES_IN_MEMORY_LEDGER)

{

mNeverCommittingLedgerTxn.reset();

mInMemoryLedgerTxnRoot = std::make_unique<InMemoryLedgerTxnRoot>(

#ifdef BEST_OFFER_DEBUGGING

mConfig.BEST_OFFER_DEBUGGING_ENABLED

#endif

);

mNeverCommittingLedgerTxn = std::make_unique<InMemoryLedgerTxn>(

*mInMemoryLedgerTxnRoot, getDatabase(), *mLedgerTxnRoot);

}

else

#endif

{

auto& lsRoot = getLedgerTxnRoot();

lsRoot.deleteOffersModifiedOnOrAfterLedger(0);

}

}

void

ApplicationImpl::newDB()

{

mDatabase->initialize();

upgradeToCurrentSchemaAndMaybeRebuildLedger(false, true);

mLedgerManager->startNewLedger();

}

void

ApplicationImpl::upgradeToCurrentSchemaAndMaybeRebuildLedger(bool applyBuckets,

bool forceRebuild)

{

if (forceRebuild)

{

auto& ps = getPersistentState();

ps.setRebuildForOfferTable();

}

mDatabase->upgradeToCurrentSchema();

maybeRebuildLedger(*this, applyBuckets);

}

void

ApplicationImpl::reportCfgMetrics()

{

if (!mMetrics)

{

return;

}

std::set<std::string> metricsToReport;

std::set<std::string> allMetrics;

for (auto& kv : mMetrics->GetAllMetrics())

{

allMetrics.insert(kv.first.ToString());

}

bool reportAvailableMetrics = false;

for (auto const& name : mConfig.REPORT_METRICS)

{

if (allMetrics.find(name) == allMetrics.end())

{

LOG_INFO(DEFAULT_LOG, "");

LOG_WARNING(DEFAULT_LOG, "Metric not found: {}", name);

reportAvailableMetrics = true;

}

metricsToReport.insert(name);

}

if (reportAvailableMetrics)

{

LOG_INFO(DEFAULT_LOG,

"Update REPORT_METRICS value in configuration file to "

"only include available metrics:");

for (auto const& n : allMetrics)

{

LOG_INFO(DEFAULT_LOG, " {}", n);

}

LOG_INFO(DEFAULT_LOG, "");

}

std::ostringstream oss;

medida::reporting::ConsoleReporter reporter(*mMetrics, oss);

for (auto& kv : mMetrics->GetAllMetrics())

{

auto name = kv.first;

auto metric = kv.second;

auto nstr = name.ToString();

if (metricsToReport.find(nstr) != metricsToReport.end())

{

LOG_INFO(DEFAULT_LOG, "");

LOG_INFO(DEFAULT_LOG, "metric '{}':", nstr);

metric->Process(reporter);

std::istringstream iss(oss.str());

char buf[128];

while (iss.getline(buf, 128))

{

LOG_INFO(DEFAULT_LOG, "{}", std::string(buf));

}

oss.str("");

LOG_INFO(DEFAULT_LOG, "");

}

}

}

Json::Value

ApplicationImpl::getJsonInfo(bool verbose)

{

auto root = Json::Value{};

auto& lm = getLedgerManager();

auto& info = root["info"];

info["build"] = STELLAR_CORE_VERSION;

info["protocol_version"] = getConfig().LEDGER_PROTOCOL_VERSION;

info["state"] = getStateHuman();

info["startedOn"] = VirtualClock::systemPointToISOString(mStartedOn);

auto const& lcl = lm.getLastClosedLedgerHeader();

info["ledger"]["num"] = (int)lcl.header.ledgerSeq;

info["ledger"]["hash"] = binToHex(lcl.hash);

info["ledger"]["closeTime"] = (Json::UInt64)lcl.header.scpValue.closeTime;

info["ledger"]["version"] = lcl.header.ledgerVersion;

info["ledger"]["baseFee"] = lcl.header.baseFee;

info["ledger"]["baseReserve"] = lcl.header.baseReserve;

info["ledger"]["maxTxSetSize"] = lcl.header.maxTxSetSize;

if (lm.hasLastClosedSorobanNetworkConfig())

{

info["ledger"]["maxSorobanTxSetSize"] =

static_cast<Json::Int64>(lm.maxLedgerResources(/* isSoroban */ true)

.getVal(Resource::Type::OPERATIONS));

}

auto currentHeaderFlags = LedgerHeaderUtils::getFlags(lcl.header);

if (currentHeaderFlags != 0)

{

info["ledger"]["flags"] = currentHeaderFlags;

}

info["ledger"]["age"] = (int)lm.secondsSinceLastLedgerClose();

if (verbose)

{

auto has = lm.getLastClosedLedgerHAS();

auto& levels = info["ledger"]["bucketlist"];

for (auto const& l : has.currentBuckets)

{

Json::Value levelInfo;

levelInfo["curr"] = l.curr;

levelInfo["snap"] = l.snap;

levels.append(levelInfo);

}

}

info["peers"]["pending_count"] = getOverlayManager().getPendingPeersCount();

info["peers"]["authenticated_count"] =

getOverlayManager().getAuthenticatedPeersCount();

info["network"] = getConfig().NETWORK_PASSPHRASE;

auto& statusMessages = getStatusManager();

auto counter = 0;

for (auto statusMessage : statusMessages)

{

info["status"][counter++] = statusMessage.second;

}

auto& herder = getHerder();

auto& quorumInfo = info["quorum"];

// Try to get quorum set info for the previous ledger closed by the

// network.

auto ledgerSeq = lcl.header.ledgerSeq;

if (herder.getState() != Herder::HERDER_BOOTING_STATE)

{

ledgerSeq = herder.trackingConsensusLedgerIndex();

}

if (ledgerSeq > 1)

{

quorumInfo = herder.getJsonQuorumInfo(

getConfig().NODE_SEED.getPublicKey(), true, false, ledgerSeq - 1);

}

// If the quorum set info for the previous ledger is missing, use the

// current ledger

auto qset = quorumInfo.get("qset", "");

if (quorumInfo.empty() || qset.empty())

{

quorumInfo = herder.getJsonQuorumInfo(

getConfig().NODE_SEED.getPublicKey(), true, false, ledgerSeq);

}

auto invariantFailures = getInvariantManager().getJsonInfo();

if (!invariantFailures.empty())

{

info["invariant_failures"] = invariantFailures;

}

return root;

}

void

ApplicationImpl::reportInfo(bool verbose)

{

mLedgerManager->partiallyLoadLastKnownLedgerForUtils();

LOG_INFO(DEFAULT_LOG, "Reporting application info");

std::cout << getJsonInfo(verbose).toStyledString() << std::endl;

}

std::shared_ptr<BasicWork>

ApplicationImpl::scheduleSelfCheck(bool waitUntilNextCheckpoint)

{

// Ensure the timer is re-triggered no matter what.

if (mConfig.AUTOMATIC_SELF_CHECK_PERIOD.count() != 0)

{

mSelfCheckTimer.expires_from_now(mConfig.AUTOMATIC_SELF_CHECK_PERIOD);

mSelfCheckTimer.async_wait(

[this, waitUntilNextCheckpoint]() {

scheduleSelfCheck(waitUntilNextCheckpoint);

},

VirtualTimer::onFailureNoop);

}

// We store any self-check that gets scheduled so that a second call to

// schedule a self-check while one is running returns the running one.

if (auto existing = mRunningSelfCheck.lock())

{

return existing;

}

auto& ws = getWorkScheduler();

auto& lm = getLedgerManager();

auto& ham = getHistoryArchiveManager();

LedgerHeaderHistoryEntry lhhe = lm.getLastClosedLedgerHeader();

std::vector<std::shared_ptr<BasicWork>> seq;

seq.emplace_back(ham.getHistoryArchiveReportWork());

auto checkLedger = ham.getCheckLedgerHeaderWork(lhhe);

if (waitUntilNextCheckpoint)

{

// Delay until a second full checkpoint-period after the next checkpoint

// publication. The captured lhhe should usually be published by then.

auto targetLedger =

HistoryManager::firstLedgerAfterCheckpointContaining(

lhhe.header.ledgerSeq, getConfig());

targetLedger = HistoryManager::firstLedgerAfterCheckpointContaining(

targetLedger, getConfig());

auto cond = [targetLedger](Application& app) -> bool {

auto& lm = app.getLedgerManager();

return lm.getLastClosedLedgerNum() > targetLedger;

};

seq.emplace_back(std::make_shared<ConditionalWork>(

*this, "wait-for-lcl", cond, checkLedger,

std::chrono::seconds(10)));

}

else

{

seq.emplace_back(checkLedger);

}

// Stash a weak ptr to the shared ptr we're returning, for subsequent calls.

std::shared_ptr<BasicWork> ptr =

ws.scheduleWork<WorkSequence>("self-check", seq, BasicWork::RETRY_NEVER,

/*stopOnFirstFailure=*/false);

mRunningSelfCheck = ptr;

return ptr;

}

Hash const&

ApplicationImpl::getNetworkID() const

{

return mNetworkID;

}

ApplicationImpl::~ApplicationImpl()

{

LOG_INFO(DEFAULT_LOG, "Application destructing");

mStopping = true;

try

{

idempotentShutdown(false);

}

catch (std::exception const& e)

{

LOG_ERROR(DEFAULT_LOG, "Exception occurred while shutting down: {}",

e.what());

}

reportCfgMetrics();

LOG_INFO(DEFAULT_LOG, "Application destroyed");

}

uint64_t

ApplicationImpl::timeNow()

{

return VirtualClock::to_time_t(getClock().system_now());

}

void

ApplicationImpl::validateAndLogConfig()

{

if (mConfig.FORCE_SCP && !mConfig.NODE_IS_VALIDATOR)

{

throw std::invalid_argument(

"FORCE_SCP is set but NODE_IS_VALIDATOR not set");

}

auto const isNetworkedValidator =

mConfig.NODE_IS_VALIDATOR && !mConfig.RUN_STANDALONE;

if (mConfig.ENABLE_SOROBAN_DIAGNOSTIC_EVENTS && isNetworkedValidator)

{

throw std::invalid_argument("ENABLE_SOROBAN_DIAGNOSTIC_EVENTS is set, "

"NODE_IS_VALIDATOR is set, and "

"RUN_STANDALONE is not set");

}

if (mConfig.METADATA_OUTPUT_STREAM != "" && isNetworkedValidator)

{

throw std::invalid_argument(

"METADATA_OUTPUT_STREAM is set, NODE_IS_VALIDATOR is set, and "

"RUN_STANDALONE is not set");

}

if (!mDatabase->isSqlite())

{

CLOG_WARNING(Database,

"Non-sqlite3 database detected. Support for other sql "

"backends is deprecated and will be removed in a future "

"release. Please use sqlite3 for non-ledger state data.");

}

auto pageSizeExp = mConfig.BUCKETLIST_DB_INDEX_PAGE_SIZE_EXPONENT;

// If the page size is less than 256 bytes, it is essentially

// indexing individual keys, so page size should be set to 0

// instead.

static auto const pageSizeMinExponent = 8;

// Any exponent above 31 will cause overflow

static auto const pageSizeMaxExponent = 31;

if (pageSizeExp != 0)

{

if (pageSizeExp < pageSizeMinExponent)

{

throw std::invalid_argument(

"BUCKETLIST_DB_INDEX_PAGE_SIZE_EXPONENT "

"must be at least 8 or set to 0 for individual entry "

"indexing");

}

if (pageSizeExp > pageSizeMaxExponent)

{

throw std::invalid_argument(

"BUCKETLIST_DB_INDEX_PAGE_SIZE_EXPONENT "

"must be less than 32");

}

}

CLOG_INFO(Bucket,

"BucketListDB enabled: pageSizeExponent: {} indexCutOff: "

"{}MB, persist indexes: {}",

pageSizeExp, mConfig.BUCKETLIST_DB_INDEX_CUTOFF,

mConfig.BUCKETLIST_DB_PERSIST_INDEX);

if (mConfig.HTTP_QUERY_PORT != 0)

{

if (isNetworkedValidator)

{

throw std::invalid_argument("HTTP_QUERY_PORT is non-zero, "

"NODE_IS_VALIDATOR is set, and "

"RUN_STANDALONE is not set");

}

if (mConfig.HTTP_QUERY_PORT == mConfig.HTTP_PORT)

{

throw std::invalid_argument(

"HTTP_QUERY_PORT must be different from HTTP_PORT");

}

if (mConfig.QUERY_THREAD_POOL_SIZE == 0)

{

throw std::invalid_argument(

"HTTP_QUERY_PORT requires QUERY_THREAD_POOL_SIZE > 0");

}

}

if (getHistoryArchiveManager().publishEnabled())

{

if (!mConfig.MODE_STORES_HISTORY_MISC)

{

throw std::invalid_argument(

"Core is not configured to store history, but "

"some history archives are writable");

}

}

if (mConfig.QUORUM_SET.threshold == 0)

{

throw std::invalid_argument("Quorum not configured");

}

mConfig.logBasicInfo();

}

void

ApplicationImpl::startServices()

{

mInvariantManager->start(*mLedgerManager);

// restores Herder's state before starting overlay

mHerder->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, "* ");

mHerder->bootstrap();

}

if (mConfig.AUTOMATIC_SELF_CHECK_PERIOD.count() != 0)

{

scheduleSelfCheck(true);

}

if (mConfig.TESTING_UPGRADE_DATETIME.time_since_epoch().count() != 0)

{

mHerder->setUpgrades(mConfig);

}

}

void

ApplicationImpl::start()

{

if (mStarted)

{

CLOG_INFO(Ledger, "Skipping application start up");

return;

}

CLOG_INFO(Ledger, "Starting up application");

mStarted = true;

mLedgerManager->loadLastKnownLedger();

// LCL is now loaded; unblock HTTP endpoints that were gated during boot.

mCommandHandler->setReady();

// Check if we're already on protocol V_24 or later and enable Rust Dalek

auto const& lcl = mLedgerManager->getLastClosedLedgerHeader();

if (protocolVersionStartsFrom(lcl.header.ledgerVersion,

ProtocolVersion::V_25))

{

PubKeyUtils::enableRustDalekVerify();

}

startServices();

}

void

ApplicationImpl::idempotentShutdown(bool forgetBuckets)

{

// Graceful shutdown sequence:

// Perform a graceful shutdown by first signaling all managers to stop

// scheduling new work, then stopping the main-thread io_service to prevent

// further dispatch. Once no new tasks can be queued, release the io_context

// work guard so worker threads exit naturally after completing their

// current tasks. Drain any remaining operations while the application state

// is still valid, join all worker threads, and finally allow destructors to

// run once no asynchronous activity remains.

// Shutdown state-modifying ledger close thread first, while all subsystems

// are live and valid. Note that `joinAllThreads` will also attempt to

// shutdown mLedgerCloseThread for completeness, but since this method is

// idempotent, the extra call is harmless.

shutdownThread(mLedgerCloseThread, mLedgerCloseWork, "ledger close");

if (mCommandHandler)

{

mCommandHandler->shutdown();

}

if (mOverlayManager)

{

mOverlayManager->shutdown();

}

mSelfCheckTimer.cancel();

shutdownWorkScheduler();

if (mProcessManager)

{

mProcessManager->shutdown();

}

if (mBucketManager)

{

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)

{

mHerder->shutdown();

}

shutdownMainIOContext();

joinAllThreads();

}

void

ApplicationImpl::gracefulStop()

{

releaseAssert(threadIsMain());

if (mStopping)

{

return;

}

mStopping = true;

idempotentShutdown(true);

mStoppingTimer.expires_from_now(

std::chrono::seconds(SHUTDOWN_DELAY_SECONDS));

mStoppingTimer.async_wait(

std::bind(&ApplicationImpl::shutdownMainIOContext, this),

VirtualTimer::onFailureNoop);

}

void

ApplicationImpl::shutdownMainIOContext()

{

mVirtualClock.shutdown();

}

void

ApplicationImpl::shutdownWorkScheduler()

{

if (mWorkScheduler)

{

mWorkScheduler->shutdown();

}

}

bool

ApplicationImpl::shutdownThread(

std::unique_ptr<std::thread>& threadPtr,

std::unique_ptr<asio::io_context::work>& workPtr,

std::string const& threadName)

{

if (threadPtr)

{

// We never strictly stop the worker IO service, just release the

// work-lock

// that keeps the worker threads alive. This gives them the chance to

// finish any work that the main thread queued.

if (workPtr)

{

workPtr.reset();

}

LOG_INFO(DEFAULT_LOG, "Joining {} thread", threadName);

threadPtr->join();

threadPtr.reset();

return true;

}

return false;

}

void

ApplicationImpl::joinAllThreads()

{

uint32_t joined = 0;

joined +=

shutdownThread(mLedgerCloseThread, mLedgerCloseWork, "ledger close");

for (auto& w : mWorkerThreads)

{

joined += shutdownThread(w, mWork, "worker");

}

mWorkerThreads.clear();

joined += shutdownThread(mOverlayThread, mOverlayWork, "overlay");

joined += shutdownThread(mEvictionThread, mEvictionWork, "eviction");

if (joined)

{

LOG_INFO(DEFAULT_LOG, "Joined all {} threads", joined);

}

}

std::string

ApplicationImpl::manualClose(std::optional<uint32_t> const& manualLedgerSeq,

std::optional<TimePoint> const& manualCloseTime)

{

releaseAssert(threadIsMain());

// Manual close only makes sense for validating nodes

if (!mConfig.NODE_IS_VALIDATOR)

{

throw std::invalid_argument(

"Issuing a manual ledger close requires NODE_IS_VALIDATOR to be "

"set to true.");

}

if (mConfig.MANUAL_CLOSE)

{

auto const targetLedgerSeq =

targetManualCloseLedgerSeqNum(manualLedgerSeq);

if (mConfig.RUN_STANDALONE)

{

setManualCloseVirtualTime(manualCloseTime);

advanceToLedgerBeforeManualCloseTarget(targetLedgerSeq);

}

else

{

// These may only be specified if RUN_STANDALONE is configured.

releaseAssert(!manualCloseTime);

releaseAssert(!manualLedgerSeq);

}

mHerder->triggerNextLedger(targetLedgerSeq, true);

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)

{

throw std::runtime_error(fmt::format(