LedgerPublisher.hpp

#pragma once
 
#include "data/BackendInterface.hpp"
#include "data/DBHelpers.hpp"
#include "etl/LedgerPublisherInterface.hpp"
#include "etl/SystemState.hpp"
#include "etl/impl/Loading.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Assert.hpp"
#include "util/Mutex.hpp"
#include "util/async/AnyExecutionContext.hpp"
#include "util/async/AnyStrand.hpp"
#include "util/log/Logger.hpp"
#include "util/prometheus/Counter.hpp"
#include "util/prometheus/Prometheus.hpp"
 
#include <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/strand.hpp>
#include <fmt/format.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/protocol/Fees.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/SField.h>
#include <xrpl/protocol/STObject.h>
#include <xrpl/protocol/Serializer.h>
 
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <string>
#include <thread>
#include <utility>
#include <vector>
 
namespace etl::impl {

class LedgerPublisher : public LedgerPublisherInterface {
    util::Logger log_{"ETL"};
 
    util::async::AnyStrand publishStrand_;
 
    std::atomic_bool stop_{false};
 
    std::shared_ptr<BackendInterface> backend_;
    std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
    std::reference_wrapper<SystemState const> state_;  // shared state for ETL
 
    util::Mutex<std::chrono::time_point<ripple::NetClock>, std::shared_mutex> lastCloseTime_;
 
    std::reference_wrapper<util::prometheus::CounterInt> lastPublishSeconds_ =
        PrometheusService::counterInt(
            "etl_last_publish_seconds",
            {},
            "Seconds since epoch of the last published ledger"
        );
 
    util::Mutex<std::optional<uint32_t>, std::shared_mutex> lastPublishedSequence_;
 
public:
    LedgerPublisher(
        util::async::AnyExecutionContext ctx,
        std::shared_ptr<BackendInterface> backend,
        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
        SystemState const& state
    )
        : publishStrand_{ctx.makeStrand()}
        , backend_{std::move(backend)}
        , subscriptions_{std::move(subscriptions)}
        , state_{std::cref(state)}
    {
    }

    bool
    publish(
        uint32_t ledgerSequence,
        std::optional<uint32_t> maxAttempts,
        std::chrono::steady_clock::duration attemptsDelay = std::chrono::seconds{1}
    ) override
    {
        LOG(log_.info()) << "Attempting to publish ledger = " << ledgerSequence;
        size_t numAttempts = 0;
        while (not stop_) {
            auto range = backend_->hardFetchLedgerRangeNoThrow();
 
            if (!range || range->maxSequence < ledgerSequence) {
                ++numAttempts;
                LOG(log_.debug()) << "Trying to publish. Could not find ledger with sequence = "
                                  << ledgerSequence;
 
                // We try maxAttempts times to publish the ledger, waiting one second in between
                // each attempt.
                if (maxAttempts && numAttempts >= maxAttempts) {
                    LOG(log_.debug())
                        << "Failed to publish ledger after " << numAttempts << " attempts.";
                    return false;
                }
                std::this_thread::sleep_for(attemptsDelay);
                continue;
            }
 
            auto lgr = data::synchronousAndRetryOnTimeout([&](auto yield) {
                return backend_->fetchLedgerBySequence(ledgerSequence, yield);
            });
 
            ASSERT(
                lgr.has_value(),
                "Ledger must exist in database. Ledger sequence = {}",
                ledgerSequence
            );
            if (!lgr)
                return false;
            publish(*lgr);
 
            return true;
        }
        return false;
    }

