HerderPersistenceImpl.cpp

// Copyright 2014 Stellar Development Foundation and contributors. Licensed
// under the Apache License, Version 2.0. See the COPYING file at the root
// of this distribution or at http://www.apache.org/licenses/LICENSE-2.0

#include "herder/HerderPersistenceImpl.h"
#include "crypto/Hex.h"
#include "database/Database.h"
#include "database/DatabaseUtils.h"
#include "herder/Herder.h"
#include "main/Application.h"
#include "scp/Slot.h"
#include "util/Decoder.h"
#include "util/XDRStream.h"
#include <Tracy.hpp>

#include <optional>
#include <soci.h>
#include <xdrpp/marshal.h>

namespace stellar
{

std::unique_ptr<HerderPersistence>
HerderPersistence::create(Application& app)
{
    return std::make_unique<HerderPersistenceImpl>(app);
}

HerderPersistenceImpl::HerderPersistenceImpl(Application& app) : mApp(app)
{
}

HerderPersistenceImpl::~HerderPersistenceImpl()
{
}

void
HerderPersistenceImpl::saveSCPHistory(uint32_t seq,
                                      std::vector<SCPEnvelope> const& envs,
                                      QuorumTracker::QuorumMap const& qmap)
{
    ZoneScoped;
    releaseAssert(threadIsMain());
    if (envs.empty())
    {
        return;
    }

    auto usedQSets = UnorderedMap<Hash, SCPQuorumSetPtr>{};
    auto& db = mApp.getDatabase();
    auto& sess = db.getMiscSession();

    soci::transaction txscope(sess.session());

    {
        auto prepClean = db.getPreparedStatement(
            "DELETE FROM scphistory WHERE ledgerseq =:l", sess);

        prepClean.exchange(soci::use(seq));
        prepClean.define_and_bind();
        {
            ZoneNamedN(deleteSCPHistoryZone, "delete scphistory", true);
            prepClean.execute(true);
        }
    }

    // Prepare multi-row insert into scphistory
    std::vector<std::string> nodeIDs;
    std::vector<uint32_t> seqs;
    std::vector<std::string> envelopes;
    for (auto const& e : envs)
    {
        auto const& qHash =
            Slot::getCompanionQuorumSetHashFromStatement(e.statement);
        usedQSets.insert(
            std::make_pair(qHash, mApp.getHerder().getQSet(qHash)));

        std::string nodeIDStrKey = KeyUtils::toStrKey(e.statement.nodeID);

        auto envelopeBytes(xdr::xdr_to_opaque(e));

        std::string envelopeEncoded;
        envelopeEncoded = decoder::encode_b64(envelopeBytes);

        nodeIDs.push_back(nodeIDStrKey);
        seqs.push_back(seq);
        envelopes.push_back(envelopeEncoded);
    }

    if (!envs.empty())
    {
        // Perform multi-row insert into scphistory
        auto prepEnv =
            db.getPreparedStatement("INSERT INTO scphistory "
                                    "(nodeid, ledgerseq, envelope) VALUES "
                                    "(:n, :l, :e)",
                                    sess);
        prepEnv.exchange(soci::use(nodeIDs, "n"));
        prepEnv.exchange(soci::use(seqs, "l"));
        prepEnv.exchange(soci::use(envelopes, "e"));
        prepEnv.define_and_bind();
        {
            ZoneNamedN(insertSCPHistoryZone, "insert scphistory", true);
            prepEnv.execute(true);
        }
        if (prepEnv.get_affected_rows() != envs.size())
        {
            throw std::runtime_error("Could not update data in SQL");
        }
    }

    // save quorum information
    for (auto const& p : qmap)
    {
        auto const& nodeID = p.first;
        if (!p.second.mQuorumSet)
        {
            // skip node if we don't have its quorum set
            continue;
        }
        auto qSetH = xdrSha256(*(p.second.mQuorumSet));
        usedQSets.insert(std::make_pair(qSetH, p.second.mQuorumSet));

        std::string nodeIDStrKey = KeyUtils::toStrKey(nodeID);
        std::string qSetHHex(binToHex(qSetH));

        auto prep = db.getPreparedStatement(
            "UPDATE quoruminfo SET qsethash = :h WHERE nodeid = :id", sess);
        prep.exchange(soci::use(qSetHHex));
        prep.exchange(soci::use(nodeIDStrKey));
        prep.define_and_bind();
        {
            ZoneNamedN(updateQsetZone, "update quoruminfo", true);
            prep.execute(true);
        }
        if (prep.get_affected_rows() != 1)
        {
            auto prepI = db.getPreparedStatement(
                "INSERT INTO quoruminfo (nodeid, qsethash) VALUES (:id, :h)",
                sess);
            prepI.exchange(soci::use(nodeIDStrKey));
            prepI.exchange(soci::use(qSetHHex));
            prepI.define_and_bind();
            {
                ZoneNamedN(insertQsetZone, "insert quoruminfo", true);
                prepI.execute(true);
            }
            if (prepI.get_affected_rows() != 1)
            {
                throw std::runtime_error("Could not update data in SQL");
            }
        }
    }
    // save quorum sets
    for (auto const& p : usedQSets)
    {
        std::string qSetH = binToHex(p.first);

        uint32_t lastSeenSeq;
        auto prepSelQSet = db.getPreparedStatement(
            "SELECT lastledgerseq FROM scpquorums WHERE qsethash = :h", sess);
        prepSelQSet.exchange(soci::into(lastSeenSeq));
        prepSelQSet.exchange(soci::use(qSetH));
        prepSelQSet.define_and_bind();
        {
            ZoneNamedN(selectSCPQuorumsZone, "select scpquorums", true);
            prepSelQSet.execute(true);
        }

        if (prepSelQSet.got_data())
        {
            if (lastSeenSeq >= seq)
            {
                continue;
            }

            auto prepUpQSet = db.getPreparedStatement(
                "UPDATE scpquorums SET "
                "lastledgerseq = :l WHERE qsethash = :h",
                sess);

            prepUpQSet.exchange(soci::use(seq));
            prepUpQSet.exchange(soci::use(qSetH));
            prepUpQSet.define_and_bind();
            {
                ZoneNamedN(updateSCPQuorumsZone, "update scpquorums", true);
                prepUpQSet.execute(true);
            }
            if (prepUpQSet.get_affected_rows() != 1)
            {
                throw std::runtime_error("Could not update data in SQL");
            }
        }
        else
        {
            auto qSetBytes(xdr::xdr_to_opaque(*p.second));

            std::string qSetEncoded;
            qSetEncoded = decoder::encode_b64(qSetBytes);

            auto prepInsQSet = db.getPreparedStatement(
                "INSERT INTO scpquorums "
                "(qsethash, lastledgerseq, qset) VALUES "
                "(:h, :l, :v);",
                sess);

            prepInsQSet.exchange(soci::use(qSetH));
            prepInsQSet.exchange(soci::use(seq));
            prepInsQSet.exchange(soci::use(qSetEncoded));
            prepInsQSet.define_and_bind();
            {
                ZoneNamedN(insertSCPQuorumsZone, "insert scpquorums", true);
                prepInsQSet.execute(true);
            }
            if (prepInsQSet.get_affected_rows() != 1)
            {
                throw std::runtime_error("Could not update data in SQL");
            }
        }
    }

    txscope.commit();
}

size_t
HerderPersistence::copySCPHistoryToStream(soci::session& sess,
                                          uint32_t ledgerSeq,
                                          uint32_t ledgerCount,
                                          XDROutputFileStream& scpHistory)
{
    ZoneScoped;
    // Subtle: changing these queries may cause conflicts with main thread
    // https://github.com/stellar/stellar-core/issues/4589

    uint32_t begin = ledgerSeq, end = ledgerSeq + ledgerCount;
    size_t n = 0;

    // all known quorum sets
    UnorderedMap<Hash, SCPQuorumSet> qSets;

    for (uint32_t curLedgerSeq = begin; curLedgerSeq < end; curLedgerSeq++)
    {
        // SCP envelopes for this ledger
        // quorum sets missing in this batch of envelopes
        std::set<Hash> missingQSets;

        SCPHistoryEntry hEntryV;
        hEntryV.v(0);
        auto& hEntry = hEntryV.v0();
        auto& lm = hEntry.ledgerMessages;
        lm.ledgerSeq = curLedgerSeq;

        auto& curEnvs = lm.messages;

        // fetch SCP messages from history
        {
            std::string envB64;

            ZoneNamedN(selectSCPHistoryZone, "select scphistory", true);

            soci::statement st =
                (sess.prepare << "SELECT envelope FROM scphistory "
                                 "WHERE ledgerseq = :cur ORDER BY nodeid",
                 soci::into(envB64), soci::use(curLedgerSeq));

            st.execute(true);

            while (st.got_data())
            {
                curEnvs.emplace_back();
                auto& env = curEnvs.back();

                std::vector<uint8_t> envBytes;
                decoder::decode_b64(envB64, envBytes);
                xdr::xdr_from_opaque(envBytes, env);

                // record new quorum sets encountered
                Hash const& qSetHash =
                    Slot::getCompanionQuorumSetHashFromStatement(env.statement);
                if (qSets.find(qSetHash) == qSets.end())
                {
                    missingQSets.insert(qSetHash);
                }

                n++;

                st.fetch();
            }
        }

        // fetch the quorum sets from the db
        for (auto const& q : missingQSets)
        {
            std::string qset64, qSetHashHex;

            auto qset = getQuorumSet(sess, q);
            if (!qset)
            {
                throw std::runtime_error(
                    "corrupt database state: missing quorum set");
            }
            hEntry.quorumSets.emplace_back(std::move(*qset));
        }

        if (curEnvs.size() != 0)
        {
            scpHistory.writeOne(hEntryV);
        }
    }

    return n;
}

std::optional<Hash>
HerderPersistence::getNodeQuorumSet(soci::session& sess, NodeID const& nodeID)
{
    ZoneScoped;
    std::string nodeIDStrKey = KeyUtils::toStrKey(nodeID);
    std::string qsethHex;

    {
        ZoneNamedN(selectQsetZone, "select quoruminfo", true);
        soci::statement st = (sess.prepare << "SELECT qsethash FROM quoruminfo "
                                              "WHERE nodeid = :id",
                              soci::into(qsethHex), soci::use(nodeIDStrKey));

        st.execute(true);

        std::optional<Hash> res;
        if (st.got_data())
        {
            auto h = hexToBin256(qsethHex);
            res = std::make_optional<Hash>(std::move(h));
        }
        return res;
    }
}

SCPQuorumSetPtr
HerderPersistence::getQuorumSet(soci::session& sess, Hash const& qSetHash)
{
    ZoneScoped;
    SCPQuorumSetPtr res;
    SCPQuorumSet qset;
    std::string qset64, qSetHashHex;

    qSetHashHex = binToHex(qSetHash);

    {
        ZoneNamedN(selectSCPQuorumsZone, "select scpquorums", true);
        soci::statement st = (sess.prepare << "SELECT qset FROM scpquorums "
                                              "WHERE qsethash = :h",
                              soci::into(qset64), soci::use(qSetHashHex));

        st.execute(true);

        if (st.got_data())
        {
            std::vector<uint8_t> qSetBytes;
            decoder::decode_b64(qset64, qSetBytes);

            xdr::xdr_get g1(&qSetBytes.front(), &qSetBytes.back() + 1);
            xdr_argpack_archive(g1, qset);

            res = std::make_shared<SCPQuorumSet>(std::move(qset));
        }
        return res;
    }
}

void
HerderPersistence::maybeDropAndCreateNew(soci::session& sess)
{
    ZoneScoped;
    sess << "DROP TABLE IF EXISTS scphistory";

    sess << "DROP TABLE IF EXISTS scpquorums";

    sess << "CREATE TABLE scphistory ("
            "nodeid      CHARACTER(56) NOT NULL,"
            "ledgerseq   INT NOT NULL CHECK (ledgerseq >= 0),"
            "envelope    TEXT NOT NULL"
            ")";

    sess << "CREATE INDEX scpenvsbyseq ON scphistory(ledgerseq)";

    sess << "CREATE TABLE scpquorums ("
            "qsethash      CHARACTER(64) NOT NULL,"
            "lastledgerseq INT NOT NULL CHECK (lastledgerseq >= 0),"
            "qset          TEXT NOT NULL,"
            "PRIMARY KEY (qsethash)"
            ")";

    sess << "CREATE INDEX scpquorumsbyseq ON scpquorums(lastledgerseq)";

    sess << "DROP TABLE IF EXISTS quoruminfo";
    sess << "CREATE TABLE quoruminfo ("
            "nodeid      CHARACTER(56) NOT NULL,"
            "qsethash    CHARACTER(64) NOT NULL,"
            "PRIMARY KEY (nodeid))";
}

void
HerderPersistence::deleteOldEntries(soci::session& sess, uint32_t ledgerSeq,
                                    uint32_t count)
{
    ZoneScoped;
    DatabaseUtils::deleteOldEntriesHelper(sess, ledgerSeq, count, "scphistory",
                                          "ledgerseq");
    DatabaseUtils::deleteOldEntriesHelper(sess, ledgerSeq, count, "scpquorums",
                                          "lastledgerseq");
}
}