SociDB.cpp

#include <xrpl/basics/Log.h>
#include <xrpl/config/BasicConfig.h>
#include <xrpl/config/Constants.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobQueue.h>
#include <xrpl/core/ServiceRegistry.h>
 
#include <boost/filesystem/operations.hpp>
#include <boost/filesystem/path.hpp>
 
#include <soci/blob.h>
 
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
#if defined(__clang__)
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wdeprecated"
#endif
 
#include <xrpl/basics/ByteUtilities.h>
#include <xrpl/basics/contract.h>
#include <xrpl/rdb/DatabaseCon.h>
#include <xrpl/rdb/SociDB.h>
 
#include <soci/sqlite3/soci-sqlite3.h>  // IWYU pragma: keep
 
#include <memory>
 
namespace xrpl {
 
static auto gCheckpointPageCount = 1000;
 
namespace detail {
 
std::string
getSociSqliteInit(std::string const& name, std::string const& dir, std::string const& ext)
{
    if (name.empty())
    {
        Throw<std::runtime_error>(
            "Sqlite databases must specify a dir and a name. Name: " + name + " Dir: " + dir);
    }
    boost::filesystem::path file(dir);
    if (is_directory(file))
        file /= name + ext;
    return file.string();
}
 
std::string
getSociInit(BasicConfig const& config, std::string const& dbName)
{
    auto const& section = config.section(Sections::kSqdb);
    auto const backendName = get(section, Keys::kBackend, "sqlite");
 
    if (backendName != "sqlite")
        Throw<std::runtime_error>("Unsupported soci backend: " + backendName);
 
    auto const path = config.legacy(Sections::kDatabasePath);
    auto const ext = dbName == "validators" || dbName == "peerfinder" ? ".sqlite" : ".db";
    return detail::getSociSqliteInit(dbName, path, ext);
}
 
}  // namespace detail
 
DBConfig::DBConfig(std::string dbPath) : connectionString_(std::move(dbPath))
{
}
 
DBConfig::DBConfig(BasicConfig const& config, std::string const& dbName)
    : DBConfig(detail::getSociInit(config, dbName))
{
}
 
std::string
DBConfig::connectionString() const
{
    return connectionString_;
}
 
void
DBConfig::open(soci::session& s) const
{
    s.open(soci::sqlite3, connectionString());
}
 
void
open(soci::session& s, BasicConfig const& config, std::string const& dbName)
{
    DBConfig(config, dbName).open(s);
}
 
void
open(soci::session& s, std::string const& beName, std::string const& connectionString)
{
    if (beName == "sqlite")
    {
        s.open(soci::sqlite3, connectionString);
    }
    else
    {
        Throw<std::runtime_error>("Unsupported soci backend: " + beName);
    }
}
 
static sqlite_api::sqlite3*
getConnection(soci::session& s)
{
    sqlite_api::sqlite3* result = nullptr;  // NOLINT(misc-const-correctness)
    auto be = s.get_backend();
    if (auto b = dynamic_cast<soci::sqlite3_session_backend*>(be))
        result = b->conn_;
 
    if (result == nullptr)
        Throw<std::logic_error>("Didn't get a database connection.");
 
    return result;
}
 
std::uint32_t
getKBUsedAll(soci::session& s)
{
    if (getConnection(s) == nullptr)
        Throw<std::logic_error>("No connection found.");
    return static_cast<size_t>(sqlite_api::sqlite3_memory_used() / kilobytes(1));
}
 
std::uint32_t
getKBUsedDB(soci::session& s)
{
    // This function will have to be customized when other backends are added
    if (auto conn = getConnection(s))
    {
        int cur = 0, hiw = 0;
        sqlite_api::sqlite3_db_status(conn, SQLITE_DBSTATUS_CACHE_USED, &cur, &hiw, 0);
        return cur / kilobytes(1);
    }
    Throw<std::logic_error>("");
    return 0;  // Silence compiler warning.
}
 
void
convert(soci::blob& from, std::vector<std::uint8_t>& to)
{
    to.resize(from.get_len());
    if (to.empty())
        return;
    from.read(0, reinterpret_cast<char*>(&to[0]), from.get_len());
}
 
void
convert(soci::blob& from, std::string& to)
{
    std::vector<std::uint8_t> tmp;
    convert(from, tmp);
    to.assign(tmp.begin(), tmp.end());
}
 
void
convert(std::vector<std::uint8_t> const& from, soci::blob& to)
{
    if (!from.empty())
    {
        to.write(0, reinterpret_cast<char const*>(&from[0]), from.size());
    }
    else
    {
        to.trim(0);
    }
}
 
void
convert(std::string const& from, soci::blob& to)
{
    if (!from.empty())
    {
        to.write(0, from.data(), from.size());
    }
    else
    {
        to.trim(0);
    }
}
 
namespace {

 
class WALCheckpointer : public Checkpointer
{
public:
    WALCheckpointer(
        std::uintptr_t id,
        std::weak_ptr<soci::session> session,
        JobQueue& q,
        ServiceRegistry& registry)
        : id_(id)
        , session_(std::move(session))
        , jobQueue_(q)
        , j_(registry.getJournal("WALCheckpointer"))
    {
        if (auto [conn, keepAlive] = getConnection(); conn)
        {
            (void)keepAlive;
            sqlite_api::sqlite3_wal_hook(conn, &sqliteWALHook, reinterpret_cast<void*>(id_));
        }
    }
 
