clean up ledger page interface. wire up account_tx

reporting/ETLSource.cpp

@@ -31,7 +31,7 @@
 // Primarly used in read-only mode, to monitor when ledgers are validated
 ETLSource::ETLSource(
     boost::json::object const& config,
-    CassandraFlatMapBackend& backend,
+    BackendInterface& backend,
     NetworkValidatedLedgers& networkValidatedLedgers,
     boost::asio::io_context& ioContext)
     : ioc_(ioContext)
@@ -400,7 +400,7 @@ public:
     process(
         std::unique_ptr<org::xrpl::rpc::v1::XRPLedgerAPIService::Stub>& stub,
         grpc::CompletionQueue& cq,
-        CassandraFlatMapBackend& backend,
+        BackendInterface& backend,
         bool abort = false)
     {
         BOOST_LOG_TRIVIAL(info) << "Processing response. "
@@ -582,7 +582,7 @@ ETLSource::fetchLedger(uint32_t ledgerSequence, bool getObjects)
 }
 ETLLoadBalancer::ETLLoadBalancer(
     boost::json::array const& config,
-    CassandraFlatMapBackend& backend,
+    BackendInterface& backend,
     NetworkValidatedLedgers& nwvl,
     boost::asio::io_context& ioContext)
 {

reporting/ETLSource.h

@@ -25,7 +25,7 @@
 #include <boost/beast/core.hpp>
 #include <boost/beast/core/string.hpp>
 #include <boost/beast/websocket.hpp>
-#include <reporting/ReportingBackend.h>
+#include <reporting/BackendInterface.h>
 
 #include "org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h"
 #include <grpcpp/grpcpp.h>
@@ -85,7 +85,7 @@ class ETLSource
     // used for retrying connections
     boost::asio::steady_timer timer_;
 
-    CassandraFlatMapBackend& backend_;
+    BackendInterface& backend_;
 
 public:
     bool
@@ -113,7 +113,7 @@ public:
     /// Primarly used in read-only mode, to monitor when ledgers are validated
     ETLSource(
         boost::json::object const& config,
-        CassandraFlatMapBackend& backend,
+        BackendInterface& backend,
         NetworkValidatedLedgers& networkValidatedLedgers,
         boost::asio::io_context& ioContext);
 
@@ -285,7 +285,7 @@ private:
 public:
     ETLLoadBalancer(
         boost::json::array const& config,
-        CassandraFlatMapBackend& backend,
+        BackendInterface& backend,
         NetworkValidatedLedgers& nwvl,
         boost::asio::io_context& ioContext);
 

reporting/ReportingBackend.h

@@ -55,7 +55,7 @@ flatMapWriteLedgerHeaderCallback(CassFuture* fut, void* cbData);
 void
 flatMapWriteLedgerHashCallback(CassFuture* fut, void* cbData);
 
-class CassandraFlatMapBackend : BackendInterface
+class CassandraFlatMapBackend : public BackendInterface
 {
 private:
     // convenience function for one-off queries. For normal reads and writes,
@@ -141,7 +141,7 @@ public:
     {
     }
 
-    ~CassandraFlatMapBackend()
+    ~CassandraFlatMapBackend() override
     {
         if (open_)
             close();
@@ -915,16 +915,23 @@ public:
     }
     BackendInterface::LedgerPage
     fetchLedgerPage(
-        std::optional<BackendInterface::LedgerCursor> const& cursor,
+        std::optional<ripple::uint256> const& cursor,
         std::uint32_t ledgerSequence,
         std::uint32_t limit) const override
     {
         BOOST_LOG_TRIVIAL(debug) << "Starting doUpperBound";
         CassStatement* statement = cass_prepared_bind(upperBound_);
         cass_statement_set_consistency(statement, CASS_CONSISTENCY_QUORUM);
-        int64_t cursorVal = cursor.has_value() ? cursor.value() : INT64_MIN;
 
-        CassError rc = cass_statement_bind_int64(statement, 0, cursorVal);
+        int64_t intCursor = INT64_MIN;
+        if (cursor)
+        {
+            auto token = getToken(cursor->data());
+            if (token)
+                intCursor = *token;
+        }
+
+        CassError rc = cass_statement_bind_int64(statement, 0, intCursor);
         if (rc != CASS_OK)
         {
             cass_statement_free(statement);
@@ -1017,9 +1024,7 @@ public:
             {
                 results.push_back({keys[i], objs[i]});
             }
-            auto token = getToken(results[results.size() - 1].key.data());
-            assert(token);
-            return {results, token};
+            return {results, keys[keys.size() - 1]};
         }
 
         return {{}, {}};

reporting/ReportingETL.cpp

@@ -18,6 +18,7 @@
 //==============================================================================
 
 #include <ripple/basics/StringUtilities.h>
+#include <reporting/BackendFactory.h>
 #include <reporting/DBHelpers.h>
 #include <reporting/ReportingETL.h>
 
@@ -76,7 +77,7 @@ ReportingETL::insertTransactions(
         auto journal = ripple::debugLog();
         accountTxData.emplace_back(txMeta, std::move(nodestoreHash), journal);
         std::string keyStr{(const char*)sttx.getTransactionID().data(), 32};
-        flatMapBackend_.writeTransaction(
+        flatMapBackend_->writeTransaction(
             std::move(keyStr),
             ledger.seq,
             std::move(*raw),
@@ -89,7 +90,7 @@ std::optional<ripple::LedgerInfo>
 ReportingETL::loadInitialLedger(uint32_t startingSequence)
 {
     // check that database is actually empty
-    auto ledger = flatMapBackend_.fetchLedgerBySequence(startingSequence);
+    auto ledger = flatMapBackend_->fetchLedgerBySequence(startingSequence);
     if (ledger)
     {
         BOOST_LOG_TRIVIAL(fatal) << __func__ << " : "
@@ -128,9 +129,9 @@ ReportingETL::loadInitialLedger(uint32_t startingSequence)
     {
         for (auto& data : accountTxData)
         {
-            flatMapBackend_.writeAccountTransactions(std::move(data));
+            flatMapBackend_->writeAccountTransactions(std::move(data));
         }
-        bool success = flatMapBackend_.writeLedger(
+        bool success = flatMapBackend_->writeLedger(
             lgrInfo, std::move(*ledgerData->mutable_ledger_header()));
     }
     auto end = std::chrono::system_clock::now();
@@ -155,7 +156,7 @@ ReportingETL::publishLedger(uint32_t ledgerSequence, uint32_t maxAttempts)
     size_t numAttempts = 0;
     while (!stopping_)
     {
-        auto ledger = flatMapBackend_.fetchLedgerBySequence(ledgerSequence);
+        auto ledger = flatMapBackend_->fetchLedgerBySequence(ledgerSequence);
 
         if (!ledger)
         {
@@ -292,7 +293,7 @@ ReportingETL::buildNextLedger(org::xrpl::rpc::v1::GetLedgerResponse& rawData)
         }
 
         assert(not(isCreated and isDeleted));
-        flatMapBackend_.writeLedgerObject(
+        flatMapBackend_->writeLedgerObject(
             std::move(*obj.mutable_key()),
             lgrInfo.seq,
             std::move(*obj.mutable_data()),
@@ -302,9 +303,9 @@ ReportingETL::buildNextLedger(org::xrpl::rpc::v1::GetLedgerResponse& rawData)
     }
     for (auto& data : accountTxData)
     {
-        flatMapBackend_.writeAccountTransactions(std::move(data));
+        flatMapBackend_->writeAccountTransactions(std::move(data));
     }
-    bool success = flatMapBackend_.writeLedger(
+    bool success = flatMapBackend_->writeLedger(
         lgrInfo, std::move(*rawData.mutable_ledger_header()));
     BOOST_LOG_TRIVIAL(debug)
         << __func__ << " : "
@@ -347,7 +348,7 @@ ReportingETL::runETLPipeline(uint32_t startSequence)
                              << "Starting etl pipeline";
     writing_ = true;
 
-    auto parent = flatMapBackend_.fetchLedgerBySequence(startSequence - 1);
+    auto parent = flatMapBackend_->fetchLedgerBySequence(startSequence - 1);
     if (!parent)
     {
         assert(false);
@@ -482,7 +483,7 @@ void
 ReportingETL::monitor()
 {
     std::optional<uint32_t> latestSequence =
-        flatMapBackend_.fetchLatestLedgerSequence();
+        flatMapBackend_->fetchLatestLedgerSequence();
     if (!latestSequence)
     {
         BOOST_LOG_TRIVIAL(info) << __func__ << " : "
@@ -637,17 +638,16 @@ ReportingETL::ReportingETL(
     boost::asio::io_context& ioc)
     : publishStrand_(ioc)
     , ioContext_(ioc)
-    , flatMapBackend_(
-          config.at("database").as_object().at("cassandra").as_object())
+    , flatMapBackend_(makeBackend(config))
     , pgPool_(make_PgPool(
           config.at("database").as_object().at("postgres").as_object()))
     , loadBalancer_(
           config.at("etl_sources").as_array(),
-          flatMapBackend_,
+          *flatMapBackend_,
           networkValidatedLedgers_,
           ioc)
 {
-    flatMapBackend_.open();
+    flatMapBackend_->open();
     initSchema(pgPool_);
 }
 

reporting/ReportingETL.h

@@ -25,10 +25,10 @@
 #include <boost/beast/core.hpp>
 #include <boost/beast/core/string.hpp>
 #include <boost/beast/websocket.hpp>
+#include <reporting/BackendInterface.h>
 #include <reporting/ETLHelpers.h>
 #include <reporting/ETLSource.h>
 #include <reporting/Pg.h>
-#include <reporting/ReportingBackend.h>
 
 #include "org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h"
 #include <grpcpp/grpcpp.h>
@@ -59,7 +59,7 @@ struct AccountTransactionsData;
 class ReportingETL
 {
 private:
-    CassandraFlatMapBackend flatMapBackend_;
+    std::unique_ptr<BackendInterface> flatMapBackend_;
     std::shared_ptr<PgPool> pgPool_;
 
     std::thread worker_;
@@ -321,10 +321,10 @@ public:
         return loadBalancer_;
     }
 
-    CassandraFlatMapBackend&
+    BackendInterface&
     getFlatMapBackend()
     {
-        return flatMapBackend_;
+        return *flatMapBackend_;
     }
 
     std::shared_ptr<PgPool>&