rippled/src/xrpld/app/ledger/detail/LedgerDeltaAcquire.cpp

#include <xrpld/app/ledger/detail/LedgerDeltaAcquire.h>

#include <xrpld/app/ledger/BuildLedger.h>
#include <xrpld/app/ledger/InboundLedger.h>
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/LedgerMaster.h>
#include <xrpld/app/ledger/LedgerReplay.h>
#include <xrpld/app/ledger/LedgerReplayer.h>
#include <xrpld/app/ledger/detail/TimeoutCounter.h>
#include <xrpld/app/main/Application.h>
#include <xrpld/overlay/Peer.h>
#include <xrpld/overlay/PeerSet.h>
#include <xrpld/telemetry/MetricMacros.h>
#include <xrpld/telemetry/MetricNames.h>

#include <xrpl/basics/Log.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/contract.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/ledger/ApplyView.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/Rules.h>

#include <xrpl.pb.h>

#include <cstddef>
#include <cstdint>
#include <map>
#include <memory>
#include <optional>
#include <stdexcept>
#include <utility>
#include <vector>

namespace xrpl {

LedgerDeltaAcquire::LedgerDeltaAcquire(
    Application& app,
    InboundLedgers& inboundLedgers,
    uint256 const& ledgerHash,
    std::uint32_t ledgerSeq,
    std::unique_ptr<PeerSet> peerSet)
    : TimeoutCounter(
          app,
          ledgerHash,
          ledger_replay_parameters::kSubTaskTimeout,
          {.jobType = JtReplayTask,
           .jobName = "LedReplDelta",
           .jobLimit = ledger_replay_parameters::kMaxQueuedTasks},
          app.getJournal("LedgerReplayDelta"))
    , inboundLedgers_(inboundLedgers)
    , ledgerSeq_(ledgerSeq)
    , peerSet_(std::move(peerSet))
{
    JLOG(journal_.trace()) << "Create " << hash_ << " Seq " << ledgerSeq;
}

LedgerDeltaAcquire::~LedgerDeltaAcquire()
{
    JLOG(journal_.trace()) << "Destroy " << hash_;
}

void
LedgerDeltaAcquire::init(int numPeers)
{
    ScopedLockType sl(mtx_);
    if (!isDone())
    {
        trigger(numPeers, sl);
        setTimer(sl);
    }
}

void
LedgerDeltaAcquire::trigger(std::size_t limit, ScopedLockType& sl)
{
    fullLedger_ = app_.getLedgerMaster().getLedgerByHash(hash_);
    if (fullLedger_)
    {
        complete_ = true;
        JLOG(journal_.trace()) << "existing ledger " << hash_;
        notify(sl);
        return;
    }

    if (!fallBack_)
    {
        peerSet_->addPeers(
            limit,
            [this](auto peer) {
                return peer->supportsFeature(ProtocolFeature::LedgerReplay) &&
                    peer->hasLedger(hash_, ledgerSeq_);
            },
            [this](auto peer) {
                if (peer->supportsFeature(ProtocolFeature::LedgerReplay))
                {
                    JLOG(journal_.trace()) << "Add a peer " << peer->id() << " for " << hash_;
                    protocol::TMReplayDeltaRequest request;
                    request.set_ledgerhash(hash_.data(), hash_.size());
                    peerSet_->sendRequest(request, peer);
                }
                else
                {
                    if (++noFeaturePeerCount_ >= ledger_replay_parameters::kMaxNoFeaturePeerCount)
                    {
                        JLOG(journal_.debug()) << "Fall back for " << hash_;
                        timerInterval_ = ledger_replay_parameters::kSubTaskFallbackTimeout;

                        // Same fallback as the skip-list stage, for the delta
                        // stage: too few replay-capable peers, so the whole
                        // ledger is acquired instead of just its delta. The
                        // `stage` label separates the two, because they fail
                        // for different reasons and are fixed differently.
                        // Emitted once, on the transition into fallback.
                        XRPL_METRIC_COUNTER_INC_LABELED(
                            app_,
                            telemetry::metric::ledgerReplayFallbackTotal,
                            "Replay sub-acquires that fell back to a full ledger acquire",
                            {{telemetry::label::stage,
                              std::string(telemetry::lval::replay_fallback::delta)}});

                        fallBack_ = true;
                    }
                }
            });
    }

    if (fallBack_)
        inboundLedgers_.acquire(hash_, ledgerSeq_, InboundLedger::Reason::GENERIC);
}

void
LedgerDeltaAcquire::onTimer(bool progress, ScopedLockType& sl)
{
    JLOG(journal_.trace()) << "timeouts_=" << timeouts_ << " for " << hash_;
    if (timeouts_ > ledger_replay_parameters::kSubTaskMaxTimeouts)
    {
        failed_ = true;
        JLOG(journal_.debug()) << "too many timeouts " << hash_;
        notify(sl);
    }
    else
    {
        trigger(1, sl);
    }
}

std::weak_ptr<TimeoutCounter>
LedgerDeltaAcquire::pmDowncast()
{
    return shared_from_this();
}

void
LedgerDeltaAcquire::processData(
    LedgerHeader const& info,
    std::map<std::uint32_t, std::shared_ptr<STTx const>>&& orderedTxns)
{
    ScopedLockType sl(mtx_);
    JLOG(journal_.trace()) << "got data for " << hash_;
    if (isDone())
        return;

    if (info.seq == ledgerSeq_)
    {
        // create a temporary ledger for building a LedgerReplay object later
        Rules const rules{app_.config().features};
        replayTemp_ = std::make_shared<Ledger>(info, rules, app_.getNodeFamily());
        if (replayTemp_)
        {
            complete_ = true;
            orderedTxns_ = std::move(orderedTxns);
            JLOG(journal_.debug()) << "ready to replay " << hash_;
            notify(sl);
            return;
        }
    }

    failed_ = true;
    JLOG(journal_.error()) << "failed to create a (info only) ledger from verified data " << hash_;
    notify(sl);
}

void
LedgerDeltaAcquire::addDataCallback(InboundLedger::Reason reason, OnDeltaDataCB&& cb)
{
    ScopedLockType sl(mtx_);
    dataReadyCallbacks_.emplace_back(std::move(cb));
    if (!reasons_.contains(reason))
    {
        reasons_.emplace(reason);
        if (fullLedger_)
            onLedgerBuilt(sl, reason);
    }

    if (isDone())
    {
        JLOG(journal_.debug()) << "task added to a finished LedgerDeltaAcquire " << hash_;
        notify(sl);
    }
}

std::shared_ptr<Ledger const>
LedgerDeltaAcquire::tryBuild(std::shared_ptr<Ledger const> const& parent)
{
    ScopedLockType sl(mtx_);

    if (fullLedger_)
        return fullLedger_;

    if (failed_ || !complete_ || !replayTemp_)
        return {};

    XRPL_ASSERT(
        parent->seq() + 1 == replayTemp_->seq(),
        "xrpl::LedgerDeltaAcquire::tryBuild : parent sequence match");
    XRPL_ASSERT(
        parent->header().hash == replayTemp_->header().parentHash,
        "xrpl::LedgerDeltaAcquire::tryBuild : parent hash match");
    // build ledger
    LedgerReplay const replayData(parent, replayTemp_, std::move(orderedTxns_));
    fullLedger_ = buildLedger(replayData, TapNone, app_, journal_);
    if (fullLedger_ && fullLedger_->header().hash == hash_)
    {
        JLOG(journal_.info()) << "Built " << hash_;
        onLedgerBuilt(sl);
        return fullLedger_;
    }

    failed_ = true;
    complete_ = false;
    JLOG(journal_.error()) << "tryBuild failed " << hash_ << " with parent "
                           << parent->header().hash;
    Throw<std::runtime_error>("Cannot replay ledger");
}

void
LedgerDeltaAcquire::onLedgerBuilt(ScopedLockType& sl, std::optional<InboundLedger::Reason> reason)
{
    JLOG(journal_.debug()) << "onLedgerBuilt " << hash_ << (reason ? " for a new reason" : "");

    std::vector<InboundLedger::Reason> reasons(reasons_.begin(), reasons_.end());
    bool firstTime = true;
    if (reason)  // small chance
    {
        reasons.clear();
        reasons.push_back(*reason);
        firstTime = false;
    }
    app_.getJobQueue().addJob(
        JtReplayTask, "OnLedBuilt", [=, ledger = this->fullLedger_, &app = this->app_]() {
            for (auto reason : reasons)
            {
                switch (reason)
                {
                    case InboundLedger::Reason::GENERIC:
                        app.getLedgerMaster().storeLedger(ledger);
                        break;
                    default:
                        // TODO for other use cases
                        break;
                }
            }

            if (firstTime)
                app.getLedgerMaster().tryAdvance();
        });
}

void
LedgerDeltaAcquire::notify(ScopedLockType& sl)
{
    XRPL_ASSERT(isDone(), "xrpl::LedgerDeltaAcquire::notify : is done");
    std::vector<OnDeltaDataCB> toCall;
    std::swap(toCall, dataReadyCallbacks_);
    auto const good = !failed_;
    sl.unlock();

    for (auto& cb : toCall)
    {
        cb(good, hash_);
    }

    sl.lock();
}

}  // namespace xrpl