diff --git a/src/data/BackendInterface.hpp b/src/data/BackendInterface.hpp index 5c0bc05e0..17cd08574 100644 --- a/src/data/BackendInterface.hpp +++ b/src/data/BackendInterface.hpp @@ -408,6 +408,60 @@ public: boost::asio::yield_context yield ) const = 0; + /** + * @brief Fetches transactions for a particular MPTokenIssuance ID. + * + * Returns one page of transactions for this issuance, newest-first (or oldest-first when + * @p forward is set). Each row of the mptoken_issuance_transactions table holds only a + * transaction hash and its ledger position, not the transaction itself or its type. So the + * query cannot filter by type. When a type filter is requested, the handler looks up each full + * transaction by hash and drops rows whose type does not match, the same way account_tx does. + * + * @param mptIssuanceID The 24-byte MPTokenIssuance ID. + * @param limit The maximum number of transactions per result page. + * @param forward Whether to fetch the page forwards or backwards from the given cursor. + * @param cursorIn The cursor to resume fetching from. + * @param yield The coroutine context. + * @return Results and a cursor to resume from. + */ + virtual TransactionsAndCursor + fetchMPTokenIssuanceTransactions( + ripple::uint192 const& mptIssuanceID, + std::uint32_t limit, + bool forward, + std::optional const& cursorIn, + boost::asio::yield_context yield + ) const = 0; + + /** + * @brief Fetches transactions for a particular MPTokenIssuance ID involving a particular + * account. + * + * Returns one page of transactions for this issuance and account, newest-first (or oldest-first + * when @p forward is set). Each row of the account_mptoken_issuance_transactions table holds + * only a transaction hash and its ledger position, not the transaction itself or its type. So + * the query cannot filter by type. When a type filter is requested, the handler looks up each + * full transaction by hash and drops rows whose type does not match, the same way account_tx + * does. + * + * @param mptIssuanceID The 24-byte MPTokenIssuance ID. + * @param account The account that must be affected by the transaction. + * @param limit The maximum number of transactions per result page. + * @param forward Whether to fetch the page forwards or backwards from the given cursor. + * @param cursorIn The cursor to resume fetching from. + * @param yield The coroutine context. + * @return Results and a cursor to resume from. + */ + virtual TransactionsAndCursor + fetchAccountMPTokenIssuanceTransactions( + ripple::uint192 const& mptIssuanceID, + ripple::AccountID const& account, + std::uint32_t limit, + bool forward, + std::optional const& cursorIn, + boost::asio::yield_context yield + ) const = 0; + /** * @brief Fetches a specific ledger object. * @@ -725,6 +779,28 @@ public: virtual void writeNFTTransactions(std::vector const& data) = 0; + /** + * @brief Write MPTokenIssuance transaction index rows to the `mptoken_issuance_transactions` + * table. + * + * @param data A vector of MPTokenIssuanceTransactionsData objects. + */ + virtual void + writeMPTokenIssuanceTransactions(std::vector const& data) = 0; + + /** + * @brief Write MPTokenIssuance transaction index rows to the + * `account_mptoken_issuance_transactions` table. + * + * One row is written per affected account in each record. + * + * @param data A vector of MPTokenIssuanceTransactionsData objects. + */ + virtual void + writeAccountMPTokenIssuanceTransactions( + std::vector const& data + ) = 0; + /** * @brief Write accounts that started holding onto a MPT. * diff --git a/src/data/DBHelpers.hpp b/src/data/DBHelpers.hpp index 7e1c42607..181e4ea1c 100644 --- a/src/data/DBHelpers.hpp +++ b/src/data/DBHelpers.hpp @@ -199,6 +199,20 @@ struct MPTHolderData { ripple::AccountID holder; }; +/** + * @brief Represents a transaction link for an MPTokenIssuance. + * + * @note Writing one of these records inserts into two tables: + * mptoken_issuance_transactions and account_mptoken_issuance_transactions. + */ +struct MPTokenIssuanceTransactionsData { + ripple::uint192 mptIssuanceID; + boost::container::flat_set accounts; + std::uint32_t ledgerSequence{}; + std::uint32_t transactionIndex{}; + ripple::uint256 txHash; +}; + /** * @brief Check whether the supplied object is a dir node. * diff --git a/src/data/cassandra/CassandraBackendFamily.hpp b/src/data/cassandra/CassandraBackendFamily.hpp index 3d29998f5..c74b067cc 100644 --- a/src/data/cassandra/CassandraBackendFamily.hpp +++ b/src/data/cassandra/CassandraBackendFamily.hpp @@ -77,6 +77,13 @@ protected: // TODO: move to interface level mutable FetchLedgerCacheType ledgerCache_{}; + static constexpr std::size_t kTransactionCursorBindIndex = 1; + static constexpr std::size_t kTransactionLimitBindIndex = 2; + static constexpr std::size_t kMPTokenIssuanceTxCursorBindIndex = 1; + static constexpr std::size_t kMPTokenIssuanceTxLimitBindIndex = 2; + static constexpr std::size_t kAccountMPTokenIssuanceTxCursorBindIndex = 2; + static constexpr std::size_t kAccountMPTokenIssuanceTxLimitBindIndex = 3; + public: /** * @brief Create a new cassandra/scylla backend instance. @@ -154,14 +161,16 @@ public: auto cursor = txnCursor; if (cursor) { - statement.bindAt(1, cursor->asTuple()); + statement.bindAt(kTransactionCursorBindIndex, cursor->asTuple()); LOG(log_.debug()) << "account = " << ripple::strHex(account) << " tuple = " << cursor->ledgerSequence << cursor->transactionIndex; } else { auto const seq = forward ? rng->minSequence : rng->maxSequence; auto const placeHolder = forward ? 0u : std::numeric_limits::max(); - statement.bindAt(1, std::make_tuple(placeHolder, placeHolder)); + statement.bindAt( + kTransactionCursorBindIndex, std::make_tuple(placeHolder, placeHolder) + ); LOG(log_.debug()) << "account = " << ripple::strHex(account) << " idx = " << seq << " tuple = " << placeHolder; } @@ -169,7 +178,7 @@ public: // FIXME: Limit is a hack to support uint32_t properly for the time // being. Should be removed later and schema updated to use proper // types. - statement.bindAt(2, Limit{limit}); + statement.bindAt(kTransactionLimitBindIndex, Limit{limit}); auto const res = executor_.read(yield, statement); auto const& results = res.value(); if (not results.hasRows()) { @@ -435,19 +444,21 @@ public: auto cursor = cursorIn; if (cursor) { - statement.bindAt(1, cursor->asTuple()); + statement.bindAt(kTransactionCursorBindIndex, cursor->asTuple()); LOG(log_.debug()) << "token_id = " << ripple::strHex(tokenID) << " tuple = " << cursor->ledgerSequence << cursor->transactionIndex; } else { auto const seq = forward ? rng->minSequence : rng->maxSequence; auto const placeHolder = forward ? 0 : std::numeric_limits::max(); - statement.bindAt(1, std::make_tuple(placeHolder, placeHolder)); + statement.bindAt( + kTransactionCursorBindIndex, std::make_tuple(placeHolder, placeHolder) + ); LOG(log_.debug()) << "token_id = " << ripple::strHex(tokenID) << " idx = " << seq << " tuple = " << placeHolder; } - statement.bindAt(2, Limit{limit}); + statement.bindAt(kTransactionLimitBindIndex, Limit{limit}); auto const res = executor_.read(yield, statement); auto const& results = res.value(); @@ -485,6 +496,59 @@ public: return {txns, {}}; } + TransactionsAndCursor + fetchMPTokenIssuanceTransactions( + ripple::uint192 const& mptIssuanceID, + std::uint32_t const limit, + bool const forward, + std::optional const& cursorIn, + boost::asio::yield_context yield + ) const override + { + auto const statement = [this, forward, &mptIssuanceID]() { + if (forward) + return schema_->selectMPTokenIssuanceTxForward.bind(mptIssuanceID); + + return schema_->selectMPTokenIssuanceTx.bind(mptIssuanceID); + }(); + return fetchMPTokenIssuanceTransactionsImpl( + statement, + kMPTokenIssuanceTxCursorBindIndex, + kMPTokenIssuanceTxLimitBindIndex, + limit, + forward, + cursorIn, + yield + ); + } + + TransactionsAndCursor + fetchAccountMPTokenIssuanceTransactions( + ripple::uint192 const& mptIssuanceID, + ripple::AccountID const& account, + std::uint32_t const limit, + bool const forward, + std::optional const& cursorIn, + boost::asio::yield_context yield + ) const override + { + auto const statement = [this, forward, &mptIssuanceID, &account]() { + if (forward) + return schema_->selectAccountMPTokenIssuanceTxForward.bind(mptIssuanceID, account); + + return schema_->selectAccountMPTokenIssuanceTx.bind(mptIssuanceID, account); + }(); + return fetchMPTokenIssuanceTransactionsImpl( + statement, + kAccountMPTokenIssuanceTxCursorBindIndex, + kAccountMPTokenIssuanceTxLimitBindIndex, + limit, + forward, + cursorIn, + yield + ); + } + MPTHoldersAndCursor fetchMPTHolders( ripple::uint192 const& mptID, @@ -877,6 +941,55 @@ public: executor_.write(std::move(statements)); } + void + writeMPTokenIssuanceTransactions( + std::vector const& data + ) override + { + std::vector statements; + statements.reserve(data.size()); + + std::ranges::transform(data, std::back_inserter(statements), [this](auto const& record) { + return schema_->insertMPTokenIssuanceTx.bind( + record.mptIssuanceID, + std::make_tuple(record.ledgerSequence, record.transactionIndex), + record.txHash + ); + }); + + executor_.write(std::move(statements)); + } + + void + writeAccountMPTokenIssuanceTransactions( + std::vector const& data + ) override + { + std::size_t numStatements = 0u; + for (auto const& record : data) + numStatements += record.accounts.size(); + + std::vector statements; + statements.reserve(numStatements); + + for (auto const& record : data) { + std::ranges::transform( + record.accounts, + std::back_inserter(statements), + [this, &record](auto const& account) { + return schema_->insertAccountMPTokenIssuanceTx.bind( + record.mptIssuanceID, + account, + std::make_tuple(record.ledgerSequence, record.transactionIndex), + record.txHash + ); + } + ); + } + + executor_.write(std::move(statements)); + } + void writeTransaction( std::string&& hash, @@ -1016,6 +1129,86 @@ protected: return true; } + + /** + * @brief Shared implementation of the two MPTokenIssuance transaction-index fetchers. + * + * @note The forward path queries with an inclusive seq_idx >=, + * so the returned cursor's transaction index is advanced + * by one to avoid re-reading the last row on the next page. + * + * @param statement The statement already bound with the partition-key columns. + * @param cursorIdx The bind index for the `seq_idx` cursor tuple. + * @param limitIdx The bind index for the `LIMIT`. + * @param limit The maximum number of transactions per result page. + * @param forward Whether the page is fetched forwards or backwards. + * @param cursorIn The cursor to resume fetching from. + * @param yield The coroutine context. + * @return Results and a cursor to resume from. + */ + TransactionsAndCursor + fetchMPTokenIssuanceTransactionsImpl( + Statement const& statement, + std::size_t const cursorIdx, + std::size_t const limitIdx, + std::uint32_t const limit, + bool const forward, + std::optional const& cursorIn, + boost::asio::yield_context yield + ) const + { + auto rng = fetchLedgerRange(); + if (!rng) + return {.txns = {}, .cursor = {}}; + + auto cursor = cursorIn; + if (cursor.has_value()) { + statement.bindAt(cursorIdx, cursor->asTuple()); + } else { + // Forward uses the nft_history-style inclusive lower bound; reverse starts just past + // the latest validated ledger so its exclusive `<` query includes that ledger's rows. + auto const ledgerSequence = forward ? rng->minSequence : rng->maxSequence; + auto const transactionIndex = forward ? 0u : std::numeric_limits::max(); + statement.bindAt(cursorIdx, std::make_tuple(ledgerSequence, transactionIndex)); + } + + statement.bindAt(limitIdx, Limit{limit}); + + auto const res = executor_.read(yield, statement); + auto const& results = res.value(); + if (not results.hasRows()) { + LOG(log_.debug()) << "No rows returned"; + return {}; + } + + std::vector hashes = {}; + auto numRows = results.numRows(); + + for (auto const& [hash, data] : + extract>(results)) { + hashes.push_back(hash); + + if (--numRows == 0) { + LOG(log_.debug()) << "Setting cursor"; + cursor = data; + + // forward queries by ledger/tx sequence `>=` + // so we have to advance the index by one + if (forward) + ++cursor->transactionIndex; + } + } + + auto txns = fetchTransactions(hashes, yield); + LOG(log_.debug()) << "MPTokenIssuance Txns = " << txns.size(); + + if (txns.size() == limit) { + LOG(log_.debug()) << "Returning cursor"; + return {std::move(txns), cursor}; + } + + return {std::move(txns), {}}; + } }; } // namespace data::cassandra diff --git a/src/data/cassandra/Schema.hpp b/src/data/cassandra/Schema.hpp index 8725ee302..eda31318a 100644 --- a/src/data/cassandra/Schema.hpp +++ b/src/data/cassandra/Schema.hpp @@ -278,9 +278,42 @@ public: R"( CREATE TABLE IF NOT EXISTS {} ( - mpt_id blob, - holder blob, - PRIMARY KEY (mpt_id, holder) + mptoken_issuance_id blob, + seq_idx tuple, + hash blob, + PRIMARY KEY (mptoken_issuance_id, seq_idx) + ) + WITH CLUSTERING ORDER BY (seq_idx DESC) + )", + qualifiedTableName(settingsProvider_.get(), "mptoken_issuance_transactions") + ) + ); + + statements.emplace_back( + fmt::format( + R"( + CREATE TABLE IF NOT EXISTS {} + ( + mptoken_issuance_id blob, + account blob, + seq_idx tuple, + hash blob, + PRIMARY KEY ((mptoken_issuance_id, account), seq_idx) + ) + WITH CLUSTERING ORDER BY (seq_idx DESC) + )", + qualifiedTableName(settingsProvider_.get(), "account_mptoken_issuance_transactions") + ) + ); + + statements.emplace_back( + fmt::format( + R"( + CREATE TABLE IF NOT EXISTS {} + ( + mpt_id blob, + holder blob, + PRIMARY KEY (mpt_id, holder) ) WITH CLUSTERING ORDER BY (holder ASC) )", @@ -474,6 +507,34 @@ public: ); }(); + PreparedStatement insertMPTokenIssuanceTx = [this]() { + return handle_.get().prepare( + fmt::format( + R"( + INSERT INTO {} + (mptoken_issuance_id, seq_idx, hash) + VALUES (?, ?, ?) + )", + qualifiedTableName(settingsProvider_.get(), "mptoken_issuance_transactions") + ) + ); + }(); + + PreparedStatement insertAccountMPTokenIssuanceTx = [this]() { + return handle_.get().prepare( + fmt::format( + R"( + INSERT INTO {} + (mptoken_issuance_id, account, seq_idx, hash) + VALUES (?, ?, ?, ?) + )", + qualifiedTableName( + settingsProvider_.get(), "account_mptoken_issuance_transactions" + ) + ) + ); + }(); + PreparedStatement insertMPTHolder = [this]() { return handle_.get().prepare( fmt::format( @@ -740,6 +801,76 @@ public: ); }(); + PreparedStatement selectMPTokenIssuanceTx = [this]() { + return handle_.get().prepare( + fmt::format( + R"( + SELECT hash, seq_idx + FROM {} + WHERE mptoken_issuance_id = ? + AND seq_idx < ? + ORDER BY seq_idx DESC + LIMIT ? + )", + qualifiedTableName(settingsProvider_.get(), "mptoken_issuance_transactions") + ) + ); + }(); + + PreparedStatement selectMPTokenIssuanceTxForward = [this]() { + return handle_.get().prepare( + fmt::format( + R"( + SELECT hash, seq_idx + FROM {} + WHERE mptoken_issuance_id = ? + AND seq_idx >= ? + ORDER BY seq_idx ASC + LIMIT ? + )", + qualifiedTableName(settingsProvider_.get(), "mptoken_issuance_transactions") + ) + ); + }(); + + PreparedStatement selectAccountMPTokenIssuanceTx = [this]() { + return handle_.get().prepare( + fmt::format( + R"( + SELECT hash, seq_idx + FROM {} + WHERE mptoken_issuance_id = ? + AND account = ? + AND seq_idx < ? + ORDER BY seq_idx DESC + LIMIT ? + )", + qualifiedTableName( + settingsProvider_.get(), "account_mptoken_issuance_transactions" + ) + ) + ); + }(); + + PreparedStatement selectAccountMPTokenIssuanceTxForward = [this]() { + return handle_.get().prepare( + fmt::format( + R"( + SELECT hash, seq_idx + FROM {} + WHERE mptoken_issuance_id = ? + AND account = ? + AND seq_idx >= ? + ORDER BY seq_idx ASC + LIMIT ? + )", + qualifiedTableName( + settingsProvider_.get(), "account_mptoken_issuance_transactions" + ) + ) + ); + }(); + PreparedStatement selectNFTIDsByIssuerTaxon = [this]() { return handle_.get().prepare( fmt::format( diff --git a/tests/common/util/MockBackend.hpp b/tests/common/util/MockBackend.hpp index 111af2bc2..cb8d622be 100644 --- a/tests/common/util/MockBackend.hpp +++ b/tests/common/util/MockBackend.hpp @@ -103,6 +103,29 @@ struct MockBackend : public BackendInterface { (const, override) ); + MOCK_METHOD( + data::TransactionsAndCursor, + fetchMPTokenIssuanceTransactions, + (ripple::uint192 const&, + std::uint32_t, + bool, + std::optional const&, + boost::asio::yield_context), + (const, override) + ); + + MOCK_METHOD( + data::TransactionsAndCursor, + fetchAccountMPTokenIssuanceTransactions, + (ripple::uint192 const&, + ripple::AccountID const&, + std::uint32_t, + bool, + std::optional const&, + boost::asio::yield_context), + (const, override) + ); + MOCK_METHOD( data::NFTsAndCursor, fetchNFTsByIssuer, @@ -204,6 +227,20 @@ struct MockBackend : public BackendInterface { MOCK_METHOD(void, writeNFTTransactions, (std::vector const&), (override)); + MOCK_METHOD( + void, + writeMPTokenIssuanceTransactions, + (std::vector const&), + (override) + ); + + MOCK_METHOD( + void, + writeAccountMPTokenIssuanceTransactions, + (std::vector const&), + (override) + ); + MOCK_METHOD( void, writeSuccessor, diff --git a/tests/integration/data/cassandra/BackendTests.cpp b/tests/integration/data/cassandra/BackendTests.cpp index f05b03b01..50ddb2412 100644 --- a/tests/integration/data/cassandra/BackendTests.cpp +++ b/tests/integration/data/cassandra/BackendTests.cpp @@ -33,6 +33,7 @@ #include #include #include +#include #include #include #include @@ -48,6 +49,7 @@ #include #include #include +#include #include #include #include @@ -1461,8 +1463,403 @@ TEST_F(CacheBackendCassandraTest, CacheFetchLedgerBySeq) }); } +struct BackendCassandraMPTokenIssuanceTest : BackendCassandraTest { + static ripple::uint192 + makeMptIssuanceId() + { + return ripple::makeMptID(1, makeAccount(0x01)); + } + + static ripple::AccountID + makeAccount(std::uint8_t seed) + { + return ripple::AccountID{seed}; + } + + static ripple::uint256 + makeHash(std::uint8_t seed) + { + return ripple::uint256{seed}; + } + + // Writes a single ledger so that fetchLedgerRange() is populated, which the fetchers + // require before reading any index rows. + void + setupLedgerRange(std::uint32_t seq) + { + std::string rawHeaderBlob = hexStringToBinaryString(kRawheader); + ripple::LedgerHeader lgrInfo = util::deserializeHeader(ripple::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)); + auto const rng = backend_->fetchLedgerRange(); + ASSERT_TRUE(rng.has_value()); + } + + void + writeTxBlob(ripple::uint256 const& hash, std::uint32_t seq) + { + backend_->writeTransaction( + uint256ToString(hash), + seq, + 0, + "tx_" + ripple::strHex(hash), + "meta_" + ripple::strHex(hash) + ); + } +}; + +TEST_F(BackendCassandraMPTokenIssuanceTest, RoundTripIssuanceAndAccountIndexes) +{ + runSpawn([this](boost::asio::yield_context yield) { + auto const mptIssuanceId = makeMptIssuanceId(); + auto const account = makeAccount(0x42); + auto const secondAccount = makeAccount(0x43); + std::uint32_t const seq = 100; + + setupLedgerRange(seq); + + auto const hash = makeHash(0x01); + writeTxBlob(hash, seq); + auto const expectedTxBlob = "tx_" + ripple::strHex(hash); + auto const expectedMetaBlob = "meta_" + ripple::strHex(hash); + + auto expectFetchedSingle = [&](auto const& txns) { + ASSERT_EQ(txns.size(), 1); + EXPECT_EQ( + std::string(txns[0].transaction.begin(), txns[0].transaction.end()), expectedTxBlob + ); + EXPECT_EQ( + std::string(txns[0].metadata.begin(), txns[0].metadata.end()), expectedMetaBlob + ); + EXPECT_EQ(txns[0].ledgerSequence, seq); + }; + + MPTokenIssuanceTransactionsData const record{ + .mptIssuanceID = mptIssuanceId, + .accounts = {account, secondAccount}, + .ledgerSequence = seq, + .transactionIndex = 1, + .txHash = hash + }; + backend_->writeMPTokenIssuanceTransactions({record}); + backend_->writeAccountMPTokenIssuanceTransactions({record}); + backend_->waitForWritesToFinish(); + + { + auto [txns, cursor] = + backend_->fetchMPTokenIssuanceTransactions(mptIssuanceId, 100, false, {}, yield); + expectFetchedSingle(txns); + EXPECT_FALSE(cursor); + } + { + auto [txns, cursor] = backend_->fetchAccountMPTokenIssuanceTransactions( + mptIssuanceId, account, 100, false, {}, yield + ); + expectFetchedSingle(txns); + EXPECT_FALSE(cursor); + } + // Both affected accounts are indexed: one row was written per account. + { + auto [txns, cursor] = backend_->fetchAccountMPTokenIssuanceTransactions( + mptIssuanceId, secondAccount, 100, false, {}, yield + ); + expectFetchedSingle(txns); + EXPECT_FALSE(cursor); + } + { + auto [txns, cursor] = backend_->fetchAccountMPTokenIssuanceTransactions( + mptIssuanceId, makeAccount(0x99), 100, false, {}, yield + ); + EXPECT_EQ(txns.size(), 0); + } + { + auto const secondHash = makeHash(0x02); + writeTxBlob(secondHash, seq); + auto const expectedSecondTxBlob = "tx_" + ripple::strHex(secondHash); + auto const expectedSecondMetaBlob = "meta_" + ripple::strHex(secondHash); + auto expectFetchedSecond = [&](auto const& txns) { + ASSERT_EQ(txns.size(), 1); + EXPECT_EQ( + std::string(txns[0].transaction.begin(), txns[0].transaction.end()), + expectedSecondTxBlob + ); + EXPECT_EQ( + std::string(txns[0].metadata.begin(), txns[0].metadata.end()), + expectedSecondMetaBlob + ); + EXPECT_EQ(txns[0].ledgerSequence, seq); + }; + MPTokenIssuanceTransactionsData const secondRecord{ + .mptIssuanceID = mptIssuanceId, + .accounts = {account}, + .ledgerSequence = seq, + .transactionIndex = 2, + .txHash = secondHash + }; + backend_->writeMPTokenIssuanceTransactions({secondRecord}); + backend_->writeAccountMPTokenIssuanceTransactions({secondRecord}); + backend_->waitForWritesToFinish(); + + { + auto [txns, cursor] = + backend_->fetchMPTokenIssuanceTransactions(mptIssuanceId, 1, false, {}, yield); + expectFetchedSecond(txns); + EXPECT_TRUE(cursor); + } + { + auto [txns, cursor] = backend_->fetchAccountMPTokenIssuanceTransactions( + mptIssuanceId, account, 1, false, {}, yield + ); + expectFetchedSecond(txns); + EXPECT_TRUE(cursor); + } + } + }); +} + +TEST_F(BackendCassandraMPTokenIssuanceTest, DescendingOrderForwardAndReverse) +{ + runSpawn([this](boost::asio::yield_context yield) { + auto const mptIssuanceId = makeMptIssuanceId(); + std::uint32_t const baseSeq = 200; + + // Three txns in three different ledgers, so ordering is checked across ledgers, + // not just by transaction index within one ledger. + std::vector hashes; + std::vector seqs; + for (std::uint8_t i = 1; i <= 3; ++i) { + auto const seq = baseSeq + i; + seqs.push_back(seq); + setupLedgerRange(seq); + auto const hash = makeHash(i); + hashes.push_back(hash); + writeTxBlob(hash, seq); + MPTokenIssuanceTransactionsData const record{ + .mptIssuanceID = mptIssuanceId, + .accounts = {}, + .ledgerSequence = seq, + .transactionIndex = i, + .txHash = hash + }; + backend_->writeMPTokenIssuanceTransactions({record}); + } + backend_->waitForWritesToFinish(); + + auto txBlobToString = [](data::TransactionAndMetadata const& tx) { + return std::string(tx.transaction.begin(), tx.transaction.end()); + }; + auto expectedBlob = [&](std::uint8_t i) { return "tx_" + ripple::strHex(makeHash(i)); }; + + // Reverse (forward=false): newest first -> rows 3, 2, 1. + { + auto [txns, cursor] = + backend_->fetchMPTokenIssuanceTransactions(mptIssuanceId, 100, false, {}, yield); + ASSERT_EQ(txns.size(), 3); + EXPECT_FALSE(cursor); + EXPECT_EQ(txBlobToString(txns[0]), expectedBlob(3)); + EXPECT_EQ(txBlobToString(txns[1]), expectedBlob(2)); + EXPECT_EQ(txBlobToString(txns[2]), expectedBlob(1)); + } + // Forward (forward=true): oldest first -> rows 1, 2, 3 (the reverse order). + { + auto [txns, cursor] = + backend_->fetchMPTokenIssuanceTransactions(mptIssuanceId, 100, true, {}, yield); + ASSERT_EQ(txns.size(), 3); + EXPECT_FALSE(cursor); + EXPECT_EQ(txBlobToString(txns[0]), expectedBlob(1)); + EXPECT_EQ(txBlobToString(txns[1]), expectedBlob(2)); + EXPECT_EQ(txBlobToString(txns[2]), expectedBlob(3)); + } + }); +} + +TEST_F(BackendCassandraMPTokenIssuanceTest, MarkerPaginationRoundTrip) +{ + runSpawn([this](boost::asio::yield_context yield) { + auto const mptIssuanceId = makeMptIssuanceId(); + std::uint32_t const baseSeq = 300; + + enum class ExpectedPaginationEnd { PartialPage, EmptyPage }; + + auto txBlobToString = [](data::TransactionAndMetadata const& tx) { + return std::string(tx.transaction.begin(), tx.transaction.end()); + }; + auto expectedBlob = [&](std::uint8_t i) { return "tx_" + ripple::strHex(makeHash(i)); }; + auto expectSeenInOrder = [](std::vector const& seen, + bool forward, + std::set const& expected) { + std::vector expectedOrder(expected.begin(), expected.end()); + if (not forward) + std::ranges::reverse(expectedOrder); + + EXPECT_EQ(seen, expectedOrder) + << "pagination returned rows out of order for forward=" << forward; + }; + + // Writes `total` rows, each in its own ledger and at a distinct transaction index, + // so paging covers ordering across both. + auto setup = + [&](ripple::uint192 const& issuanceId, std::uint8_t total, std::uint32_t firstSeq) { + std::set expected; + for (std::uint8_t i = 1; i <= total; ++i) { + auto const seq = firstSeq + i - 1; + setupLedgerRange(seq); + auto const hash = makeHash(i); + writeTxBlob(hash, seq); + MPTokenIssuanceTransactionsData const record{ + .mptIssuanceID = issuanceId, + .accounts = {}, + .ledgerSequence = seq, + .transactionIndex = i, + .txHash = hash + }; + backend_->writeMPTokenIssuanceTransactions({record}); + expected.insert(expectedBlob(i)); + } + backend_->waitForWritesToFinish(); + return expected; + }; + + // Page through every row; assert page order plus that the union of pages equals `expected` + // exactly: every row seen exactly once (no duplicates from a repeated cursor row, no gaps + // from a dropped row). Each full page (one that returns a cursor) must be exactly `limit`. + auto pageThrough = [&](ripple::uint192 const& issuanceId, + bool forward, + std::uint32_t limit, + std::set const& expected, + ExpectedPaginationEnd expectedEnd) { + ASSERT_NE(limit, 0u); + auto const limitSize = static_cast(limit); + + std::vector seen; + std::optional cursor; + std::size_t pages = 0; + auto const maxPages = (expected.size() / limitSize) + 2; + bool sawEmptyTerminator = false; + do { + auto [txns, retCursor] = backend_->fetchMPTokenIssuanceTransactions( + issuanceId, limit, forward, cursor, yield + ); + ++pages; + // Guard against an infinite loop from a non-advancing cursor. + ASSERT_LE(pages, maxPages) + << "pagination did not terminate cleanly for forward=" << forward; + if (txns.empty()) { + sawEmptyTerminator = true; + EXPECT_FALSE(retCursor); + } else { + EXPECT_LE(txns.size(), limitSize); + if (retCursor) + EXPECT_EQ(txns.size(), limitSize); + } + for (auto const& tx : txns) + seen.push_back(txBlobToString(tx)); + cursor = retCursor; + } while (cursor); + + if (expectedEnd == ExpectedPaginationEnd::EmptyPage) { + EXPECT_TRUE(sawEmptyTerminator) + << "expected a trailing empty page after the last full page's cursor"; + } else { + EXPECT_FALSE(sawEmptyTerminator) + << "did not expect a trailing empty pagination page"; + } + + // No duplicates. + std::set const seenSet(seen.begin(), seen.end()); + EXPECT_EQ(seen.size(), seenSet.size()) << "pagination returned duplicate rows"; + // No gaps: union equals the full expected set. + EXPECT_EQ(seenSet, expected) << "pagination dropped or repeated rows"; + EXPECT_EQ(seen.size(), expected.size()); + expectSeenInOrder(seen, forward, expected); + }; + + // Case A: total not a multiple of the limit (25 rows, limit 10 -> 10,10,5). + { + auto const expected = setup(mptIssuanceId, 25, baseSeq + 1); + pageThrough(mptIssuanceId, false, 10, expected, ExpectedPaginationEnd::PartialPage); + pageThrough(mptIssuanceId, true, 10, expected, ExpectedPaginationEnd::PartialPage); + } + + // Case B: total is an exact multiple of the limit (20 rows, limit 10). + // The last full page still returns a cursor, so the next fetch must return an empty page + // and end the loop -- no infinite loop and no spurious trailing duplicate. + { + // Use a dedicated issuance id so rows from case A do not bleed in. + ripple::uint192 mptIssuanceIdB; + EXPECT_TRUE( + mptIssuanceIdB.parseHex("00000002BE223A7216F1B07AE9C36F107879B6E9D3A3C1B0") + ); + // Continue the ledger sequence contiguously after case A (which ended at + // baseSeq + 25); the backend's finishWrites enforces contiguous ledgers. + auto const expectedB = setup(mptIssuanceIdB, 20, baseSeq + 26); + + pageThrough(mptIssuanceIdB, false, 10, expectedB, ExpectedPaginationEnd::EmptyPage); + pageThrough(mptIssuanceIdB, true, 10, expectedB, ExpectedPaginationEnd::EmptyPage); + } + }); +} + +TEST_F(BackendCassandraMPTokenIssuanceTest, MissingBlobYieldsInPositionEmptyRecord) +{ + runSpawn([this](boost::asio::yield_context yield) { + auto const mptIssuanceId = makeMptIssuanceId(); + std::uint32_t const seq = 400; + setupLedgerRange(seq); + + // hash 1 and 3 have a Transactions row; hash 2 does NOT (missing blob). + auto const h1 = makeHash(1); + auto const h2 = makeHash(2); + auto const h3 = makeHash(3); + writeTxBlob(h1, seq); + writeTxBlob(h3, seq); + + backend_->writeMPTokenIssuanceTransactions({MPTokenIssuanceTransactionsData{ + .mptIssuanceID = mptIssuanceId, + .accounts = {}, + .ledgerSequence = seq, + .transactionIndex = 1, + .txHash = h1 + }}); + backend_->writeMPTokenIssuanceTransactions({MPTokenIssuanceTransactionsData{ + .mptIssuanceID = mptIssuanceId, + .accounts = {}, + .ledgerSequence = seq, + .transactionIndex = 2, + .txHash = h2 + }}); + backend_->writeMPTokenIssuanceTransactions({MPTokenIssuanceTransactionsData{ + .mptIssuanceID = mptIssuanceId, + .accounts = {}, + .ledgerSequence = seq, + .transactionIndex = 3, + .txHash = h3 + }}); + backend_->waitForWritesToFinish(); + + auto [txns, cursor] = + backend_->fetchMPTokenIssuanceTransactions(mptIssuanceId, 100, false, {}, yield); + // The page is NOT shortened: the missing blob yields an in-position empty record. + ASSERT_EQ(txns.size(), 3); + EXPECT_FALSE(cursor); + + // Reverse order is newest-first: index 3, 2, 1. Middle entry (index 2) is empty. + EXPECT_EQ(txns[1], data::TransactionAndMetadata{}); + EXPECT_NE(txns[0], data::TransactionAndMetadata{}); + EXPECT_NE(txns[2], data::TransactionAndMetadata{}); + }); +} + struct BackendCassandraNodeMessageTest : BackendCassandraTest { - boost::uuids::random_generator generateUuid{}; + static boost::uuids::uuid + generateUuid() + { + return boost::uuids::random_generator{}(); + } }; TEST_F(BackendCassandraNodeMessageTest, UpdateFetch)