    void
    publish(ripple::LedgerHeader const& lgrInfo)
    {
        publishStrand_.submit([this, lgrInfo = lgrInfo] {
            LOG(log_.info()) << "Publishing ledger " << std::to_string(lgrInfo.seq);
 
            setLastClose(lgrInfo.closeTime);
            auto age = lastCloseAgeSeconds();
 
            // if the ledger closed over MAX_LEDGER_AGE_SECONDS ago, assume we are still catching up
            // and don't publish
            static constexpr std::uint32_t kMaxLedgerAgeSeconds = 600;
            if (age < kMaxLedgerAgeSeconds) {
                std::optional<ripple::Fees> fees =
                    data::synchronousAndRetryOnTimeout([&](auto yield) {
                        return backend_->fetchFees(lgrInfo.seq, yield);
                    });
                ASSERT(fees.has_value(), "Fees must exist for ledger {}", lgrInfo.seq);
 
                auto transactions = data::synchronousAndRetryOnTimeout([&](auto yield) {
                    return backend_->fetchAllTransactionsInLedger(lgrInfo.seq, yield);
                });
 
                auto const ledgerRange = backend_->fetchLedgerRange();
                ASSERT(ledgerRange.has_value(), "Ledger range must exist");
 
                auto const range =
                    fmt::format("{}-{}", ledgerRange->minSequence, ledgerRange->maxSequence);
                subscriptions_->pubLedger(lgrInfo, *fees, range, transactions.size());
 
                // order with transaction index
                std::ranges::sort(transactions, [](auto const& t1, auto const& t2) {
                    ripple::SerialIter iter1{t1.metadata.data(), t1.metadata.size()};
                    ripple::STObject const object1(iter1, ripple::sfMetadata);
                    ripple::SerialIter iter2{t2.metadata.data(), t2.metadata.size()};
                    ripple::STObject const object2(iter2, ripple::sfMetadata);
                    return object1.getFieldU32(ripple::sfTransactionIndex) <
                        object2.getFieldU32(ripple::sfTransactionIndex);
                });
 
                for (auto const& txAndMeta : transactions)
                    subscriptions_->pubTransaction(txAndMeta, lgrInfo);
 
                subscriptions_->pubBookChanges(lgrInfo, transactions);
 
                setLastPublishTime();
                LOG(log_.info()) << "Published ledger " << lgrInfo.seq;
            } else {
                LOG(log_.info()) << "Skipping publishing ledger " << lgrInfo.seq;
            }
        });
 
        // we track latest publish-requested seq, not necessarily already published
        setLastPublishedSequence(lgrInfo.seq);
    }

    std::uint32_t
    lastPublishAgeSeconds() const override
    {
        return std::chrono::duration_cast<std::chrono::seconds>(
                   std::chrono::system_clock::now() - getLastPublish()
        )
            .count();
    }

    std::chrono::time_point<std::chrono::system_clock>
    getLastPublish() const override
    {
        return std::chrono::time_point<std::chrono::system_clock>{
            std::chrono::seconds{lastPublishSeconds_.get().value()}
        };
    }

    std::uint32_t
    lastCloseAgeSeconds() const override
    {
        auto closeTime = lastCloseTime_.lock()->time_since_epoch().count();
        auto now = std::chrono::duration_cast<std::chrono::seconds>(
                       std::chrono::system_clock::now().time_since_epoch()
        )
                       .count();
        if (now < (kRippleEpochStart + closeTime))
            return 0;
        return now - (kRippleEpochStart + closeTime);
    }

    std::optional<uint32_t>
    getLastPublishedSequence() const
    {
        return *lastPublishedSequence_.lock();
    }

    void
    stop()
    {
        stop_ = true;
    }
 
private:
    void
    setLastClose(std::chrono::time_point<ripple::NetClock> lastCloseTime)
    {
        auto closeTime = lastCloseTime_.lock<std::scoped_lock>();
        *closeTime = lastCloseTime;
    }
 
    void
    setLastPublishTime()
    {
        using namespace std::chrono;
        auto const nowSeconds =
            duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
        lastPublishSeconds_.get().set(nowSeconds);
    }
 
    void
    setLastPublishedSequence(std::optional<uint32_t> lastPublishedSequence)
    {
        auto lastPublishSeq = lastPublishedSequence_.lock();
        *lastPublishSeq = lastPublishedSequence;
    }
};
 
}  // namespace etl::impl