diff --git a/src/data/cassandra/impl/Result.cpp b/src/data/cassandra/impl/Result.cpp index e5ffb66196..a6461085ec 100644 --- a/src/data/cassandra/impl/Result.cpp +++ b/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} { diff --git a/src/data/cassandra/impl/Result.hpp b/src/data/cassandra/impl/Result.hpp index 75a1b61a8a..06d6026978 100644 --- a/src/data/cassandra/impl/Result.hpp +++ b/src/data/cassandra/impl/Result.hpp @@ -107,6 +107,9 @@ struct Result : public ManagedObject { [[nodiscard]] bool hasRows() const; + [[nodiscard]] bool + hasMorePages() const; + template [[nodiscard]] std::optional> get() const diff --git a/src/data/cassandra/impl/Statement.hpp b/src/data/cassandra/impl/Statement.hpp index 99fd6ab04c..e95a7d189f 100644 --- a/src/data/cassandra/impl/Statement.hpp +++ b/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 { (this->bindAt(idx++, std::forward(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. * diff --git a/src/main/Main.cpp b/src/main/Main.cpp index bf0df548e2..9e23702823 100644 --- a/src/main/Main.cpp +++ b/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; diff --git a/src/migration/CMakeLists.txt b/src/migration/CMakeLists.txt index b62a09563e..888aabebfa 100644 --- a/src/migration/CMakeLists.txt +++ b/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) diff --git a/src/migration/MigrationApplication.cpp b/src/migration/MigrationApplication.cpp index e942690d1e..99c3db1cfe 100644 --- a/src/migration/MigrationApplication.cpp +++ b/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 #include +#include #include #include #include @@ -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 migrationManager +) + : migrationManager_(std::move(migrationManager)), cmd_(std::move(command)) +{ +} + int MigratorApplication::run() { diff --git a/src/migration/MigrationApplication.hpp b/src/migration/MigrationApplication.hpp index f0e425678c..9d748fb843 100644 --- a/src/migration/MigrationApplication.hpp +++ b/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 migrationManager + ); + /** * @brief Run the application * diff --git a/src/migration/cassandra/CassandraMigrationBackend.hpp b/src/migration/cassandra/CassandraMigrationBackend.hpp index 33a15f474e..8d1b047354 100644 --- a/src/migration/cassandra/CassandraMigrationBackend.hpp +++ b/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 +#include #include +#include +#include +#include #include +#include #include 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> + fullScanStatements_; + + template + std::shared_ptr + 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( + 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(); + 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(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(results); }, - typename TableDesc::Row{} - )) { - callback(row); + if (not sawRows) { + LOG(log_.debug()) << "No rows returned - table: " << TableDesc::kTableName + << " range: " << start << " - " << end; } } }; diff --git a/src/migration/cassandra/CassandraMigrationManager.hpp b/src/migration/cassandra/CassandraMigrationManager.hpp index 2e0f8e4c7c..894f4f7fc9 100644 --- a/src/migration/cassandra/CassandraMigrationManager.hpp +++ b/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 template -using CassandraSupportedMigrators = migration::impl::MigratorsRegister; +using CassandraSupportedMigrators = migration::impl:: + MigratorsRegister; // Instantiates with the backend which supports actual migration running using MigrationProcessor = diff --git a/src/migration/cassandra/MPTTransactionHistoryMigrator.cpp b/src/migration/cassandra/MPTTransactionHistoryMigrator.cpp new file mode 100644 index 0000000000..d179c31365 --- /dev/null +++ b/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 +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace migration::cassandra { + +void +MPTTransactionHistoryMigrator::runMigration( + std::shared_ptr const& backend, + util::config::ObjectView const& config +) +{ + auto const fullScanThreads = config.get("full_scan_threads"); + auto const fullScanJobs = config.get("full_scan_jobs"); + auto const cursorsPerJob = config.get("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 batchBuffer{ + kWriteBatchRecords, + [&backend](std::vector const& records) { + backend->writeMPTokenIssuanceTransactions(records); + backend->writeAccountMPTokenIssuanceTransactions(records); + } + }; + + // Atomic because the adapter invokes the callback concurrently across scan workers. + std::atomic 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 diff --git a/src/migration/cassandra/MPTTransactionHistoryMigrator.hpp b/src/migration/cassandra/MPTTransactionHistoryMigrator.hpp new file mode 100644 index 0000000000..d80afa236a --- /dev/null +++ b/src/migration/cassandra/MPTTransactionHistoryMigrator.hpp @@ -0,0 +1,40 @@ +#pragma once + +#include "migration/cassandra/CassandraMigrationBackend.hpp" +#include "util/config/ObjectView.hpp" + +#include + +namespace migration::cassandra { + +/** + * @brief Backfill migrator that indexes historical MPT issuance transactions. + * + * Full-scans the transactions table and reuses the live ETL extractor to populate the + * mptoken_issuance_transactions and account_mptoken_issuance_transactions index tables, letting a + * node serve complete results for transactions that predate live MPT indexing. The migrator is + * non-blocking: it deliberately omits kCanBlockClio so a missing backfill never blocks Clio + * startup. + */ +struct MPTTransactionHistoryMigrator { + /** @brief Unique name used to identify and invoke this migrator. */ + static constexpr char const* kName = "MPTTransactionHistoryMigrator"; + + /** @brief Human-readable description of the migrator. */ + static constexpr char const* kDescription = + "Backfills the MPT issuance transaction index tables from historical transactions"; + + /** @brief The backend type this migrator runs against. */ + using Backend = CassandraMigrationBackend; + + /** + * @brief Run the backfill: full-scan the transactions table and write MPT index rows. + * + * @param backend The migration backend. + * @param config The migration configuration (the .migration config section). + */ + static void + runMigration(std::shared_ptr const& backend, util::config::ObjectView const& config); +}; + +} // namespace migration::cassandra diff --git a/src/migration/cassandra/impl/CassandraMigrationSchema.hpp b/src/migration/cassandra/impl/CassandraMigrationSchema.hpp index eb1bbb9f93..c71e05ea94 100644 --- a/src/migration/cassandra/impl/CassandraMigrationSchema.hpp +++ b/src/migration/cassandra/impl/CassandraMigrationSchema.hpp @@ -36,6 +36,7 @@ class CassandraMigrationSchema { * * @param handler The database handler * @param tableName The name of the table + * @param selectColumns Explicit columns to select, in TableDesc::Row order * @param key The partition key of the table * @return The prepared statement */ @@ -43,16 +44,18 @@ class CassandraMigrationSchema { getPreparedFullScanStatement( data::cassandra::Handle const& handler, std::string const& tableName, + std::string const& selectColumns, std::string const& key ) { return handler.prepare( fmt::format( R"( - SELECT * + SELECT {} FROM {} WHERE TOKEN({}) >= ? AND TOKEN({}) <= ? )", + selectColumns, data::cassandra::qualifiedTableName( settingsProvider_.get(), tableName ), diff --git a/src/migration/cassandra/impl/FullTableScanner.hpp b/src/migration/cassandra/impl/FullTableScanner.hpp index 9cb0e39af9..24bafb71a6 100644 --- a/src/migration/cassandra/impl/FullTableScanner.hpp +++ b/src/migration/cassandra/impl/FullTableScanner.hpp @@ -7,12 +7,15 @@ #include "util/async/context/BasicExecutionContext.hpp" #include +#include #include #include #include #include #include +#include +#include #include namespace migration::cassandra::impl { @@ -156,13 +159,33 @@ class FullTableScanner { } /** - * @brief Wait for all workers to finish. + * @brief Wait once for all workers to finish, propagating any worker failure. + * + * Inspects each worker's result rather than discarding it, so a failed token-range read (or any + * exception thrown while scanning) surfaces here instead of being silently dropped. Every + * worker is awaited before throwing, so all are joined regardless of failure. This consumes the + * worker operations and must only be called once. + * + * @throws std::runtime_error if any worker reported an error, so the migration fails closed. */ void - wait() + waitForAllAndThrowOnError() { + std::vector errors; for (auto& task : tasks_) { - task.wait(); + if (auto const result = task.get(); not result.has_value()) + errors.push_back(result.error().message); + } + + if (not errors.empty()) { + throw std::runtime_error( + fmt::format( + "Full table scan failed: {} of {} workers reported errors; first error: {}", + errors.size(), + tasks_.size(), + errors.front() + ) + ); } } }; diff --git a/src/migration/cassandra/impl/ObjectsAdapter.hpp b/src/migration/cassandra/impl/ObjectsAdapter.hpp index e8938128b3..aeb6886176 100644 --- a/src/migration/cassandra/impl/ObjectsAdapter.hpp +++ b/src/migration/cassandra/impl/ObjectsAdapter.hpp @@ -22,8 +22,10 @@ namespace migration::cassandra::impl { * @brief The description of the objects table. It has to be a TableSpec. */ struct TableObjectsDesc { + // Must match kSelectColumns order. using Row = std::tuple; static constexpr char const* kPartitionKey = "key"; + static constexpr char const* kSelectColumns = "key, sequence, object"; static constexpr char const* kTableName = "objects"; }; diff --git a/src/migration/cassandra/impl/Spec.hpp b/src/migration/cassandra/impl/Spec.hpp index 6e64263477..a596adeb83 100644 --- a/src/migration/cassandra/impl/Spec.hpp +++ b/src/migration/cassandra/impl/Spec.hpp @@ -2,21 +2,35 @@ #include +#include #include +#include +#include #include -#include namespace migration::cassandra::impl { + +// The number of columns in a comma-separated column list. +constexpr std::size_t +columnCount(std::string_view const selectColumns) +{ + return static_cast(std::ranges::count(selectColumns, ',')) + 1; +} + // Define the concept for a class like TableObjectsDesc template concept TableSpec = requires { - // Check that 'row' exists and is a tuple - // keys types are at the beginning and the other fields types sort in alphabetical order + // Check that 'Row' exists and is a tuple whose element types match kSelectColumns order typename T::Row; - requires std::tuple_size_v >= 0; // Ensures 'row' is a tuple + requires std::tuple_size_v >= 0; // Ensures 'Row' is a tuple - // Check that static constexpr members 'partitionKey' and 'tableName' exist + // Check that static constexpr members for the scan query exist. { T::kPartitionKey } -> std::convertible_to; + { T::kSelectColumns } -> std::convertible_to; { T::kTableName } -> std::convertible_to; + + // One selected column per Row element, so the two cannot drift apart when columns are added or + // removed (a reorder of same-typed columns is still not detectable). + requires columnCount(T::kSelectColumns) == std::tuple_size_v; }; } // namespace migration::cassandra::impl diff --git a/src/migration/cassandra/impl/TransactionsAdapter.cpp b/src/migration/cassandra/impl/TransactionsAdapter.cpp index b1c69c1f92..c2eb094a55 100644 --- a/src/migration/cassandra/impl/TransactionsAdapter.cpp +++ b/src/migration/cassandra/impl/TransactionsAdapter.cpp @@ -1,9 +1,15 @@ #include "migration/cassandra/impl/TransactionsAdapter.hpp" +#include "util/log/Logger.hpp" + +#include #include #include #include +#include +#include + namespace migration::cassandra::impl { void @@ -11,9 +17,22 @@ TransactionsAdapter::onRowRead(TableTransactionsDesc::Row const& row) { auto const& [txHash, date, ledgerSeq, metaBlob, txBlob] = row; - xrpl::SerialIter it{txBlob.data(), txBlob.size()}; - xrpl::STTx const sttx{it}; - xrpl::TxMeta const txMeta{sttx.getTransactionID(), ledgerSeq, metaBlob}; - onTransactionRead_(sttx, txMeta); + // Catch only deserialization errors (xrpl throws std::runtime_error) and report them via + // onDecodeFailure_; the owner decides the policy. Other exceptions (e.g. std::bad_alloc) + // propagate to fail the scan closed in migrateInTokenRange. + std::optional sttx; + std::optional txMeta; + try { + xrpl::SerialIter it{txBlob.data(), txBlob.size()}; + sttx.emplace(it); + txMeta.emplace(sttx->getTransactionID(), ledgerSeq, metaBlob); + } catch (std::runtime_error const& e) { + LOG(log_.error()) << "Failed to deserialize transaction: hash " << xrpl::strHex(txHash) + << ", ledger " << ledgerSeq << ": " << e.what(); + onDecodeFailure_(); + return; + } + + onTransactionRead_(*sttx, *txMeta); } } // namespace migration::cassandra::impl diff --git a/src/migration/cassandra/impl/TransactionsAdapter.hpp b/src/migration/cassandra/impl/TransactionsAdapter.hpp index bbb0b5a546..f75f0a32e0 100644 --- a/src/migration/cassandra/impl/TransactionsAdapter.hpp +++ b/src/migration/cassandra/impl/TransactionsAdapter.hpp @@ -2,6 +2,7 @@ #include "migration/cassandra/CassandraMigrationBackend.hpp" #include "migration/cassandra/impl/FullTableScannerAdapterBase.hpp" +#include "util/log/Logger.hpp" #include #include @@ -21,32 +22,39 @@ namespace migration::cassandra::impl { * @brief The description of the transactions table. It has to be a TableSpec. */ struct TableTransactionsDesc { - // hash, date, ledger_seq, metadata, transaction + // Must match kSelectColumns order. using Row = std::tuple; static constexpr char const* kPartitionKey = "hash"; + static constexpr char const* kSelectColumns = + "hash, date, ledger_sequence, metadata, transaction"; static constexpr char const* kTableName = "transactions"; }; /** * @brief The adapter for the transactions table. This class is responsible for reading the - * transactions from the FullTableScanner and converting the blobs to the STTx and TxMeta. + * transactions from the FullTableScanner and deserializing them into STTx and TxMeta. */ class TransactionsAdapter : public impl::FullTableScannerAdapterBase { public: - using OnTransactionRead = std::function; + using OnTransactionRead = std::function; + using OnDecodeFailure = std::function; /** * @brief Construct a new Transactions Adapter object * * @param backend The backend * @param onTxRead The callback to call when a transaction is read + * @param onDecodeFailure The callback to call when a transaction fails to deserialize; the + * owner decides the policy (count, abort, etc.). Invoked concurrently across scan workers. */ explicit TransactionsAdapter( std::shared_ptr backend, - OnTransactionRead onTxRead + OnTransactionRead onTxRead, + OnDecodeFailure onDecodeFailure ) : FullTableScannerAdapterBase(backend) , onTransactionRead_{std::move(onTxRead)} + , onDecodeFailure_{std::move(onDecodeFailure)} { } @@ -59,7 +67,9 @@ class TransactionsAdapter : public impl::FullTableScannerAdapterBase #include #include +#include +#include +#include +#include namespace util { @@ -36,4 +40,78 @@ forEachBatch(std::ranges::forward_range auto&& container, std::size_t batchSize, } } +/** + * @brief Accumulates a stream of records and flushes them to a sink in batches. + * + * Where @ref forEachBatch splits an already-complete container, this buffers records arriving over + * many @ref add calls and invokes the sink once a full batch has accumulated, coalescing many small + * inputs into fewer, larger writes. Any partial final batch is flushed by @ref flush. It is + * thread-safe: @ref add may be called concurrently, and the sink is invoked outside the internal + * lock so batch writes can proceed in parallel with further accumulation. A single @ref add may + * push the buffer past the batch size, in which case the whole buffer is flushed as one batch. + * + * @tparam T The record type being batched. + */ +template +class BatchBuffer { + std::mutex mutex_; + std::vector buffer_; + std::size_t batchSize_; + std::function const&)> sink_; + +public: + /** + * @brief Construct a batch buffer. + * + * @param batchSize The buffered-record count that triggers a flush; must be greater than zero. + * @param sink The callback invoked with each full batch, and with the final partial batch on + * @ref flush. + */ + BatchBuffer(std::size_t batchSize, std::function const&)> sink) + : batchSize_{batchSize}, sink_{std::move(sink)} + { + ASSERT(batchSize_ > 0, "Batch size must be greater than 0"); + } + + /** + * @brief Append records, flushing a full batch to the sink once the threshold is reached. + * + * @param records The records to append; moved into the buffer. + */ + void + add(std::vector records) + { + std::vector batch; + { + std::scoped_lock const lock{mutex_}; + buffer_.insert( + buffer_.end(), + std::make_move_iterator(records.begin()), + std::make_move_iterator(records.end()) + ); + if (buffer_.size() < batchSize_) + return; + batch = std::exchange(buffer_, {}); + } + sink_(batch); + } + + /** + * @brief Flush any remaining buffered records to the sink. + * + * Does not invoke the sink when the buffer is empty. + */ + void + flush() + { + std::vector batch; + { + std::scoped_lock const lock{mutex_}; + batch = std::exchange(buffer_, {}); + } + if (not batch.empty()) + sink_(batch); + } +}; + } // namespace util diff --git a/tests/common/migration/TestMigrators.hpp b/tests/common/migration/TestMigrators.hpp index 290d534e33..f425ef239b 100644 --- a/tests/common/migration/TestMigrators.hpp +++ b/tests/common/migration/TestMigrators.hpp @@ -4,6 +4,7 @@ #include "util/config/ObjectView.hpp" #include +#include struct SimpleTestMigrator { using Backend = MockMigrationBackend; @@ -27,6 +28,18 @@ struct SimpleTestMigrator2 { } }; +struct ThrowingTestMigrator { + using Backend = MockMigrationBackend; + static constexpr auto kName = "ThrowingTestMigrator"; + static constexpr auto kDescription = "A migrator whose run fails"; + + static void + runMigration(std::shared_ptr, util::config::ObjectView const&) + { + throw std::runtime_error("migration failed"); + } +}; + struct SimpleTestMigrator3 { using Backend = MockMigrationBackend; static constexpr auto kName = "SimpleTestMigrator3"; diff --git a/tests/common/util/MockMigrationManager.hpp b/tests/common/util/MockMigrationManager.hpp new file mode 100644 index 0000000000..a44ac2b5c7 --- /dev/null +++ b/tests/common/util/MockMigrationManager.hpp @@ -0,0 +1,29 @@ +#pragma once + +#include "migration/MigrationManagerInterface.hpp" +#include "migration/MigratiorStatus.hpp" + +#include + +#include +#include +#include + +struct MockMigrationManager : public migration::MigrationManagerInterface { + MOCK_METHOD( + (std::vector>), + allMigratorsStatusPairs, + (), + (const, override) + ); + MOCK_METHOD(std::vector, allMigratorsNames, (), (const, override)); + MOCK_METHOD( + migration::MigratorStatus, + getMigratorStatusByName, + (std::string const&), + (const, override) + ); + MOCK_METHOD(std::string, getMigratorDescriptionByName, (std::string const&), (const, override)); + MOCK_METHOD(bool, isBlockingClio, (), (const, override)); + MOCK_METHOD(void, runMigration, (std::string const&), (override)); +}; diff --git a/tests/integration/CMakeLists.txt b/tests/integration/CMakeLists.txt index 03b72f750d..648b83b22a 100644 --- a/tests/integration/CMakeLists.txt +++ b/tests/integration/CMakeLists.txt @@ -8,6 +8,7 @@ target_sources( data/cassandra/BaseTests.cpp migration/cassandra/DBRawData.cpp migration/cassandra/CassandraMigrationManagerTests.cpp + migration/cassandra/MPTTransactionHistoryMigratorTests.cpp migration/cassandra/ExampleTransactionsMigrator.cpp migration/cassandra/ExampleObjectsMigrator.cpp migration/cassandra/ExampleLedgerMigrator.cpp diff --git a/tests/integration/data/cassandra/BaseTests.cpp b/tests/integration/data/cassandra/BaseTests.cpp index af0b6a419b..601d6cf082 100644 --- a/tests/integration/data/cassandra/BaseTests.cpp +++ b/tests/integration/data/cassandra/BaseTests.cpp @@ -343,6 +343,53 @@ TEST_F(BackendCassandraBaseTest, BatchInsert) dropKeyspace(handle, "test"); } +TEST_F(BackendCassandraBaseTest, StatementPagingStateReadsAllPages) +{ + auto handle = createHandle(TestGlobals::instance().backendHost, "test_paging_state"); + auto const create = fmt::format( + R"( + CREATE TABLE IF NOT EXISTS strings (hash blob PRIMARY KEY, sequence bigint) + WITH default_time_to_live = {} + )", + 5000 + ); + ASSERT_TRUE(handle.execute(create)); + + auto const insert = handle.prepare("INSERT INTO strings (hash, sequence) VALUES (?, ?)"); + std::vector statements; + statements.push_back(insert.bind(std::string{"first"}, 1)); + statements.push_back(insert.bind(std::string{"second"}, 2)); + statements.push_back(insert.bind(std::string{"third"}, 3)); + ASSERT_TRUE(handle.execute(statements)); + + auto select = handle.prepare("SELECT hash, sequence FROM strings").bind(); + select.setPagingSize(1); + + std::vector hashes; + bool sawMorePages = false; + for (;;) { + auto const res = handle.execute(select); + ASSERT_TRUE(res) << res.error(); + auto const& results = res.value(); + + for (auto const& [hash, seq] : extract(results)) { + hashes.push_back(hash); + EXPECT_GT(seq, 0); + } + + if (not results.hasMorePages()) + break; + + sawMorePages = true; + select.setPagingState(results); + } + + EXPECT_TRUE(sawMorePages); + std::ranges::sort(hashes); + EXPECT_EQ(hashes, (std::vector{"first", "second", "third"})); + dropKeyspace(handle, "test_paging_state"); +} + TEST_F(BackendCassandraBaseTest, BatchInsertAsync) { using std::to_string; diff --git a/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp b/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp index 9aa8ad5f3f..a090a9e749 100644 --- a/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp +++ b/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp @@ -2,7 +2,9 @@ #include "data/DBHelpers.hpp" #include "data/LedgerCache.hpp" #include "data/cassandra/Handle.hpp" +#include "data/cassandra/Schema.hpp" #include "data/cassandra/SettingsProvider.hpp" +#include "data/cassandra/Types.hpp" #include "migration/MigrationManagerInterface.hpp" #include "migration/MigratiorStatus.hpp" #include "migration/cassandra/CassandraMigrationTestBackend.hpp" @@ -21,16 +23,22 @@ #include "util/config/Types.hpp" #include +#include #include #include #include +#include #include +#include #include #include +#include #include #include +#include #include +#include using namespace util; using namespace std; @@ -52,6 +60,13 @@ using CassandraMigrationTestManager = migration::impl::MigrationManagerBase; namespace { +struct FullScanPagingTableDesc { + using Row = std::tuple; + static constexpr char const* kPartitionKey = "id"; + static constexpr char const* kSelectColumns = "id, value"; + static constexpr char const* kTableName = "full_scan_paging_test"; +}; + std::pair< std::shared_ptr, std::shared_ptr> @@ -197,6 +212,56 @@ TEST_F(MigrationCassandraManagerCleanDBTest, MigratorStatus) EXPECT_EQ(status, MigratorStatus::Status::NotKnown); } +TEST_F(MigrationCassandraManagerCleanDBTest, MigrateInTokenRangeReadsAllDriverPages) +{ + // Exceeds CassandraMigrationBackend's internal full-scan page size, so this fails if the + // migration scan reads only the first driver page. + constexpr std::int64_t kRowsBeyondFullScanPage = 5'001; + + auto const dbCfg = cfg_.getObject("database.cassandra"); + SettingsProvider const provider{dbCfg}; + Handle const handle{TestGlobals::instance().backendHost}; + ASSERT_TRUE(handle.connect()); + + auto const tableName = + data::cassandra::qualifiedTableName(provider, FullScanPagingTableDesc::kTableName); + ASSERT_TRUE(handle.execute( + fmt::format( + R"( + CREATE TABLE IF NOT EXISTS {} + (id bigint PRIMARY KEY, value bigint) + )", + tableName + ) + )); + + auto const insert = + handle.prepare(fmt::format("INSERT INTO {} (id, value) VALUES (?, ?)", tableName)); + std::vector statements; + statements.reserve(kRowsBeyondFullScanPage); + for (std::int64_t id = 0; id < kRowsBeyondFullScanPage; ++id) + statements.push_back(insert.bind(id, id * 2)); + ASSERT_TRUE(handle.executeEach(statements)); + + std::set seenIds; + data::synchronous([&](auto yield) { + testMigrationBackend_->migrateInTokenRange( + std::numeric_limits::min(), + std::numeric_limits::max(), + [&](FullScanPagingTableDesc::Row const& row) { + auto const& [id, value] = row; + EXPECT_EQ(value, id * 2); + seenIds.insert(id); + }, + yield + ); + }); + + EXPECT_EQ(seenIds.size(), static_cast(kRowsBeyondFullScanPage)); + EXPECT_TRUE(seenIds.contains(0)); + EXPECT_TRUE(seenIds.contains(kRowsBeyondFullScanPage - 1)); +} + // The test suite for testing migration process for ExampleTransactionsMigrator. In this test suite, // the transactions are written to the database before running the migration class MigrationCassandraManagerTxTableTest : public MigrationCassandraSimpleTest { diff --git a/tests/integration/migration/cassandra/ExampleObjectsMigrator.cpp b/tests/integration/migration/cassandra/ExampleObjectsMigrator.cpp index d568d75ce6..35f498fbd6 100644 --- a/tests/integration/migration/cassandra/ExampleObjectsMigrator.cpp +++ b/tests/integration/migration/cassandra/ExampleObjectsMigrator.cpp @@ -46,5 +46,5 @@ ExampleObjectsMigrator::runMigration( } ) ); - scanner.wait(); + scanner.waitForAllAndThrowOnError(); } diff --git a/tests/integration/migration/cassandra/ExampleTransactionsMigrator.cpp b/tests/integration/migration/cassandra/ExampleTransactionsMigrator.cpp index 7401ae5ee4..50d044e93b 100644 --- a/tests/integration/migration/cassandra/ExampleTransactionsMigrator.cpp +++ b/tests/integration/migration/cassandra/ExampleTransactionsMigrator.cpp @@ -35,14 +35,16 @@ ExampleTransactionsMigrator::runMigration( .jobsNum = jobsFullScan, .cursorsPerJob = cursorPerJobsFullScan}, migration::cassandra::impl::TransactionsAdapter( - backend, [&](xrpl::STTx const& tx, xrpl::TxMeta const&) { + backend, + [&](xrpl::STTx const& tx, xrpl::TxMeta const&) { hashSet.lock()->insert(xrpl::to_string(tx.getTransactionID())); auto const json = tx.getJson(xrpl::JsonOptions::Values::None); auto const txType = json["TransactionType"].asString(); backend->writeTxIndexExample(uint256ToString(tx.getTransactionID()), txType); - } + }, + [] {} ) ); - scanner.wait(); + scanner.waitForAllAndThrowOnError(); count = hashSet.lock()->size(); } diff --git a/tests/integration/migration/cassandra/MPTTransactionHistoryMigratorTests.cpp b/tests/integration/migration/cassandra/MPTTransactionHistoryMigratorTests.cpp new file mode 100644 index 0000000000..714e358da6 --- /dev/null +++ b/tests/integration/migration/cassandra/MPTTransactionHistoryMigratorTests.cpp @@ -0,0 +1,583 @@ +#include "data/DBHelpers.hpp" +#include "data/LedgerCache.hpp" +#include "data/Types.hpp" +#include "data/cassandra/Handle.hpp" +#include "data/cassandra/Schema.hpp" +#include "data/cassandra/SettingsProvider.hpp" +#include "etl/MPTHelpers.hpp" +#include "etl/Models.hpp" +#include "etl/impl/ext/MPT.hpp" +#include "migration/MigrationManagerInterface.hpp" +#include "migration/MigratiorStatus.hpp" +#include "migration/cassandra/CassandraMigrationBackend.hpp" +#include "migration/cassandra/MPTTransactionHistoryMigrator.hpp" +#include "migration/impl/MigrationManagerBase.hpp" +#include "migration/impl/MigratorsRegister.hpp" +#include "util/LedgerUtils.hpp" +#include "util/MockPrometheus.hpp" +#include "util/StringUtils.hpp" +#include "util/TestObject.hpp" +#include "util/config/ConfigConstraints.hpp" +#include "util/config/ConfigDefinition.hpp" +#include "util/config/ConfigValue.hpp" +#include "util/config/Types.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace util; +using namespace data::cassandra; +using namespace migration; +using namespace util::config; + +// Wire the production MPTTransactionHistoryMigrator against the production +// CassandraMigrationBackend (the test backend's Backend type would not satisfy the migrator's +// Backend alias). +using SupportedMigrators = migration::impl::MigratorsRegister< + migration::cassandra::CassandraMigrationBackend, + migration::cassandra::MPTTransactionHistoryMigrator>; +using TestManager = migration::impl::MigrationManagerBase; + +namespace { + +constexpr auto kMigratorName = migration::cassandra::MPTTransactionHistoryMigrator::kName; + +// Accounts shared across fixtures (valid base58 r-addresses). +constexpr auto kIssuer = "rM2AGCCCRb373FRuD8wHyUwUsh2dV4BW5Q"; +constexpr auto kHolder = "rK1EX542EgA9m948JrJRaEzwLVEhqWvnr9"; +constexpr auto kHolder2 = "rnd1nHuzceyQDqnLH8urWNr4QBKt4v7WVk"; +constexpr auto kObserver = "rf1BiGeXwwQoi8Z2ueFYTEXSwuJYfV2Jpn"; +constexpr auto kUnknownAccount = "rLEsXccBGNR3UPuPu2hUXPjziKC3qKSBun"; + +constexpr std::uint32_t kLedgerSeq = 100; +constexpr std::uint32_t kIssuanceSeq = 7; + +// Minimal ledger header (seq is overwritten before writing) used to seed a valid ledger range. +constexpr auto kRawHeader = + "03C3141A01633CD656F91B4EBB5EB89B791BD34DBC8A04BB6F407C5335BC54351E" + "DD733898497E809E04074D14D271E4832D7888754F9230800761563A292FA2315A" + "6DB6FE30CC5909B285080FCD6773CC883F9FE0EE4D439340AC592AADB973ED3CF5" + "3E2232B33EF57CECAC2816E3122816E31A0A00F8377CD95DFA484CFAE282656A58" + "CE5AA29652EFFD80AC59CD91416E4E13DBBE"; + +// An MPToken holder node carrying the issuance ID directly (the sfMPTokenIssuanceID path). +xrpl::STObject +createMPTokenNode(xrpl::uint192 const& issuanceID, std::string_view holder) +{ + xrpl::STObject fields(xrpl::sfFinalFields); + fields.setAccountID(xrpl::sfAccount, getAccountIdWithString(holder)); + fields[xrpl::sfMPTokenIssuanceID] = issuanceID; + + xrpl::STObject node(xrpl::sfModifiedNode); + node.setFieldU16(xrpl::sfLedgerEntryType, xrpl::ltMPTOKEN); + node.setFieldH256(xrpl::sfLedgerIndex, xrpl::uint256{}); + node.set(std::move(fields)); + return node; +} + +// An MPTokenIssuance node whose ID must be reconstructed from sfSequence + sfIssuer. +xrpl::STObject +createMPTokenIssuanceNode(std::uint32_t seq, std::string_view issuer) +{ + xrpl::STObject fields(xrpl::sfFinalFields); + fields.setFieldU32(xrpl::sfSequence, seq); + fields.setAccountID(xrpl::sfIssuer, getAccountIdWithString(issuer)); + + xrpl::STObject node(xrpl::sfModifiedNode); + node.setFieldU16(xrpl::sfLedgerEntryType, xrpl::ltMPTOKEN_ISSUANCE); + node.setFieldH256(xrpl::sfLedgerIndex, xrpl::uint256{}); + node.set(std::move(fields)); + return node; +} + +// A single Payment whose metadata touches two distinct issuances and three affected accounts, +// exercising the multi-issuance fan-out and per-account indexing. +std::pair +makeMultiIssuancePayment( + std::uint32_t ledgerSeq, + std::uint32_t txIndex, + std::uint32_t txSequence = 1 +) +{ + xrpl::Slice const signingKey("test", 4); + xrpl::STObject tx(xrpl::sfTransaction); + tx.setFieldU16(xrpl::sfTransactionType, xrpl::ttPAYMENT); + tx.setAccountID(xrpl::sfAccount, getAccountIdWithString(kHolder)); + tx.setFieldAmount(xrpl::sfAmount, xrpl::STAmount(100, false)); + tx.setFieldAmount(xrpl::sfFee, xrpl::STAmount(10, false)); + tx.setAccountID(xrpl::sfDestination, getAccountIdWithString(kHolder2)); + tx.setFieldU32(xrpl::sfSequence, txSequence); + tx.setFieldVL(xrpl::sfSigningPubKey, signingKey); + + auto const serialized = tx.getSerializer(); + xrpl::STTx const sttx{xrpl::SerialIter{serialized.slice()}}; + + auto const issuanceA = xrpl::makeMptID(1, getAccountIdWithString(kHolder)); + auto const issuanceB = xrpl::makeMptID(2, getAccountIdWithString(kHolder)); + + xrpl::STObject metaObj(xrpl::sfTransactionMetaData); + metaObj.setFieldU8(xrpl::sfTransactionResult, xrpl::tesSUCCESS); + metaObj.setFieldU32(xrpl::sfTransactionIndex, txIndex); + + xrpl::STArray affectedNodes(xrpl::sfAffectedNodes); + affectedNodes.push_back(createMPTokenNode(issuanceA, kHolder)); + affectedNodes.push_back(createMPTokenNode(issuanceB, kHolder2)); + affectedNodes.push_back(createMPTokenNode(issuanceA, kObserver)); + affectedNodes.push_back(createMPTokenIssuanceNode(1, kHolder)); // resolves to issuanceA + metaObj.setFieldArray(xrpl::sfAffectedNodes, affectedNodes); + + xrpl::TxMeta const txMeta{ + sttx.getTransactionID(), ledgerSeq, metaObj.getSerializer().peekData() + }; + return {sttx, txMeta}; +} + +std::string +blobToString(xrpl::Blob const& blob) +{ + return {reinterpret_cast(blob.data()), blob.size()}; +} + +// A failed transaction with a top-level sfMPTokenIssuanceID exercises parity for records derived +// from the transaction body rather than successful metadata nodes. +std::pair +makeFailedExplicitMPTReferenceTx( + xrpl::uint192 const& issuanceID, + std::uint32_t ledgerSeq, + std::uint32_t txIndex +) +{ + xrpl::Slice const signingKey("test", 4); + xrpl::STObject tx(xrpl::sfTransaction); + tx.setFieldU16(xrpl::sfTransactionType, xrpl::ttMPTOKEN_ISSUANCE_SET); + tx.setAccountID(xrpl::sfAccount, getAccountIdWithString(kIssuer)); + tx[xrpl::sfMPTokenIssuanceID] = issuanceID; + tx.setFieldAmount(xrpl::sfFee, xrpl::STAmount(10, false)); + tx.setFieldU32(xrpl::sfSequence, 2); + tx.setFieldVL(xrpl::sfSigningPubKey, signingKey); + + auto const serialized = tx.getSerializer(); + xrpl::STTx const sttx{xrpl::SerialIter{serialized.slice()}}; + + xrpl::STObject metaObj(xrpl::sfTransactionMetaData); + metaObj.setFieldU8(xrpl::sfTransactionResult, xrpl::tecINCOMPLETE); + metaObj.setFieldU32(xrpl::sfTransactionIndex, txIndex); + metaObj.setFieldArray(xrpl::sfAffectedNodes, xrpl::STArray{xrpl::sfAffectedNodes}); + + xrpl::TxMeta const txMeta{ + sttx.getTransactionID(), ledgerSeq, metaObj.getSerializer().peekData() + }; + return {sttx, txMeta}; +} + +template +std::string +uintToString(UInt const& value) +{ + return {reinterpret_cast(value.data()), value.size()}; +} + +} // namespace + +class MPTTransactionHistoryMigratorTest : public util::prometheus::WithPrometheus { +public: + MPTTransactionHistoryMigratorTest() + { + auto const cfg = cfg_.getObject("database.cassandra"); + backend_ = std::make_shared( + SettingsProvider{cfg}, cache_ + ); + manager_ = std::make_shared(backend_, cfg_.getObject("migration")); + } + + ~MPTTransactionHistoryMigratorTest() override + { + Handle const handle{TestGlobals::instance().backendHost}; + EXPECT_TRUE(handle.connect()); + EXPECT_TRUE(handle.execute("DROP KEYSPACE " + TestGlobals::instance().backendKeyspace)); + } + +protected: + static constexpr auto kCassandra = "cassandra"; + + ClioConfigDefinition cfg_{ + {{"database.type", ConfigValue{ConfigType::String}.defaultValue(kCassandra)}, + {"database.cassandra.contact_points", + ConfigValue{ConfigType::String}.defaultValue(TestGlobals::instance().backendHost)}, + {"database.cassandra.keyspace", + ConfigValue{ConfigType::String}.defaultValue(TestGlobals::instance().backendKeyspace)}, + {"database.cassandra.provider", ConfigValue{ConfigType::String}.defaultValue(kCassandra)}, + {"database.cassandra.replication_factor", + ConfigValue{ConfigType::Integer}.defaultValue(1)}, + {"database.cassandra.connect_timeout", ConfigValue{ConfigType::Integer}.defaultValue(2)}, + {"database.cassandra.secure_connect_bundle", ConfigValue{ConfigType::String}.optional()}, + {"database.cassandra.port", + ConfigValue{ConfigType::Integer}.withConstraint(gValidatePort).optional()}, + {"database.cassandra.table_prefix", ConfigValue{ConfigType::String}.optional()}, + {"database.cassandra.max_write_requests_outstanding", + ConfigValue{ConfigType::Integer}.defaultValue(10'000).withConstraint(gValidateUint32)}, + {"database.cassandra.max_read_requests_outstanding", + ConfigValue{ConfigType::Integer}.defaultValue(100'000).withConstraint(gValidateUint32)}, + {"database.cassandra.threads", + ConfigValue{ConfigType::Integer} + .defaultValue(static_cast(std::thread::hardware_concurrency())) + .withConstraint(gValidateUint32)}, + {"database.cassandra.core_connections_per_host", + ConfigValue{ConfigType::Integer}.defaultValue(1).withConstraint(gValidateUint16)}, + {"database.cassandra.queue_size_io", + ConfigValue{ConfigType::Integer}.optional().withConstraint(gValidateUint16)}, + {"database.cassandra.write_batch_size", + ConfigValue{ConfigType::Integer}.defaultValue(20).withConstraint(gValidateUint16)}, + {"database.cassandra.request_timeout", + ConfigValue{ConfigType::Integer}.optional().withConstraint(gValidateUint32)}, + {"database.cassandra.username", ConfigValue{ConfigType::String}.optional()}, + {"database.cassandra.password", ConfigValue{ConfigType::String}.optional()}, + {"database.cassandra.certfile", ConfigValue{ConfigType::String}.optional()}, + {"migration.full_scan_threads", + ConfigValue{ConfigType::Integer}.defaultValue(2).withConstraint(gValidateUint32)}, + {"migration.full_scan_jobs", + ConfigValue{ConfigType::Integer}.defaultValue(4).withConstraint(gValidateUint32)}, + {"migration.cursors_per_job", + ConfigValue{ConfigType::Integer}.defaultValue(100).withConstraint(gValidateUint32)}} + }; + + data::LedgerCache cache_; + std::shared_ptr backend_; + std::shared_ptr manager_; + + using SeqIdx = std::tuple; + using IssuanceRow = std::tuple; + using AccountRow = std::tuple; + + struct RawIndexRows { + std::set issuance; + std::set account; + + bool + operator==(RawIndexRows const&) const = default; + }; + + // Seed a single-ledger range so index fetches resolve a valid sequence range. + void + setupLedgerRange(std::uint32_t seq) + { + std::string rawHeaderBlob = hexStringToBinaryString(kRawHeader); + xrpl::LedgerHeader lgrInfo = util::deserializeHeader(xrpl::makeSlice(rawHeaderBlob)); + lgrInfo.seq = seq; + backend_->writeLedger(lgrInfo, std::move(rawHeaderBlob)); + backend_->writeSuccessor( + uint256ToString(data::kFirstKey), lgrInfo.seq, uint256ToString(data::kLastKey) + ); + ASSERT_TRUE(backend_->finishWrites(lgrInfo.seq)); + } + + // Write a (sttx, txMeta) pair into the transactions table at the given ledger sequence. + void + seedTransaction(xrpl::STTx const& sttx, xrpl::TxMeta const& txMeta, std::uint32_t seq) + { + backend_->writeTransaction( + uint256ToString(sttx.getTransactionID()), + seq, + 0, + blobToString(sttx.getSerializer().peekData()), + blobToString(txMeta.getAsObject().getSerializer().peekData()) + ); + } + + static etl::model::Transaction + makeModelTx(xrpl::STTx const& sttx, xrpl::TxMeta const& txMeta) + { + auto raw = blobToString(sttx.getSerializer().peekData()); + auto metaRaw = blobToString(txMeta.getAsObject().getSerializer().peekData()); + auto const txID = sttx.getTransactionID(); + + return etl::model::Transaction{ + .raw = std::move(raw), + .metaRaw = std::move(metaRaw), + .sttx = sttx, + .meta = txMeta, + .id = txID, + .key = uintToString(txID), + .type = sttx.getTxnType() + }; + } + + static RawIndexRows + expectedRowsFrom(std::vector const& expected) + { + RawIndexRows rows; + for (auto const& rec : expected) { + auto const issuanceID = uintToString(rec.mptIssuanceID); + auto const txHash = uintToString(rec.txHash); + auto const seqIdx = SeqIdx{rec.ledgerSequence, rec.transactionIndex}; + rows.issuance.emplace(issuanceID, seqIdx, txHash); + + for (auto const& account : rec.accounts) + rows.account.emplace(issuanceID, uintToString(account), seqIdx, txHash); + } + + return rows; + } + + static void + appendExpected(RawIndexRows& target, std::vector const& data) + { + auto rows = expectedRowsFrom(data); + target.issuance.insert(rows.issuance.begin(), rows.issuance.end()); + target.account.insert(rows.account.begin(), rows.account.end()); + } + + RawIndexRows + readRawIndexRows() + { + RawIndexRows rows; + auto const cfg = cfg_.getObject("database.cassandra"); + auto const settings = SettingsProvider{cfg}; + Handle const handle{TestGlobals::instance().backendHost}; + auto const connected = handle.connect(); + EXPECT_TRUE(connected); + if (not connected) + return rows; + + auto const issuanceRes = handle.execute( + "SELECT mptoken_issuance_id, seq_idx, hash FROM " + + qualifiedTableName(settings, "mptoken_issuance_transactions") + ); + EXPECT_TRUE(issuanceRes); + if (issuanceRes) { + for (auto const& [issuanceID, seqIdx, txHash] : + extract, SeqIdx, xrpl::uint256>(*issuanceRes)) { + rows.issuance.emplace(blobToString(issuanceID), seqIdx, uintToString(txHash)); + } + } + + auto const accountRes = handle.execute( + "SELECT mptoken_issuance_id, account, seq_idx, hash FROM " + + qualifiedTableName(settings, "account_mptoken_issuance_transactions") + ); + EXPECT_TRUE(accountRes); + if (accountRes) { + for (auto const& [issuanceID, account, seqIdx, txHash] : + extract, xrpl::AccountID, SeqIdx, xrpl::uint256>( + *accountRes + )) { + rows.account.emplace( + blobToString(issuanceID), uintToString(account), seqIdx, uintToString(txHash) + ); + } + } + + return rows; + } + + void + truncateIndexTables() + { + auto const cfg = cfg_.getObject("database.cassandra"); + auto const settings = SettingsProvider{cfg}; + Handle const handle{TestGlobals::instance().backendHost}; + ASSERT_TRUE(handle.connect()); + ASSERT_TRUE(handle.execute( + "TRUNCATE " + qualifiedTableName(settings, "mptoken_issuance_transactions") + )); + ASSERT_TRUE(handle.execute( + "TRUNCATE " + qualifiedTableName(settings, "account_mptoken_issuance_transactions") + )); + } + + void + expectRawRowsEqual(RawIndexRows const& expected) + { + auto const actual = readRawIndexRows(); + EXPECT_EQ(actual.issuance, expected.issuance); + EXPECT_EQ(actual.account, expected.account); + } +}; + +// Status is NotMigrated until the migrator runs. +TEST_F(MPTTransactionHistoryMigratorTest, StatusNotMigratedBeforeRun) +{ + EXPECT_EQ( + manager_->getMigratorStatusByName(kMigratorName), MigratorStatus::Status::NotMigrated + ); +} + +// Backfill indexes the issuance-node path (MPTokenIssuanceCreate) and the multi-issuance fan-out +// (a Payment touching two issuances and three accounts). Asserts the index tables match exactly +// what the shared extractor produces, then that status flips to Migrated. +TEST_F(MPTTransactionHistoryMigratorTest, BackfillIndexesAllShapes) +{ + setupLedgerRange(kLedgerSeq); + + auto const createData = createMPTIssuanceCreateTxWithMetadata(kIssuer, 2, kIssuanceSeq); + xrpl::STTx const createTx{ + xrpl::SerialIter{createData.transaction.data(), createData.transaction.size()} + }; + xrpl::TxMeta const createMeta{createTx.getTransactionID(), kLedgerSeq, createData.metadata}; + seedTransaction(createTx, createMeta, kLedgerSeq); + + auto const [payTx, payMeta] = makeMultiIssuancePayment(kLedgerSeq, 1); + seedTransaction(payTx, payMeta, kLedgerSeq); + + backend_->waitForWritesToFinish(); + + auto const expectedCreate = etl::getMPTokenIssuanceTxsFromTx(createMeta, createTx); + auto const expectedPay = etl::getMPTokenIssuanceTxsFromTx(payMeta, payTx); + ASSERT_EQ(expectedCreate.size(), 1u); + ASSERT_EQ(expectedPay.size(), 2u); + for (auto const& rec : expectedPay) { + ASSERT_EQ(rec.accounts.size(), 3u); + EXPECT_TRUE(rec.accounts.contains(getAccountIdWithString(kHolder))); + EXPECT_TRUE(rec.accounts.contains(getAccountIdWithString(kHolder2))); + EXPECT_TRUE(rec.accounts.contains(getAccountIdWithString(kObserver))); + } + + EXPECT_EQ( + manager_->getMigratorStatusByName(kMigratorName), MigratorStatus::Status::NotMigrated + ); + manager_->runMigration(kMigratorName); + EXPECT_EQ(manager_->getMigratorStatusByName(kMigratorName), MigratorStatus::Status::Migrated); + + RawIndexRows expected; + appendExpected(expected, expectedCreate); + appendExpected(expected, expectedPay); + expectRawRowsEqual(expected); + + // An account that never appears in any metadata has no rows. + auto actual = readRawIndexRows(); + for (auto const& row : actual.account) + EXPECT_NE(std::get<1>(row), uintToString(getAccountIdWithString(kUnknownAccount))); +} + +// Rerunning the migrator rewrites identical deterministic-key rows, so exact rows are unchanged. +TEST_F(MPTTransactionHistoryMigratorTest, RerunIsIdempotent) +{ + setupLedgerRange(kLedgerSeq); + + auto const [payTx, payMeta] = makeMultiIssuancePayment(kLedgerSeq, 1); + seedTransaction(payTx, payMeta, kLedgerSeq); + backend_->waitForWritesToFinish(); + + manager_->runMigration(kMigratorName); + auto const afterFirst = readRawIndexRows(); + EXPECT_FALSE(afterFirst.issuance.empty()); + EXPECT_FALSE(afterFirst.account.empty()); + + manager_->runMigration(kMigratorName); + EXPECT_EQ(readRawIndexRows(), afterFirst); + EXPECT_EQ(manager_->getMigratorStatusByName(kMigratorName), MigratorStatus::Status::Migrated); +} + +// An empty transactions table runs cleanly and still reports Migrated. +TEST_F(MPTTransactionHistoryMigratorTest, EmptyTableRunsCleanly) +{ + setupLedgerRange(kLedgerSeq); + + manager_->runMigration(kMigratorName); + EXPECT_EQ(manager_->getMigratorStatusByName(kMigratorName), MigratorStatus::Status::Migrated); + expectRawRowsEqual({}); +} + +// Backfill and live ETL produce identical rows for the same transactions, including a failed +// transaction whose only MPT reference is in the transaction body. +TEST_F(MPTTransactionHistoryMigratorTest, BackfillMatchesLiveETLRows) +{ + setupLedgerRange(kLedgerSeq); + + auto const [payTx, payMeta] = makeMultiIssuancePayment(kLedgerSeq, 1); + auto const [failedTx, failedMeta] = makeFailedExplicitMPTReferenceTx( + xrpl::makeMptID(kIssuanceSeq, getAccountIdWithString(kIssuer)), kLedgerSeq, 2 + ); + + seedTransaction(payTx, payMeta, kLedgerSeq); + seedTransaction(failedTx, failedMeta, kLedgerSeq); + backend_->waitForWritesToFinish(); + + auto const expectedFailed = + expectedRowsFrom(etl::getMPTokenIssuanceTxsFromTx(failedMeta, failedTx)); + ASSERT_EQ(expectedFailed.issuance.size(), 1u); + + etl::impl::MPTExt liveExt{backend_}; + liveExt.onLedgerData( + etl::model::LedgerData{ + .transactions = {makeModelTx(payTx, payMeta), makeModelTx(failedTx, failedMeta)}, + .objects = {}, + .successors = {}, + .edgeKeys = {}, + .header = xrpl::LedgerHeader{}, + .rawHeader = {}, + .seq = kLedgerSeq + } + ); + backend_->waitForWritesToFinish(); + + auto const liveRows = readRawIndexRows(); + EXPECT_FALSE(liveRows.issuance.empty()); + EXPECT_FALSE(liveRows.account.empty()); + EXPECT_TRUE(liveRows.issuance.contains(*expectedFailed.issuance.begin())); + + truncateIndexTables(); + expectRawRowsEqual({}); + + manager_->runMigration(kMigratorName); + EXPECT_EQ(readRawIndexRows(), liveRows); +} + +TEST_F(MPTTransactionHistoryMigratorTest, BackfillReadsAllTransactionDriverPages) +{ + setupLedgerRange(kLedgerSeq); + + constexpr std::uint32_t kRowsBeyondFullScanPage = 5'001; + RawIndexRows expected; + + for (std::uint32_t i = 0; i < kRowsBeyondFullScanPage; ++i) { + auto const [payTx, payMeta] = makeMultiIssuancePayment(kLedgerSeq, i, i + 1); + seedTransaction(payTx, payMeta, kLedgerSeq); + appendExpected(expected, etl::getMPTokenIssuanceTxsFromTx(payMeta, payTx)); + } + backend_->waitForWritesToFinish(); + + EXPECT_EQ(expected.issuance.size(), kRowsBeyondFullScanPage * 2u); + EXPECT_EQ(expected.account.size(), kRowsBeyondFullScanPage * 6u); + + ClioConfigDefinition const singleRangeCfg{ + {{"migration.full_scan_threads", + ConfigValue{ConfigType::Integer}.defaultValue(1).withConstraint(gValidateUint32)}, + {"migration.full_scan_jobs", + ConfigValue{ConfigType::Integer}.defaultValue(1).withConstraint(gValidateUint32)}, + {"migration.cursors_per_job", + ConfigValue{ConfigType::Integer}.defaultValue(1).withConstraint(gValidateUint32)}} + }; + + migration::cassandra::MPTTransactionHistoryMigrator::runMigration( + backend_, singleRangeCfg.getObject("migration") + ); + + expectRawRowsEqual(expected); +} diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index a5a35a4066..75082980c8 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -84,6 +84,8 @@ target_sources( # Migration migration/cassandra/FullTableScannerTests.cpp migration/cassandra/SpecTests.cpp + migration/cassandra/TransactionsAdapterTests.cpp + migration/MigrationApplicationTests.cpp migration/MigratorRegisterTests.cpp migration/MigratorStatusTests.cpp migration/MigrationInspectorFactoryTests.cpp diff --git a/tests/unit/migration/MigrationApplicationTests.cpp b/tests/unit/migration/MigrationApplicationTests.cpp new file mode 100644 index 0000000000..18fab6a1cb --- /dev/null +++ b/tests/unit/migration/MigrationApplicationTests.cpp @@ -0,0 +1,112 @@ +#include "migration/MigrationApplication.hpp" +#include "migration/MigratiorStatus.hpp" +#include "util/MockMigrationManager.hpp" +#include "util/MockPrometheus.hpp" +#include "util/config/ConfigDefinition.hpp" +#include "util/config/ConfigValue.hpp" +#include "util/config/Types.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace app; +using namespace util::config; +using testing::Return; +using testing::StrictMock; + +namespace { + +constexpr auto kMigratorName = "TestMigrator"; + +} // namespace + +struct MigrationApplicationTest : util::prometheus::WithPrometheus { + std::shared_ptr> manager = + std::make_shared>(); + + [[nodiscard]] MigratorApplication + makeApp(MigrateSubCmd command) const + { + return MigratorApplication{std::move(command), manager}; + } +}; + +// A status command lists every registered migrator with its description and status. +TEST_F(MigrationApplicationTest, StatusListsAllMigrators) +{ + EXPECT_CALL(*manager, allMigratorsStatusPairs()) + .WillOnce(Return( + std::vector>{ + {kMigratorName, migration::MigratorStatus::Status::Migrated} + } + )); + EXPECT_CALL(*manager, getMigratorDescriptionByName(kMigratorName)) + .WillOnce(Return("A test migrator")); + + auto app = makeApp(MigrateSubCmd::status()); + EXPECT_EQ(app.run(), EXIT_SUCCESS); +} + +// A status command with no registered migrators still succeeds. +TEST_F(MigrationApplicationTest, StatusWithNoMigrators) +{ + EXPECT_CALL(*manager, allMigratorsStatusPairs()) + .WillOnce(Return(std::vector>{})); + + auto app = makeApp(MigrateSubCmd::status()); + EXPECT_EQ(app.run(), EXIT_SUCCESS); +} + +// Running an already-migrated migrator is a no-op that reports success and prints the status. +TEST_F(MigrationApplicationTest, MigrateAlreadyMigrated) +{ + EXPECT_CALL(*manager, getMigratorStatusByName(kMigratorName)) + .WillOnce(Return(migration::MigratorStatus::Status::Migrated)); + EXPECT_CALL(*manager, allMigratorsStatusPairs()) + .WillOnce(Return(std::vector>{})); + + auto app = makeApp(MigrateSubCmd::migration(kMigratorName)); + EXPECT_EQ(app.run(), EXIT_SUCCESS); +} + +// Running an unknown migrator reports failure and prints the status. +TEST_F(MigrationApplicationTest, MigrateUnknownMigrator) +{ + EXPECT_CALL(*manager, getMigratorStatusByName(kMigratorName)) + .WillOnce(Return(migration::MigratorStatus::Status::NotKnown)); + EXPECT_CALL(*manager, allMigratorsStatusPairs()) + .WillOnce(Return(std::vector>{})); + + auto app = makeApp(MigrateSubCmd::migration(kMigratorName)); + EXPECT_EQ(app.run(), EXIT_FAILURE); +} + +// Running a not-yet-migrated migrator triggers the migration and reports success. +TEST_F(MigrationApplicationTest, MigrateRunsMigration) +{ + EXPECT_CALL(*manager, getMigratorStatusByName(kMigratorName)) + .WillOnce(Return(migration::MigratorStatus::Status::NotMigrated)); + EXPECT_CALL(*manager, runMigration(kMigratorName)); + + auto app = makeApp(MigrateSubCmd::migration(kMigratorName)); + EXPECT_EQ(app.run(), EXIT_SUCCESS); +} + +// The config-based constructor throws when the migration manager cannot be built. An unsupported +// database type is rejected before any backend connection is attempted, so this stays a unit test. +TEST_F(MigrationApplicationTest, ConfigConstructorThrowsOnInvalidDatabaseType) +{ + ClioConfigDefinition const config{ + {{"database.type", ConfigValue{ConfigType::String}.defaultValue("not-a-real-db")}} + }; + + EXPECT_THROW(MigratorApplication(config, MigrateSubCmd::status()), std::runtime_error); +} diff --git a/tests/unit/migration/MigratorRegisterTests.cpp b/tests/unit/migration/MigratorRegisterTests.cpp index b520975506..71cb3c7039 100644 --- a/tests/unit/migration/MigratorRegisterTests.cpp +++ b/tests/unit/migration/MigratorRegisterTests.cpp @@ -14,6 +14,7 @@ #include #include +#include #include #include @@ -255,6 +256,32 @@ TEST_F(MultipleMigratorRegisterTests, MigrateNormalMigrator) ); } +using ThrowingMigratorRegister = + migration::impl::MigratorsRegister; + +struct ThrowingMigratorRegisterTests : public util::prometheus::WithMockPrometheus, + public MockMigrationBackendTest { + std::optional migratorRegister; + + ThrowingMigratorRegisterTests() + { + migratorRegister.emplace(backend_); + } +}; + +// A migrator that fails its run must propagate the error and leave status unwritten (NotMigrated), +// so a partial migration is never recorded as complete. +TEST_F(ThrowingMigratorRegisterTests, FailedMigrationDoesNotMarkMigrated) +{ + EXPECT_CALL(*backend_, writeMigratorStatus(testing::_, testing::_)).Times(0); + ASSERT_TRUE(migratorRegister.has_value()); + EXPECT_THROW( + // NOLINTNEXTLINE(bugprone-unchecked-optional-access) + migratorRegister->runMigrator("ThrowingTestMigrator", gCfg.getObject("migration")), + std::runtime_error + ); +} + TEST_F(MultipleMigratorRegisterTests, canBlock) { ASSERT_TRUE(migratorRegister.has_value()); diff --git a/tests/unit/migration/cassandra/FullTableScannerTests.cpp b/tests/unit/migration/cassandra/FullTableScannerTests.cpp index 1e54e4a77c..d13ccbb035 100644 --- a/tests/unit/migration/cassandra/FullTableScannerTests.cpp +++ b/tests/unit/migration/cassandra/FullTableScannerTests.cpp @@ -5,9 +5,14 @@ #include #include +#include +#include #include #include #include +#include +#include +#include namespace { @@ -34,6 +39,24 @@ struct TestScannerAdapter { callback.get().Call(range, yield); } }; + +struct FailingAndSlowAdapter { + std::reference_wrapper calls; + std::reference_wrapper slowWorkerFinished; + + void + readByTokenRange( + migration::cassandra::impl::TokenRange const&, + boost::asio::yield_context + ) const + { + if (calls.get().fetch_add(1) == 0u) + throw std::runtime_error("scan failure"); + + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + slowWorkerFinished.get() = true; + } +}; } // namespace struct FullTableScannerAssertTest : common::util::WithMockAssert {}; @@ -76,7 +99,7 @@ TEST_F(FullTableScannerTests, SingleThreadCtx) auto scanner = migration::cassandra::impl::FullTableScanner( {.ctxThreadsNum = 1, .jobsNum = 1, .cursorsPerJob = 100}, TestScannerAdapter(mockCallback) ); - scanner.wait(); + scanner.waitForAllAndThrowOnError(); } TEST_F(FullTableScannerTests, MultipleThreadCtx) @@ -88,7 +111,7 @@ TEST_F(FullTableScannerTests, MultipleThreadCtx) auto scanner = migration::cassandra::impl::FullTableScanner( {.ctxThreadsNum = 2, .jobsNum = 2, .cursorsPerJob = 100}, TestScannerAdapter(mockCallback) ); - scanner.wait(); + scanner.waitForAllAndThrowOnError(); } MATCHER(rangeMinMax, "Matches the range with min and max") @@ -105,5 +128,58 @@ TEST_F(FullTableScannerTests, RangeSizeIsOne) auto scanner = migration::cassandra::impl::FullTableScanner( {.ctxThreadsNum = 2, .jobsNum = 1, .cursorsPerJob = 1}, TestScannerAdapter(mockCallback) ); - scanner.wait(); + scanner.waitForAllAndThrowOnError(); +} + +TEST_F(FullTableScannerTests, WaitPropagatesWorkerError) +{ + testing::MockFunction< + void(migration::cassandra::impl::TokenRange const&, boost::asio::yield_context)> + mockCallback; + EXPECT_CALL(mockCallback, Call(testing::_, testing::_)) + .WillRepeatedly(testing::Throw(std::runtime_error("scan failure"))); + auto scanner = migration::cassandra::impl::FullTableScanner( + {.ctxThreadsNum = 1, .jobsNum = 1, .cursorsPerJob = 1}, TestScannerAdapter(mockCallback) + ); + try { + scanner.waitForAllAndThrowOnError(); + FAIL() << "expected waitForAllAndThrowOnError() to throw"; + } catch (std::runtime_error const& e) { + EXPECT_THAT(std::string{e.what()}, testing::HasSubstr("scan failure")); + } +} + +TEST_F(FullTableScannerTests, WaitReportsPartialFailure) +{ + // Two ranges across two workers; exactly one read fails. The wait must still throw and report + // the failed-worker count. + testing::MockFunction< + void(migration::cassandra::impl::TokenRange const&, boost::asio::yield_context)> + mockCallback; + EXPECT_CALL(mockCallback, Call(testing::_, testing::_)) + .WillOnce(testing::Throw(std::runtime_error("scan failure"))) + .WillRepeatedly(testing::Return()); + auto scanner = migration::cassandra::impl::FullTableScanner( + {.ctxThreadsNum = 2, .jobsNum = 2, .cursorsPerJob = 1}, TestScannerAdapter(mockCallback) + ); + try { + scanner.waitForAllAndThrowOnError(); + FAIL() << "expected waitForAllAndThrowOnError() to throw"; + } catch (std::runtime_error const& e) { + EXPECT_THAT(std::string{e.what()}, testing::HasSubstr("1 of 2 workers")); + } +} + +TEST_F(FullTableScannerTests, WaitJoinsAllWorkersBeforeThrowing) +{ + std::atomic_uint calls = 0; + std::atomic_bool slowWorkerFinished = false; + + auto scanner = migration::cassandra::impl::FullTableScanner( + {.ctxThreadsNum = 2, .jobsNum = 2, .cursorsPerJob = 1}, + FailingAndSlowAdapter{.calls = calls, .slowWorkerFinished = slowWorkerFinished} + ); + + EXPECT_THROW(scanner.waitForAllAndThrowOnError(), std::runtime_error); + EXPECT_TRUE(slowWorkerFinished); } diff --git a/tests/unit/migration/cassandra/SpecTests.cpp b/tests/unit/migration/cassandra/SpecTests.cpp index 5d5e89a20a..9069ce26b2 100644 --- a/tests/unit/migration/cassandra/SpecTests.cpp +++ b/tests/unit/migration/cassandra/SpecTests.cpp @@ -11,6 +11,7 @@ class Empty {}; struct SimpleTestTable { using Row = std::tuple; static constexpr char const* kPartitionKey = "key"; + static constexpr char const* kSelectColumns = "key, value"; static constexpr char const* kTableName = "test"; }; } // namespace diff --git a/tests/unit/migration/cassandra/TransactionsAdapterTests.cpp b/tests/unit/migration/cassandra/TransactionsAdapterTests.cpp new file mode 100644 index 0000000000..555334ed5d --- /dev/null +++ b/tests/unit/migration/cassandra/TransactionsAdapterTests.cpp @@ -0,0 +1,118 @@ +#include "migration/cassandra/impl/TransactionsAdapter.hpp" +#include "util/TestObject.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +using namespace migration::cassandra::impl; + +namespace { + +constexpr auto kAccount1 = "rM2AGCCCRb373FRuD8wHyUwUsh2dV4BW5Q"; +constexpr auto kAccount2 = "rK1EX542EgA9m948JrJRaEzwLVEhqWvnr9"; +constexpr std::uint32_t kLedgerSeq = 100; + +// onRowRead never touches the backend pointer; the deserialization logic is pure. +xrpl::Blob +serializedPaymentTx() +{ + return createPaymentTransactionObject(kAccount1, kAccount2, 1, 10, 1) + .getSerializer() + .peekData(); +} + +xrpl::Blob +serializedPaymentMeta() +{ + return createPaymentTransactionMetaObject(kAccount1, kAccount2, 100, 200) + .getSerializer() + .peekData(); +} + +// Build a row as onRowRead expects it: {hash, date, ledgerSeq, metadata, transaction}. +TableTransactionsDesc::Row +makeRow(xrpl::Blob metaBlob, xrpl::Blob txBlob) +{ + return TableTransactionsDesc::Row{ + xrpl::uint256{}, std::uint64_t{0}, kLedgerSeq, std::move(metaBlob), std::move(txBlob) + }; +} + +} // namespace + +// A well-formed transaction row is deserialized and forwarded to the callback with the +// reconstructed STTx and TxMeta. +TEST(TransactionsAdapterTest, ValidRowInvokesCallback) +{ + auto const txBlob = serializedPaymentTx(); + bool called = false; + bool decodeFailed = false; + xrpl::uint256 seenTxId; + xrpl::uint256 seenMetaTxId; + std::uint32_t seenLgrSeq = 0; + + TransactionsAdapter adapter{ + nullptr, + [&](xrpl::STTx const& sttx, xrpl::TxMeta const& txMeta) { + called = true; + seenTxId = sttx.getTransactionID(); + seenMetaTxId = txMeta.getTxID(); + seenLgrSeq = txMeta.getLgrSeq(); + }, + [&] { decodeFailed = true; } + }; + + adapter.onRowRead(makeRow(serializedPaymentMeta(), txBlob)); + + EXPECT_FALSE(decodeFailed); + ASSERT_TRUE(called); + + xrpl::STTx const expectedTx{xrpl::SerialIter{txBlob.data(), txBlob.size()}}; + EXPECT_EQ(seenTxId, expectedTx.getTransactionID()); + EXPECT_EQ(seenMetaTxId, expectedTx.getTransactionID()); + EXPECT_EQ(seenLgrSeq, kLedgerSeq); +} + +// A transaction blob that fails to deserialize does not invoke the transaction callback and instead +// reports a decode failure to the owner, which decides the policy (count, abort). +TEST(TransactionsAdapterTest, CorruptTransactionBlobReportsDecodeFailure) +{ + bool called = false; + bool decodeFailed = false; + TransactionsAdapter adapter{ + nullptr, + [&](xrpl::STTx const&, xrpl::TxMeta const&) { called = true; }, + [&] { decodeFailed = true; } + }; + + adapter.onRowRead(makeRow(serializedPaymentMeta(), xrpl::Blob{0xFF, 0xFF, 0xFF, 0xFF})); + + EXPECT_FALSE(called); + EXPECT_TRUE(decodeFailed); +} + +// A metadata blob that fails to deserialize (valid transaction, garbage meta) also reports a decode +// failure without invoking the transaction callback. +TEST(TransactionsAdapterTest, CorruptMetadataBlobReportsDecodeFailure) +{ + bool called = false; + bool decodeFailed = false; + TransactionsAdapter adapter{ + nullptr, + [&](xrpl::STTx const&, xrpl::TxMeta const&) { called = true; }, + [&] { decodeFailed = true; } + }; + + adapter.onRowRead(makeRow(xrpl::Blob{0xFF, 0xFF, 0xFF, 0xFF}, serializedPaymentTx())); + + EXPECT_FALSE(called); + EXPECT_TRUE(decodeFailed); +} diff --git a/tests/unit/util/BatchingTests.cpp b/tests/unit/util/BatchingTests.cpp index 0e06db5e47..cbadc58df3 100644 --- a/tests/unit/util/BatchingTests.cpp +++ b/tests/unit/util/BatchingTests.cpp @@ -3,7 +3,10 @@ #include #include +#include #include +#include +#include #include TEST(BatchingTests, simpleBatch) @@ -57,3 +60,118 @@ TEST(BatchingTests, emptyInput) EXPECT_EQ(input, output); } + +namespace { + +// Collects every batch handed to a util::BatchBuffer sink for later inspection. +struct BatchSink { + std::vector> batches; + + auto + fn() + { + return [this](std::vector const& records) { batches.push_back(records); }; + } + + [[nodiscard]] std::vector + flattened() const + { + std::vector all; + for (auto const& batch : batches) + all.insert(all.end(), batch.begin(), batch.end()); + return all; + } +}; + +} // namespace + +TEST(BatchBufferTests, buffersBelowThresholdUntilFlush) +{ + BatchSink sink; + util::BatchBuffer buffer{3, sink.fn()}; + + buffer.add({1}); + buffer.add({2}); + EXPECT_TRUE(sink.batches.empty()) << "Nothing should flush before the threshold"; + + buffer.flush(); + ASSERT_EQ(sink.batches.size(), 1); + EXPECT_EQ(sink.batches[0], (std::vector{1, 2})); +} + +TEST(BatchBufferTests, flushesExactlyOnThresholdWithNoRemainder) +{ + BatchSink sink; + util::BatchBuffer buffer{3, sink.fn()}; + + buffer.add({1}); + buffer.add({2}); + buffer.add({3}); // reaches the threshold exactly + ASSERT_EQ(sink.batches.size(), 1); + EXPECT_EQ(sink.batches[0], (std::vector{1, 2, 3})); + + buffer.flush(); // buffer already empty: no extra invocation + EXPECT_EQ(sink.batches.size(), 1); +} + +TEST(BatchBufferTests, flushesMultipleFullBatchesPlusRemainder) +{ + BatchSink sink; + util::BatchBuffer buffer{2, sink.fn()}; + + for (int i = 1; i <= 5; ++i) + buffer.add({i}); + + EXPECT_EQ(sink.batches.size(), 2) << "Two full batches flushed inline"; + buffer.flush(); + ASSERT_EQ(sink.batches.size(), 3) << "Odd remainder flushed at the end"; + EXPECT_EQ(sink.flattened(), (std::vector{1, 2, 3, 4, 5})); +} + +TEST(BatchBufferTests, singleAddLargerThanThresholdFlushesWholeBufferAsOneBatch) +{ + BatchSink sink; + util::BatchBuffer buffer{3, sink.fn()}; + + buffer.add({1, 2, 3, 4, 5}); + ASSERT_EQ(sink.batches.size(), 1); + EXPECT_EQ(sink.batches[0], (std::vector{1, 2, 3, 4, 5})); + + buffer.flush(); + EXPECT_EQ(sink.batches.size(), 1); +} + +TEST(BatchBufferTests, flushOnEmptyBufferDoesNotInvokeSink) +{ + BatchSink sink; + util::BatchBuffer buffer{3, sink.fn()}; + + buffer.flush(); + EXPECT_TRUE(sink.batches.empty()); +} + +TEST(BatchBufferTests, concurrentAddsConserveAllRecords) +{ + std::mutex mutex; + std::size_t total = 0; + util::BatchBuffer buffer{10, [&](std::vector const& records) { + std::scoped_lock const lock{mutex}; + total += records.size(); + }}; + + constexpr std::size_t kThreads = 8; + constexpr std::size_t kAddsPerThread = 1'000; + std::vector threads; + threads.reserve(kThreads); + for (std::size_t t = 0; t < kThreads; ++t) { + threads.emplace_back([&] { + for (std::size_t i = 0; i < kAddsPerThread; ++i) + buffer.add({1}); + }); + } + for (auto& thread : threads) + thread.join(); + + buffer.flush(); + EXPECT_EQ(total, kThreads * kAddsPerThread) << "Every record must reach the sink exactly once"; +}