SHAMapStoreImp.h

#pragma once
 
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/app/misc/SHAMapStore.h>
 
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/config/BasicConfig.h>
#include <xrpl/ledger/Ledger.h>
#include <xrpl/nodestore/Backend.h>
#include <xrpl/nodestore/Database.h>
#include <xrpl/nodestore/DatabaseRotating.h>
#include <xrpl/nodestore/Scheduler.h>
#include <xrpl/protocol/Protocol.h>
#include <xrpl/rdb/DatabaseCon.h>
#include <xrpl/server/State.h>
#include <xrpl/shamap/FullBelowCache.h>
#include <xrpl/shamap/SHAMapTreeNode.h>
#include <xrpl/shamap/TreeNodeCache.h>
 
#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <thread>
 
namespace xrpl {
 
class NetworkOPs;
 
class SHAMapStoreImp : public SHAMapStore
{
private:
    class SavedStateDB
    {
    public:
        soci::session sqlDb;
        std::mutex mutex;
        beast::Journal const journal;
 
        // Just instantiate without any logic in case online delete is not
        // configured
        explicit SavedStateDB() : journal{beast::Journal::getNullSink()}
        {
        }
 
        // opens database and, if necessary, creates & initializes its tables.
        void
        init(BasicConfig const& config, std::string const& dbName);
        // get/set the ledger index that we can delete up to and including
        LedgerIndex
        getCanDelete();
        LedgerIndex
        setCanDelete(LedgerIndex canDelete);
        SavedState
        getState();
        void
        setState(SavedState const& state);
        void
        setLastRotated(LedgerIndex seq);
    };
 
    Application& app_;
 
    // name of state database
    std::string const dbName_ = "state";
    // prefix of on-disk nodestore backend instances
    std::string const dbPrefix_ = "rippledb";  // cspell: disable-line
    // check health/stop status as records are copied
    std::uint64_t const checkHealthInterval_ = 1000;
    // minimum # of ledgers to maintain for health of network
    static std::uint32_t const kMinimumDeletionInterval = 256;
    // minimum # of ledgers required for standalone mode.
    static std::uint32_t const kMinimumDeletionIntervalSa = 8;
    // minimum ledger to maintain online.
    std::atomic<LedgerIndex> minimumOnline_;
 
    node_store::Scheduler& scheduler_;
    beast::Journal const journal_;
    node_store::DatabaseRotating* dbRotating_ = nullptr;
    SavedStateDB stateDb_;
    std::thread thread_;
    bool stop_ = false;
    bool healthy_ = true;
    // Used to prevent ledger gaps from forming during online deletion. Keeps
    // track of the last validated ledger that was processed without gaps. There
    // are no guarantees about gaps while online delete is not running. For
    // that, use advisory_delete and check for gaps externally.
    LedgerIndex lastGoodValidatedLedger_ = 0;
    // Used to prevent the circuit breaker from tripping too quickly.
    LedgerIndex lastSuccessfulHealthCheck_ = 0;
    mutable std::condition_variable cond_;
    mutable std::condition_variable rendezvous_;
    mutable std::mutex mutex_;
    std::shared_ptr<Ledger const> newLedger_;
    std::atomic<bool> working_;
    std::atomic<LedgerIndex> canDelete_;
    int fdRequired_ = 0;
 
    std::uint32_t deleteInterval_ = 0;
    bool advisoryDelete_ = false;
    std::uint32_t deleteBatch_ = 100;
    std::chrono::milliseconds backOff_{100};
    std::chrono::seconds ageThreshold_{60};
    std::chrono::seconds recoveryWaitTime_{2};
    std::uint32_t maxWaitingLedgers_ = deleteBatch_;
 
    // these do not exist upon SHAMapStore creation, but do exist
    // as of run() or before
    NetworkOPs* netOPs_ = nullptr;
    LedgerMaster* ledgerMaster_ = nullptr;
    FullBelowCache* fullBelowCache_ = nullptr;
    TreeNodeCache* treeNodeCache_ = nullptr;
 
    static constexpr auto kNodeStoreName = "NodeStore";
 
public:
    SHAMapStoreImp(Application& app, node_store::Scheduler& scheduler, beast::Journal journal);
 
    std::uint32_t
    clampFetchDepth(std::uint32_t fetchDepth) const override
    {
        return (deleteInterval_ != 0u) ? std::min(fetchDepth, deleteInterval_) : fetchDepth;
    }
 
    std::unique_ptr<node_store::Database>
    makeNodeStore(int readThreads) override;
 
    LedgerIndex
    setCanDelete(LedgerIndex seq) override
    {
        if (advisoryDelete_)
            canDelete_ = seq;
        return stateDb_.setCanDelete(seq);
    }
 
    bool
    advisoryDelete() const override
    {
        return advisoryDelete_;
    }
 
    // All ledgers prior to this one are eligible
    // for deletion in the next rotation
    LedgerIndex
    getLastRotated() override
    {
        return stateDb_.getState().lastRotated;
    }
 
    // All ledgers before and including this are unprotected
    // and online delete may delete them if appropriate
    LedgerIndex
    getCanDelete() override
    {
        return canDelete_;
    }
 
    void
    onLedgerClosed(std::shared_ptr<Ledger const> const& ledger) override;
 
    [[nodiscard]]
    bool
    rendezvous(std::optional<std::chrono::milliseconds> const& timeout = {}) const override;
    int
    fdRequired() const override;
 
    std::optional<LedgerIndex>
    minimumOnline() const override;
 
private:
    // callback for visitNodes
    bool
    copyNode(std::uint64_t& nodeCount, SHAMapTreeNode const& node);
    void
    run();
    void
    dbPaths();
 
    std::unique_ptr<node_store::Backend>
    makeBackendRotating(std::string path = std::string());
 
    template <class CacheInstance>
    bool
    freshenCache(CacheInstance& cache)
    {
        std::uint64_t check = 0;
 
        for (auto const& key : cache.getKeys())
        {
            dbRotating_->fetchNodeObject(key, 0, node_store::FetchType::Synchronous, true);
            if (!(++check % checkHealthInterval_) && healthWait() != HealthResult::KeepGoing)
                return true;
        }
 
        return false;
    }

    void
    clearSql(
        LedgerIndex lastRotated,
        std::string const& tableName,
        std::function<std::optional<LedgerIndex>()> const& getMinSeq,
        std::function<void(LedgerIndex)> const& deleteBeforeSeq);
    void
    clearCaches(LedgerIndex validatedSeq);
    void
    freshenCaches();
    void
    clearPrior(LedgerIndex lastRotated);

    enum class HealthResult { Stopping, Expired, KeepGoing };
    [[nodiscard]] HealthResult
    healthWait();
 
public:
    void
    start() override
    {
        if (deleteInterval_ != 0u)
            thread_ = std::thread(&SHAMapStoreImp::run, this);
    }
 
    void
    stop() override;
};
 
}  // namespace xrpl