LedgerMaster.h

#pragma once
 
#include <xrpld/app/ledger/AbstractFetchPackContainer.h>
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/Ledger.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/app/misc/CanonicalTXSet.h>
 
#include <xrpl/basics/RangeSet.h>
#include <xrpl/basics/UptimeClock.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/insight/Collector.h>
#include <xrpl/protocol/Protocol.h>
#include <xrpl/protocol/RippleLedgerHash.h>
#include <xrpl/protocol/messages.h>
 
#include <mutex>
#include <optional>
 
namespace xrpl {
 
class Peer;
class Transaction;
 
// Tracks the current ledger and any ledgers in the process of closing
// Tracks ledger history
// Tracks held transactions
class LedgerMaster : public AbstractFetchPackContainer
{
public:
    explicit LedgerMaster(
        Application& app,
        Stopwatch& stopwatch,
        beast::insight::Collector::ptr const& collector,
        beast::Journal journal);
 
    virtual ~LedgerMaster() = default;
 
    LedgerIndex
    getCurrentLedgerIndex();
    LedgerIndex
    getValidLedgerIndex();
 
    bool
    isCompatible(ReadView const&, beast::Journal::Stream, char const* reason);
 
    std::recursive_mutex&
    peekMutex();
 
    // The current ledger is the ledger we believe new transactions should go in
    std::shared_ptr<ReadView const>
    getCurrentLedger();
 
    // The finalized ledger is the last closed/accepted ledger
    std::shared_ptr<Ledger const>
    getClosedLedger()
    {
        return mClosedLedger.get();
    }
 
    // The validated ledger is the last fully validated ledger.
    std::shared_ptr<Ledger const>
    getValidatedLedger();
 
    // The Rules are in the last fully validated ledger if there is one.
    Rules
    getValidatedRules();
 
    // This is the last ledger we published to clients and can lag the validated
    // ledger
    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();
 
    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);
    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();  // Returns true if path request successfully placed.
    bool
    isNewPathRequest();
    bool
    newOrderBookDB();  // Returns true if able to fulfill request.
 
    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 !mValidLedger.empty();
    }
 
    // Returns the minimum ledger sequence in SQL database, if any.
    std::optional<LedgerIndex>
    minSqlSeq();
 
    // Iff a txn exists at the specified ledger and offset then return its txnid
    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>&);
    // Try to publish ledgers, acquire missing ledgers.  Always called with
    // m_mutex locked.  The passed lock is a reminder to callers.
    void
    doAdvance(std::unique_lock<std::recursive_mutex>&);
 
    std::vector<std::shared_ptr<Ledger const>>
    findNewLedgersToPublish(std::unique_lock<std::recursive_mutex>&);
 
    void
    updatePaths();
 
    // Returns true if work started.  Always called with m_mutex locked.
    // The passed lock is a reminder to callers.
    bool
    newPFWork(char const* name, std::unique_lock<std::recursive_mutex>&);
 
    Application& app_;
    beast::Journal m_journal;
 
    std::recursive_mutex mutable m_mutex;
 
    // The ledger that most recently closed.
    LedgerHolder mClosedLedger;
 
    // The highest-sequence ledger we have fully accepted.
    LedgerHolder mValidLedger;
 
    // The last ledger we have published.
    std::shared_ptr<Ledger const> mPubLedger;
 
    // The last ledger we did pathfinding against.
    std::shared_ptr<Ledger const> mPathLedger;
 
    // The last ledger we handled fetching history
    std::shared_ptr<Ledger const> mHistLedger;
 
    // Fully validated ledger, whether or not we have the ledger resident.
    std::pair<uint256, LedgerIndex> mLastValidLedger{uint256(), 0};
 
    LedgerHistory mLedgerHistory;
 
    CanonicalTXSet mHeldTransactions{uint256()};
 
    // A set of transactions to replay during the next close
    std::unique_ptr<LedgerReplay> replayData;
 
    std::recursive_mutex mCompleteLock;
    RangeSet<std::uint32_t> mCompleteLedgers;
 
    // Publish thread is running.
    bool mAdvanceThread{false};
 
    // Publish thread has work to do.
    bool mAdvanceWork{false};
    int mFillInProgress{0};
 
    int mPathFindThread{0};  // Pathfinder jobs dispatched
    bool mPathFindNewRequest{false};
 
    std::atomic_flag mGotFetchPackThread = ATOMIC_FLAG_INIT;  // GotFetchPack jobs dispatched
 
    std::atomic<std::uint32_t> mPubLedgerClose{0};
    std::atomic<LedgerIndex> mPubLedgerSeq{0};
    std::atomic<std::uint32_t> mValidLedgerSign{0};
    std::atomic<LedgerIndex> mValidLedgerSeq{0};
    std::atomic<LedgerIndex> mBuildingLedgerSeq{0};
 
    // The server is in standalone mode
    bool const standalone_;
 
    // How many ledgers before the current ledger do we allow peers to request?
    std::uint32_t const fetch_depth_;
 
    // How much history do we want to keep
    std::uint32_t const ledger_history_;
 
    std::uint32_t const ledger_fetch_size_;
 
    TaggedCache<uint256, Blob> fetch_packs_;
 
    std::uint32_t fetch_seq_{0};
 
    // Try to keep a validator from switching from test to live network
    // without first wiping the database.
    LedgerIndex const max_ledger_difference_{1000000};
 
    // Time that the previous upgrade warning was issued.
    TimeKeeper::time_point upgradeWarningPrevTime_{};
 
private:
    struct Stats
    {
        template <class Handler>
        Stats(Handler const& handler, beast::insight::Collector::ptr const& collector)
            : hook(collector->make_hook(handler))
            , validatedLedgerAge(collector->make_gauge("LedgerMaster", "Validated_Ledger_Age"))
            , publishedLedgerAge(collector->make_gauge("LedgerMaster", "Published_Ledger_Age"))
        {
        }
 
        beast::insight::Hook hook;
        beast::insight::Gauge validatedLedgerAge;
        beast::insight::Gauge publishedLedgerAge;
    };
 
    Stats m_stats;
 
private:
    void
    collect_metrics()
    {
        std::lock_guard lock(m_mutex);
        m_stats.validatedLedgerAge.set(getValidatedLedgerAge().count());
        m_stats.publishedLedgerAge.set(getPublishedLedgerAge().count());
    }
};
 
}  // namespace xrpl