ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2026-07-24 07:30:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add storage and backend for mpt_issuance_history (#3091)

Browse Source 
Why
This is the first PR in the broader `mptoken_issuance_history` plan. It
lands the additive schema/backend layer first so later PRs can safely
add live ETL indexing, historical backfill, and finally the RPC handler
gated on backfill completion. Keeping this step dark avoids exposing
partial history while still making the stack reviewable in small pieces.

Summary
Adds the storage and backend primitives for MPT transaction-history
filtering.

- Adds Cassandra tables:

  - mpt_transactions

  - account_mpt_transactions
- Adds MPTTransactionsData and backend interface methods for
writing/fetching both MPT index shapes.
Implements Cassandra write/fetch paths with forward/reverse pagination,
raw-page cursor behavior, and case-insensitive tx_type filtering before
transaction hydration.
- Adds MockBackend coverage for the new pure-virtual methods.
- Adds Cassandra integration tests for round-trip behavior, ordering,
pagination, missing transaction blobs, sparse tx_type filtering, account
fanout, and account-side filtering.
This commit is contained in:
Bryan
2026-06-16 15:15:45 -04:00
committed by GitHub
parent 0aa7ed4919
commit e24216c8a9
6 changed files with 858 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
src/data/BackendInterface.hpp
View File

@@ -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<TransactionsCursor> 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<TransactionsCursor> 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<NFTTransactionsData> 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<MPTokenIssuanceTransactionsData> 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<MPTokenIssuanceTransactionsData> const& data
) = 0;
/**
* @brief Write accounts that started holding onto a MPT.
*

14
src/data/DBHelpers.hpp
View File

@@ -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<ripple::AccountID> accounts;
std::uint32_t ledgerSequence{};
std::uint32_t transactionIndex{};
ripple::uint256 txHash;
};
/**
* @brief Check whether the supplied object is a dir node.
*

205
src/data/cassandra/CassandraBackendFamily.hpp
View File

@@ -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<std::uint32_t>::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<std::uint32_t>::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<TransactionsCursor> 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<TransactionsCursor> 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<MPTokenIssuanceTransactionsData> const& data
) override
{
std::vector<Statement> 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<MPTokenIssuanceTransactionsData> const& data
) override
{
std::size_t numStatements = 0u;
for (auto const& record : data)
numStatements += record.accounts.size();
std::vector<Statement> 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<TransactionsCursor> 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<std::uint32_t>::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<ripple::uint256> hashes = {};
auto numRows = results.numRows();
for (auto const& [hash, data] :
extract<ripple::uint256, std::tuple<uint32_t, uint32_t>>(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

137
src/data/cassandra/Schema.hpp
View File

@@ -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<bigint, bigint>,
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<bigint, bigint>,
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(