    std::pair<sqlite_api::sqlite3*, std::shared_ptr<soci::session>>
    getConnection() const
    {
        if (auto p = session_.lock())
        {
            return {xrpl::getConnection(*p), p};
        }
        return {nullptr, std::shared_ptr<soci::session>{}};
    }
 
    std::uintptr_t
    id() const override
    {
        return id_;
    }
 
    ~WALCheckpointer() override = default;
 
    void
    schedule() override
    {
        {
            std::scoped_lock const lock(mutex_);
            if (running_)
                return;
            running_ = true;
        }
 
        // If the Job is not added to the JobQueue then we're not running_.
        if (!jobQueue_.addJob(
                JtWal,
                "WAL",
                // If the owning DatabaseCon is destroyed, no need to checkpoint
                // or keep the checkpointer alive so use a weak_ptr to this.
                // There is a separate check in `checkpoint` for a valid
                // connection in the rare case when the DatabaseCon is destroyed
                // after locking this weak_ptr
                [wp = std::weak_ptr<Checkpointer>{shared_from_this()}]() {
                    if (auto self = wp.lock())
                        self->checkpoint();
                }))
        {
            std::scoped_lock const lock(mutex_);
            running_ = false;
        }
    }
 
    void
    checkpoint() override
    {
        auto [conn, keepAlive] = getConnection();
        (void)keepAlive;
        if (conn == nullptr)
            return;
 
        int log = 0, ckpt = 0;
        int const ret = sqlite_api::sqlite3_wal_checkpoint_v2(
            conn, nullptr, SQLITE_CHECKPOINT_PASSIVE, &log, &ckpt);
 
        auto fname = sqlite_api::sqlite3_db_filename(conn, "main");
        if (ret != SQLITE_OK)
        {
            auto jm = (ret == SQLITE_LOCKED) ? j_.trace() : j_.warn();
            JLOG(jm) << "WAL(" << fname << "): error " << ret;
        }
        else
        {
            JLOG(j_.trace()) << "WAL(" << fname << "): frames=" << log << ", written=" << ckpt;
        }
 
        std::scoped_lock const lock(mutex_);
        running_ = false;
    }
 
protected:
    std::uintptr_t const id_;
    // session is owned by the DatabaseCon parent that holds the checkpointer.
    // It is possible (though rare) for the DatabaseCon class to be destroyed
    // before the checkpointer.
    std::weak_ptr<soci::session> session_;
    std::mutex mutex_;
    JobQueue& jobQueue_;
 
    bool running_ = false;
    beast::Journal const j_;
 
    static int
    sqliteWALHook(void* cpId, sqlite_api::sqlite3* conn, char const* dbName, int walSize)
    {
        if (walSize >= gCheckpointPageCount)
        {
            if (auto checkpointer = checkpointerFromId(reinterpret_cast<std::uintptr_t>(cpId)))
            {
                checkpointer->schedule();
            }
            else
            {
                sqlite_api::sqlite3_wal_hook(conn, nullptr, nullptr);
            }
        }
        return SQLITE_OK;
    }
};
 
}  // namespace
 
std::shared_ptr<Checkpointer>
makeCheckpointer(
    std::uintptr_t id,
    std::weak_ptr<soci::session> session,
    JobQueue& queue,
    ServiceRegistry& registry)
{
    return std::make_shared<WALCheckpointer>(id, std::move(session), queue, registry);
}
 
}  // namespace xrpl
 
#if defined(__clang__)
#pragma clang diagnostic pop
#endif