LedgerMaster.h

#pragma once
 
#include <xrpld/app/ledger/AbstractFetchPackContainer.h>
#include <xrpld/app/ledger/InboundLedger.h>
#include <xrpld/app/ledger/LedgerHistory.h>
#include <xrpld/app/ledger/LedgerHolder.h>
#include <xrpld/app/ledger/LedgerReplay.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/core/TimeKeeper.h>
 
#include <xrpl/basics/Blob.h>
#include <xrpl/basics/RangeSet.h>
#include <xrpl/basics/UptimeClock.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/insight/Collector.h>
#include <xrpl/beast/insight/Gauge.h>
#include <xrpl/beast/insight/Hook.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/json/json_value.h>
#include <xrpl/ledger/CanonicalTXSet.h>
#include <xrpl/ledger/Ledger.h>
#include <xrpl/ledger/ReadView.h>
#include <xrpl/protocol/Protocol.h>
#include <xrpl/protocol/RippleLedgerHash.h>
#include <xrpl/protocol/Rules.h>
 
#include <xrpl.pb.h>
 
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <utility>
#include <vector>
 
namespace xrpl {
 
class Peer;
class Transaction;

class LedgerMaster : public AbstractFetchPackContainer
{
public:
    explicit LedgerMaster(
        Application& app,
        Stopwatch& stopwatch,
        beast::insight::Collector::Ptr const& collector,
        beast::Journal journal);

    ~LedgerMaster() override = default;

    LedgerIndex
    getCurrentLedgerIndex();

    LedgerIndex
    getValidLedgerIndex();

    bool
    isCompatible(ReadView const&, beast::Journal::Stream, char const* reason);

    std::recursive_mutex&
    peekMutex();

    std::shared_ptr<ReadView const>
    getCurrentLedger();

    std::shared_ptr<Ledger const>
    getClosedLedger()
    {
        return closedLedger_.get();
    }

    std::shared_ptr<Ledger const>
    getValidatedLedger();

    Rules
    getValidatedRules();

    std::shared_ptr<ReadView const>
    getPublishedLedger();

    std::chrono::seconds
    getPublishedLedgerAge();

    std::chrono::seconds
    getValidatedLedgerAge();

    bool
    isCaughtUp(std::string& reason);

    std::uint32_t
    getEarliestFetch();

    bool
    storeLedger(std::shared_ptr<Ledger const> ledger);

    void
    setFullLedger(std::shared_ptr<Ledger const> const& ledger, bool isSynchronous, bool isCurrent);

    bool
    canBeCurrent(std::shared_ptr<Ledger const> const& ledger);

    void
    switchLCL(std::shared_ptr<Ledger const> const& lastClosed);

    void
    failedSave(std::uint32_t seq, UInt256 const& hash);

    std::string
    getCompleteLedgers() const;

    std::size_t
    missingFromCompleteLedgerRange(LedgerIndex first, LedgerIndex last) const;

    void
    applyHeldTransactions();

    std::shared_ptr<STTx const>
    popAcctTransaction(std::shared_ptr<STTx const> const& tx);

    UInt256
    getHashBySeq(std::uint32_t index);

    std::optional<LedgerHash>
    walkHashBySeq(std::uint32_t index, InboundLedger::Reason reason);

    std::optional<LedgerHash>
    walkHashBySeq(
        std::uint32_t index,
        std::shared_ptr<ReadView const> const& referenceLedger,
        InboundLedger::Reason reason);

    std::shared_ptr<Ledger const>
    getLedgerBySeq(std::uint32_t index);

    std::shared_ptr<Ledger const>
    getLedgerByHash(UInt256 const& hash);

    void
    setLedgerRangePresent(std::uint32_t minV, std::uint32_t maxV);

    std::optional<NetClock::time_point>
    getCloseTimeBySeq(LedgerIndex ledgerIndex);

    std::optional<NetClock::time_point>
    getCloseTimeByHash(LedgerHash const& ledgerHash, LedgerIndex ledgerIndex);

    void
    addHeldTransaction(std::shared_ptr<Transaction> const& trans);

    void
    fixMismatch(ReadView const& ledger);

    bool
    haveLedger(std::uint32_t seq) const;

    void
    clearLedger(std::uint32_t seq);

    bool
    isValidated(ReadView const& ledger);

    bool
    getValidatedRange(std::uint32_t& minVal, std::uint32_t& maxVal);

    bool
    getFullValidatedRange(std::uint32_t& minVal, std::uint32_t& maxVal);

    void
    sweep();

    float
    getCacheHitRate();

    void
    checkAccept(std::shared_ptr<Ledger const> const& ledger);

    void
    checkAccept(UInt256 const& hash, std::uint32_t seq);

    void
    consensusBuilt(
        std::shared_ptr<Ledger const> const& ledger,
        UInt256 const& consensusHash,
        json::Value consensus);

    void
    setBuildingLedger(LedgerIndex index);

    void
    tryAdvance();

    bool
    newPathRequest();

