feat: Add migration backfill for mptoken_issuance_history

src/data/cassandra/impl/Result.cpp

@@ -29,6 +29,12 @@ Result::hasRows() const
     return numRows() > 0;
 }
 
+[[nodiscard]] bool
+Result::hasMorePages() const
+{
+    return cass_result_has_more_pages(*this) != 0u;
+}
+
 /* implicit */ ResultIterator::ResultIterator(CassIterator* ptr)
     : ManagedObject{ptr, kResultIteratorDeleter}, hasMore_{cass_iterator_next(ptr) != 0u}
 {

src/data/cassandra/impl/Result.hpp

@@ -107,6 +107,9 @@ struct Result : public ManagedObject<CassResult const> {
     [[nodiscard]] bool
     hasRows() const;
 
+    [[nodiscard]] bool
+    hasMorePages() const;
+
     template <typename... RowTypes>
     [[nodiscard]] std::optional<std::tuple<RowTypes...>>
     get() const

src/data/cassandra/impl/Statement.hpp

@@ -3,6 +3,7 @@
 #include "data/cassandra/Types.hpp"
 #include "data/cassandra/impl/Collection.hpp"
 #include "data/cassandra/impl/ManagedObject.hpp"
+#include "data/cassandra/impl/Result.hpp"
 #include "data/cassandra/impl/Tuple.hpp"
 #include "util/UnsupportedType.hpp"
 
@@ -65,6 +66,32 @@ class Statement : public ManagedObject<CassStatement> {
         (this->bindAt<Args>(idx++, std::forward<Args>(args)), ...);
     }
 
+    /**
+     * @brief Set the Cassandra driver page size for this statement.
+     *
+     * @param pageSize Number of rows per driver page
+     */
+    void
+    setPagingSize(std::int32_t const pageSize) const
+    {
+        auto const rc = cass_statement_set_paging_size(*this, pageSize);
+        if (rc != CASS_OK)
+            throw std::logic_error(fmt::format("[Set paging size]: {}", cass_error_desc(rc)));
+    }
+
+    /**
+     * @brief Continue this statement from the paging state in a previous result.
+     *
+     * @param result Previous page result
+     */
+    void
+    setPagingState(Result const& result) const
+    {
+        auto const rc = cass_statement_set_paging_state(*this, result);
+        if (rc != CASS_OK)
+            throw std::logic_error(fmt::format("[Set paging state]: {}", cass_error_desc(rc)));
+    }
+
     /**
      * @brief Binds an argument to a specific index.
      *

src/main/Main.cpp

@@ -46,6 +46,7 @@ runApp(int argc, char const* argv[])
             if (not app::parseConfig(migrate.configPath))
                 return EXIT_FAILURE;
 
+            PrometheusService::init(getClioConfig());
             if (auto const initSuccess = util::LogService::init(getClioConfig()); not initSuccess) {
                 std::cerr << initSuccess.error() << std::endl;
                 return EXIT_FAILURE;

src/migration/CMakeLists.txt

@@ -6,8 +6,9 @@ target_sources(
         MigrationApplication.cpp
         impl/MigrationManagerFactory.cpp
         MigratorStatus.cpp
+        cassandra/MPTTransactionHistoryMigrator.cpp
         cassandra/impl/ObjectsAdapter.cpp
         cassandra/impl/TransactionsAdapter.cpp
 )
 
-target_link_libraries(clio_migration PRIVATE clio_util clio_data)
+target_link_libraries(clio_migration PRIVATE clio_util clio_data clio_etl)

src/migration/MigrationApplication.cpp

@@ -1,5 +1,6 @@
 #include "migration/MigrationApplication.hpp"
 
+#include "migration/MigrationManagerInterface.hpp"
 #include "migration/MigratiorStatus.hpp"
 #include "migration/impl/MigrationManagerFactory.hpp"
 #include "util/OverloadSet.hpp"
@@ -9,6 +10,7 @@
 
 #include <cstdlib>
 #include <iostream>
+#include <memory>
 #include <ostream>
 #include <stdexcept>
 #include <string>
@@ -23,7 +25,8 @@ MigratorApplication::MigratorApplication(
 )
     : cmd_(std::move(command))
 {
-    PrometheusService::init(config);
+    if (not PrometheusService::isInitialised())
+        PrometheusService::init(config);
 
     auto expectedMigrationManager = migration::impl::makeMigrationManager(config, cache_);
 
@@ -36,6 +39,14 @@ MigratorApplication::MigratorApplication(
     migrationManager_ = std::move(expectedMigrationManager.value());
 }
 
+MigratorApplication::MigratorApplication(
+    MigrateSubCmd command,
+    std::shared_ptr<migration::MigrationManagerInterface> migrationManager
+)
+    : migrationManager_(std::move(migrationManager)), cmd_(std::move(command))
+{
+}
+
 int
 MigratorApplication::run()
 {

src/migration/MigrationApplication.hpp

@@ -69,6 +69,20 @@ class MigratorApplication {
      */
     MigratorApplication(util::config::ClioConfigDefinition const& config, MigrateSubCmd command);
 
+    /**
+     * @brief Construct a new MigratorApplication object with an already-built migration manager.
+     *
+     * This overload bypasses backend/manager creation from config and is primarily intended for
+     * tests that inject a mock manager.
+     *
+     * @param command The command to run
+     * @param migrationManager The migration manager to use
+     */
+    MigratorApplication(
+        MigrateSubCmd command,
+        std::shared_ptr<migration::MigrationManagerInterface> migrationManager
+    );
+
     /**
      * @brief Run the application
      *

src/migration/cassandra/CassandraMigrationBackend.hpp

@@ -3,14 +3,20 @@
 #include "data/CassandraBackend.hpp"
 #include "data/LedgerCacheInterface.hpp"
 #include "data/cassandra/SettingsProvider.hpp"
+#include "data/cassandra/Types.hpp"
 #include "migration/cassandra/impl/CassandraMigrationSchema.hpp"
 #include "migration/cassandra/impl/Spec.hpp"
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/spawn.hpp>
+#include <fmt/core.h>
 
 #include <cstdint>
+#include <memory>
+#include <mutex>
+#include <stdexcept>
 #include <string>
+#include <unordered_map>
 #include <utility>
 
 namespace migration::cassandra {
@@ -20,9 +26,39 @@ namespace migration::cassandra {
  * migration specific functionalities.
  */
 class CassandraMigrationBackend : public data::cassandra::CassandraBackend {
+    // Internal full-scan page size: large enough to avoid excessive driver round trips, bounded so
+    // one migration read cannot materialize an unbounded token range page. Operators still control
+    // concurrency through the existing migration thread/job settings.
+    static constexpr std::int32_t kFullScanPageSize = 5'000;
+
     util::Logger log_{"Migration"};
     data::cassandra::SettingsProvider settingsProvider_;
     impl::CassandraMigrationSchema migrationSchema_;
+    std::mutex fullScanStatementsMutex_;
+    std::unordered_map<std::string, std::shared_ptr<data::cassandra::PreparedStatement>>
+        fullScanStatements_;
+
+    template <impl::TableSpec TableDesc>
+    std::shared_ptr<data::cassandra::PreparedStatement>
+    getPreparedFullScanStatement()
+    {
+        // The table name uniquely identifies the statement: partition key and select columns are
+        // fixed per TableDesc.
+        std::string const statementKey = TableDesc::kTableName;
+
+        std::scoped_lock const lock{fullScanStatementsMutex_};
+        if (auto const statement = fullScanStatements_.find(statementKey);
+            statement != fullScanStatements_.end()) {
+            return statement->second;
+        }
+
+        auto statement = std::make_shared<data::cassandra::PreparedStatement>(
+            migrationSchema_.getPreparedFullScanStatement(
+                handle_, TableDesc::kTableName, TableDesc::kSelectColumns, TableDesc::kPartitionKey
+            )
+        );
+        return fullScanStatements_.emplace(statementKey, std::move(statement)).first->second;
+    }
 
 public:
     /**
@@ -59,34 +95,51 @@ class CassandraMigrationBackend : public data::cassandra::CassandraBackend {
         boost::asio::yield_context yield
     )
     {
-        LOG(log_.debug()) << "Travsering token range: " << start << " - " << end
+        LOG(log_.debug()) << "Traversing token range: " << start << " - " << end
                           << " ; table: " << TableDesc::kTableName;
-        // for each table we only have one prepared statement
-        static auto kStatementPrepared = migrationSchema_.getPreparedFullScanStatement(
-            handle_, TableDesc::kTableName, TableDesc::kPartitionKey
-        );
 
-        auto const statement = kStatementPrepared.bind(start, end);
+        auto const statementPrepared = getPreparedFullScanStatement<TableDesc>();
+        auto statement = statementPrepared->bind(start, end);
+        statement.setPagingSize(kFullScanPageSize);
 
-        auto const res = this->executor_.read(yield, statement);
-        if (not res) {
-            LOG(log_.error()) << "Could not fetch data from table: " << TableDesc::kTableName
-                              << " range: " << start << " - " << end << ";" << res.error();
-            return;
-        }
+        bool sawRows = false;
+        while (true) {
+            auto const res = this->executor_.read(yield, statement);
+            if (not res) {
+                // Fail closed: a swallowed read error would leave a gap in the scanned data while
+                // the migrator is still marked Migrated. Throwing aborts the migration so its
+                // status stays NotMigrated and the operator can rerun.
+                auto const errorMessage = fmt::format(
+                    "Migration scan failed to read table '{}' in token range [{}, {}]: {}",
+                    TableDesc::kTableName,
+                    start,
+                    end,
+                    res.error().message()
+                );
+                LOG(log_.error()) << errorMessage;
+                throw std::runtime_error(errorMessage);
+            }
 
-        auto const& results = res.value();
-        if (not results.hasRows()) {
-            LOG(log_.debug()) << "No rows returned  - table: " << TableDesc::kTableName
-                              << " range: " << start << " - " << end;
-            return;
+            auto const& results = res.value();
+            for (auto const& row : std::apply(
+                     [&](auto... args) {
+                         return data::cassandra::extract<decltype(args)...>(results);
+                     },
+                     typename TableDesc::Row{}
+                 )) {
+                callback(row);
+                sawRows = true;
+            }
+
+            if (not results.hasMorePages())
+                break;
+
+            statement.setPagingState(results);
         }
 
-        for (auto const& row : std::apply(
-                 [&](auto... args) { return data::cassandra::extract<decltype(args)...>(results); },
-                 typename TableDesc::Row{}
-             )) {
-            callback(row);
+        if (not sawRows) {
+            LOG(log_.debug()) << "No rows returned - table: " << TableDesc::kTableName
+                              << " range: " << start << " - " << end;
         }
     }
 };

src/migration/cassandra/CassandraMigrationManager.hpp

@@ -2,6 +2,7 @@
 
 #include "data/BackendInterface.hpp"
 #include "migration/cassandra/CassandraMigrationBackend.hpp"
+#include "migration/cassandra/MPTTransactionHistoryMigrator.hpp"
 #include "migration/impl/MigrationInspectorBase.hpp"
 #include "migration/impl/MigrationManagerBase.hpp"
 #include "migration/impl/MigratorsRegister.hpp"
@@ -11,7 +12,8 @@ namespace {
 // Register migrators here
 // MigratorsRegister<BackendType, ExampleMigrator>
 template <typename BackendType>
-using CassandraSupportedMigrators = migration::impl::MigratorsRegister<BackendType>;
+using CassandraSupportedMigrators = migration::impl::
+    MigratorsRegister<BackendType, migration::cassandra::MPTTransactionHistoryMigrator>;
 
 //  Instantiates with the backend which supports actual migration running
 using MigrationProcessor =

src/migration/cassandra/MPTTransactionHistoryMigrator.cpp

@@ -0,0 +1,88 @@
+#include "migration/cassandra/MPTTransactionHistoryMigrator.hpp"
+
+#include "data/DBHelpers.hpp"
+#include "etl/MPTHelpers.hpp"
+#include "migration/cassandra/impl/TransactionsAdapter.hpp"
+#include "migration/cassandra/impl/Types.hpp"
+#include "util/Batching.hpp"
+#include "util/config/ObjectView.hpp"
+
+#include <fmt/format.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/protocol/TxMeta.h>
+
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <stdexcept>
+#include <utility>
+#include <vector>
+
+namespace migration::cassandra {
+
+void
+MPTTransactionHistoryMigrator::runMigration(
+    std::shared_ptr<Backend> const& backend,
+    util::config::ObjectView const& config
+)
+{
+    auto const fullScanThreads = config.get<std::uint32_t>("full_scan_threads");
+    auto const fullScanJobs = config.get<std::uint32_t>("full_scan_jobs");
+    auto const cursorsPerJob = config.get<std::uint32_t>("cursors_per_job");
+
+    // Records are buffered across transactions and flushed in batches so the backend can coalesce
+    // them into full-size write batches, mirroring the per-ledger accumulation of the live ETL
+    // path, instead of submitting one tiny write pair per transaction.
+    static constexpr std::size_t kWriteBatchRecords = 1'000;
+    util::BatchBuffer<MPTokenIssuanceTransactionsData> batchBuffer{
+        kWriteBatchRecords,
+        [&backend](std::vector<MPTokenIssuanceTransactionsData> const& records) {
+            backend->writeMPTokenIssuanceTransactions(records);
+            backend->writeAccountMPTokenIssuanceTransactions(records);
+        }
+    };
+
+    // Atomic because the adapter invokes the callback concurrently across scan workers.
+    std::atomic<std::uint64_t> undecodableRows{0};
+
+    // Full-scan the transactions table in parallel; for each transaction reuse the live ETL
+    // extractor to derive the touched MPT issuances and affected accounts, then write both index
+    // shapes. Token-range-edge re-reads and post-crash reruns re-upsert identical deterministic-key
+    // rows, so the scan is idempotent without explicit deduplication.
+    impl::TransactionsScanner scanner(
+        {.ctxThreadsNum = fullScanThreads, .jobsNum = fullScanJobs, .cursorsPerJob = cursorsPerJob},
+        impl::TransactionsAdapter(
+            backend,
+            [&](xrpl::STTx const& sttx, xrpl::TxMeta const& txMeta) {
+                auto indexData = etl::getMPTokenIssuanceTxsFromTx(txMeta, sttx);
+                if (indexData.empty())
+                    return;
+                batchBuffer.add(std::move(indexData));
+            },
+            [&] { ++undecodableRows; }
+        )
+    );
+    scanner.waitForAllAndThrowOnError();
+
+    // All workers are joined; flush the remaining buffered records.
+    batchBuffer.flush();
+
+    // Flush queued async writes so the migrator is not marked Migrated before all index rows are
+    // durable.
+    backend->waitForWritesToFinish();
+
+    // Abort on any decode failure so the migrator stays NotMigrated (status is only written when
+    // runMigration returns normally). The scan is idempotent, so a rerun re-upserts safely.
+    if (auto const skipped = undecodableRows.load(); skipped > 0) {
+        throw std::runtime_error(
+            fmt::format(
+                "MPT backfill: {} transactions failed to deserialize; aborting so the migrator "
+                "stays NotMigrated",
+                skipped
+            )
+        );
+    }
+}
+
+}  // namespace migration::cassandra