LedgerPublisher.hpp

//------------------------------------------------------------------------------
/*
    This file is part of clio: https://github.com/XRPLF/clio
    Copyright (c) 2025, the clio developers.
 
    Permission to use, copy, modify, and distribute this software for any
    purpose with or without fee is hereby granted, provided that the above
    copyright notice and this permission notice appear in all copies.
 
    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
 
#pragma once
 
#include "data/BackendInterface.hpp"
#include "data/DBHelpers.hpp"
#include "data/Types.hpp"
#include "etl/SystemState.hpp"
#include "etlng/LedgerPublisherInterface.hpp"
#include "etlng/impl/Loading.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Assert.hpp"
#include "util/Mutex.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 <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 <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 etlng::impl {
 
class LedgerPublisher : public etlng::LedgerPublisherInterface {
    util::Logger log_{"ETL"};
 
    boost::asio::strand<boost::asio::io_context::executor_type> publishStrand_;
 
    std::shared_ptr<BackendInterface> backend_;
    std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
    std::reference_wrapper<etl::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(
        boost::asio::io_context& ioc,  // TODO: replace with AsyncContext shared with ETLServiceNg
        std::shared_ptr<BackendInterface> backend,
        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
        etl::SystemState const& state
    )
        : publishStrand_{boost::asio::make_strand(ioc)}
        , 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 state_.get().isStopping) {
            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);
            publish(*lgr);
 
            return true;
        }
        return false;
    }
 
    void
    publish(ripple::LedgerHeader const& lgrInfo)
    {
        boost::asio::post(publishStrand_, [this, lgrInfo = lgrInfo]() {
            LOG(log_.info()) << "Publishing ledger " << std::to_string(lgrInfo.seq);
 
            // TODO: This should probably not be part of publisher in the future
            if (not state_.get().isWriting)
                backend_->updateRange(lgrInfo.seq);  // This can't be unit tested atm.
 
            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 kMAX_LEDGER_AGE_SECONDS = 600;
            if (age < kMAX_LEDGER_AGE_SECONDS) {
                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 < (kRIPPLE_EPOCH_START + closeTime))
            return 0;
        return now - (kRIPPLE_EPOCH_START + closeTime);
    }
 
    std::optional<uint32_t>
    getLastPublishedSequence() const
    {
        return *lastPublishedSequence_.lock();
    }
 
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 etlng::impl