    bool
    isNewPathRequest();

    bool
    newOrderBookDB();

    bool
    fixIndex(LedgerIndex ledgerIndex, LedgerHash const& ledgerHash);

    void
    clearPriorLedgers(LedgerIndex seq);

    void
    clearLedgerCachePrior(LedgerIndex seq);
 
    // ledger replay

    void
    takeReplay(std::unique_ptr<LedgerReplay> replay);

    std::unique_ptr<LedgerReplay>
    releaseReplay();
 
    // Fetch Packs

    void
    gotFetchPack(bool progress, std::uint32_t seq);

    void
    addFetchPack(UInt256 const& hash, std::shared_ptr<Blob> data);

    std::optional<Blob>
    getFetchPack(UInt256 const& hash) override;

    void
    makeFetchPack(
        std::weak_ptr<Peer> const& wPeer,
        std::shared_ptr<protocol::TMGetObjectByHash> const& request,
        UInt256 haveLedgerHash,
        UptimeClock::time_point uptime);

    std::size_t
    getFetchPackCacheSize() const;

    bool
    haveValidated()
    {
        return !validLedger_.empty();
    }

    std::optional<LedgerIndex>
    minSqlSeq();

    std::optional<UInt256>
    txnIdFromIndex(uint32_t ledgerSeq, uint32_t txnIndex);
 
private:
    void
    setValidLedger(std::shared_ptr<Ledger const> const& l);

    void
    setPubLedger(std::shared_ptr<Ledger const> const& l);

    void
    tryFill(std::shared_ptr<Ledger const> ledger);

    void
    getFetchPack(LedgerIndex missing, InboundLedger::Reason reason);

    std::optional<LedgerHash>
    getLedgerHashForHistory(LedgerIndex index, InboundLedger::Reason reason);

    std::size_t
    getNeededValidations();

    void
    fetchForHistory(
        std::uint32_t missing,
        bool& progress,
        InboundLedger::Reason reason,
        std::unique_lock<std::recursive_mutex>&);

    void
    doAdvance(std::unique_lock<std::recursive_mutex>&);

    std::vector<std::shared_ptr<Ledger const>>
    findNewLedgersToPublish(std::unique_lock<std::recursive_mutex>&);

    void
    updatePaths();

    bool
    newPFWork(char const* name, std::unique_lock<std::recursive_mutex>&);

    Application& app_;

    beast::Journal journal_;

    std::recursive_mutex mutable mutex_;

    LedgerHolder closedLedger_;

    LedgerHolder validLedger_;

    std::shared_ptr<Ledger const> pubLedger_;

    std::shared_ptr<Ledger const> pathLedger_;

    std::shared_ptr<Ledger const> histLedger_;

    std::pair<UInt256, LedgerIndex> lastValidLedger_{UInt256(), 0};

    LedgerHistory ledgerHistory_;

    CanonicalTXSet heldTransactions_{UInt256()};

    std::unique_ptr<LedgerReplay> replayData_;

    std::recursive_mutex mutable completeLock_;

    RangeSet<std::uint32_t> completeLedgers_;

    bool advanceThread_{false};

    bool advanceWork_{false};

    int fillInProgress_{0};

    int pathFindThread_{0};

    bool pathFindNewRequest_{false};

    std::atomic_flag gotFetchPackThread_ = ATOMIC_FLAG_INIT;

    std::atomic<std::uint32_t> pubLedgerClose_{0};

    std::atomic<LedgerIndex> pubLedgerSeq_{0};

    std::atomic<std::uint32_t> validLedgerSign_{0};

    std::atomic<LedgerIndex> validLedgerSeq_{0};

    std::atomic<LedgerIndex> buildingLedgerSeq_{0};

    bool const standalone_;

    std::uint32_t const fetchDepth_;

    std::uint32_t const ledgerHistorySize_;

    std::uint32_t const ledgerFetchSize_;

    TaggedCache<UInt256, Blob> fetchPacks_;

    std::uint32_t fetchSeq_{0};

    LedgerIndex const maxLedgerDifference_{1000000};

    TimeKeeper::time_point upgradeWarningPrevTime_;
 
private:
    struct Stats
    {
        template <class Handler>
        Stats(Handler const& handler, beast::insight::Collector::Ptr const& collector)
            : hook(collector->makeHook(handler))
            , validatedLedgerAge(collector->makeGauge("LedgerMaster", "Validated_Ledger_Age"))
            , publishedLedgerAge(collector->makeGauge("LedgerMaster", "Published_Ledger_Age"))
        {
        }

        beast::insight::Hook hook;

        beast::insight::Gauge validatedLedgerAge;

        beast::insight::Gauge publishedLedgerAge;
    };

    Stats stats_;
 
private:
    void
    collectMetrics()
    {
        std::scoped_lock const lock(mutex_);
        stats_.validatedLedgerAge.set(getValidatedLedgerAge().count());
        stats_.publishedLedgerAge.set(getPublishedLedgerAge().count());
    }
};
 
}  // namespace xrpl