Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

rpcdaemon: ots_searchTransactionsBefore for e3 #2498

Draft
wants to merge 9 commits into
base: master
Choose a base branch
from
+308 −40
Draft

rpcdaemon: ots_searchTransactionsBefore for e3 #2498

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
4901ee8
Step 1
Sixtysixter Nov 4, 2024
d418e46
step 2
Sixtysixter Nov 8, 2024
6fd0fd1
Step 2
Sixtysixter Nov 10, 2024
93f08cb
step 2
Sixtysixter Nov 8, 2024
1708a37
Step 2
Sixtysixter Nov 10, 2024
19e51d7
make fmt
web-flow Nov 10, 2024
58e4334
Review
Sixtysixter Nov 12, 2024
be31743
make fmt
web-flow Nov 12, 2024
b58dd89
Added unit test
Sixtysixter Nov 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions silkworm/db/kv/api/endpoint/paginated_sequence_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,12 @@ TEST_CASE_METHOD(PaginatedSequenceTest, "set_intersection", "[db][kv][api][pagin
TEST_CASE_METHOD(PaginatedSequenceTest, "set_union", "[db][kv][api][paginated_sequence]") {
const Fixtures<std::pair<PageUint64List, PageUint64List>, std::vector<uint64_t>> fixtures{
{{/*v1=*/{}, /*v2=*/{}}, /*v1_or_v2=*/{}}, // both empty => empty
{{/*v1=*/{}, /*v2=*/{{1}}}, /*v1_or_v2=*/{1}}, // one empty => other
{{/*v1=*/{{1}}, /*v2=*/{}}, /*v1_or_v2=*/{1}}, // one empty => other
{{/*v1=*/{{1, 2, 3}, {4, 5, 6}, {7, 8}}, /*v2=*/{}}, /*v1_or_v2=*/{1, 2, 3, 4, 5, 6, 7, 8}}, // one empty => other
{{/*v1=*/{{1, 3, 5}}, /*v2=*/{{2, 4, 6}}}, /*v1_or_v2=*/{1, 2, 3, 4, 5, 6}},
{{/*v1=*/{{1, 3, 5, 7}}, /*v2=*/{{1, 4, 6, 7}}}, /*v1_or_v2=*/{1, 3, 4, 5, 6, 7}},
{{/*v1=*/{{1, 2, 3, 4}}, /*v2=*/{{1, 2, 3, 5}}}, /*v1_or_v2=*/{1, 2, 3, 4, 5}},
{{/*v1=*/{{1, 2, 3}, {4, 5, 6}, {7, 8}}, /*v2=*/{{10, 11, 12}, {13}}}, /*v1_or_v2=*/{1, 2, 3, 4, 5, 6, 7, 8, 10, 11, 12, 13}},
{{/*v1=*/{{1, 2, 3}, {4, 5, 6}, {7, 8}}, /*v2=*/{{7, 8, 9}, {10, 11, 12}, {13}}}, /*v1_or_v2=*/{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13}},
{{/*v1=*/{{1, 2, 3}, {4, 5, 6}, {7, 8}}, /*v2=*/{{1, 2, 3}, {4, 5, 6}, {7, 8}}}, /*v1_and_v2=*/{1, 2, 3, 4, 5, 6, 7, 8}},
Expand Down
6 changes: 6 additions & 0 deletions silkworm/db/tables.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,4 +436,10 @@ inline constexpr const char* kAccountsHistory{"AccountsHistory"};
//! \details Inverted Index storing the account common information
inline constexpr const char* kAccountsHistoryIdx{"AccountsHistoryIdx"};

//! \details TODO
inline constexpr const char* kTracesFromIdx{"TracesFromIdx"};

//! \details TODO
inline constexpr const char* kTracesToIdx{"TracesToIdx"};

} // namespace silkworm::db::table
326 changes: 286 additions & 40 deletions silkworm/rpc/commands/ots_api.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -681,6 +681,11 @@ Task<void> OtsRpcApi::handle_ots_get_internal_operations(const nlohmann::json& r
co_await tx->close(); // RAII not (yet) available with coroutines
}

struct BlockInfo {
uint64_t number{0};
BlockDetails details;
};

Task<void> OtsRpcApi::handle_ots_search_transactions_before(const nlohmann::json& request, nlohmann::json& reply) {
const auto& params = request["params"];
if (params.size() != 3) {
Expand All @@ -694,7 +699,7 @@ Task<void> OtsRpcApi::handle_ots_search_transactions_before(const nlohmann::json
auto block_number = params[1].get<BlockNum>();
const auto page_size = params[2].get<uint64_t>();

SILK_DEBUG << "address: " << address << " block_number: " << block_number << " page_size: " << page_size;
SILK_LOG << "address: " << address << " block_number: " << block_number << " page_size: " << page_size;

if (page_size > kMaxPageSize) {
auto error_msg = "max allowed page size: " + std::to_string(kMaxPageSize);
Expand All @@ -703,52 +708,144 @@ Task<void> OtsRpcApi::handle_ots_search_transactions_before(const nlohmann::json
co_return;
}

if (block_number > 0) {
--block_number;
}
auto tx = co_await database_->begin();

try {
auto call_from_cursor = co_await tx->cursor(table::kCallFromIndexName);
auto call_to_cursor = co_await tx->cursor(table::kCallToIndexName);

bool is_first_page = false;
auto provider = ethdb::kv::canonical_body_for_storage_provider(backend_);

if (block_number == 0) {
is_first_page = true;
} else {
// Internal search code considers blockNum [including], so adjust the value
--block_number;
db::kv::api::Timestamp from_timestamp{-1};
if (block_number > 0) {
const auto max_tx_num = co_await db::txn::max_tx_num(*tx, block_number, provider);
from_timestamp = static_cast<db::kv::api::Timestamp>(max_tx_num);
SILK_LOG << "block_number: " << block_number << " max_tx_num: " << max_tx_num;
}

BackwardBlockProvider from_provider{call_from_cursor.get(), address, block_number};
BackwardBlockProvider to_provider{call_to_cursor.get(), address, block_number};
FromToBlockProvider from_to_provider{false, &from_provider, &to_provider};

uint64_t result_count = 0;
bool has_more = true;
const auto results = co_await collect_transactions_with_receipts(*tx, provider, block_number, address, from_timestamp, false, page_size);

// const auto min_txn_num = co_await db::txn::min_tx_num(*tx, block_number, provider);
// SILK_LOG << "block_number: " << block_number << " max_tx_num: " << max_tx_num << " min_txn_num: " << min_txn_num;

// db::kv::api::IndexRangeQuery query_to{
// .table = db::table::kTracesToIdx,
// .key = key,
// .from_timestamp = from_timestamp,
// .to_timestamp = -1,
// .ascending_order = false};
// auto paginated_result_to = co_await tx->index_range(std::move(query_to));
//
// SILK_LOG << "TEST *************************************";
// auto it_to1 = co_await paginated_result_to.begin();
// auto cc{0};
// while (const auto value = co_await it_to1.next()) {
// const auto txn_id = static_cast<TxnId>(*value);
// SILK_LOG << "ITERATE: " << cc++ << ", txn_id: " << txn_id;
// if (cc > 10) {
// break;
// }
// }
//
// db::kv::api::IndexRangeQuery query_from{
// .table = db::table::kTracesFromIdx,
// .key = key,
// .from_timestamp = from_timestamp,
// .to_timestamp = -1,
// .ascending_order = false};
// auto paginated_result_from = co_await tx->index_range(std::move(query_from));
// SILK_LOG << "********************************** size: " << resultssss.transactions.size();
//
// auto it_to = co_await paginated_result_to.begin();
// auto it_from = co_await paginated_result_from.begin();
// TransactionsWithReceipts results{
// .first_page = block_number == 0,
// .last_page = true};
//
// std::map<std::string, Receipt> receipts;
// std::optional<BlockInfo> block_info;
// auto block_num_changed = false;
// auto it = db::kv::api::set_union(it_from, it_to, false);
// const auto chain_storage = tx->create_storage();
// auto count{0};
// while (const auto value = co_await it.next()) {
// if (!value.has_value()) {
// SILK_LOG << "NO VALUE FROM ITR";
// break;
// }
// const auto txn_id = static_cast<TxnId>(*value);
// SILK_LOG << "ITERATE: " << count++ << ", txn_id: " << txn_id;
// const auto block_number_opt = co_await db::txn::block_num_from_tx_num(*tx, txn_id, provider);
// if (!block_number_opt) {
// SILK_LOG << "No block found for txn_id " << txn_id;
// break; // TODO
// }
// const auto bn = block_number_opt.value();
// const auto max_txn_id = co_await db::txn::max_tx_num(*tx, bn, provider);
// const auto min_txn_id = co_await db::txn::min_tx_num(*tx, bn, provider);
// const auto txn_index = txn_id - min_txn_id - 1;
// SILK_LOG
// << "txn_id: " << txn_id
// << " block_number: " << bn
// << ", txn_index: " << txn_index
// << ", max_txn_id: " << max_txn_id
// << ", min_txn_id: " << min_txn_id
// << ", final txn: " << (txn_id == max_txn_id);
//
// if (txn_id == max_txn_id) {
// continue;
// }
//
// block_num_changed = false;
// if (block_info && block_info.value().number != bn) {
// block_info.reset();
// }
//
// if (!block_info) {
// const auto block_with_hash = co_await rpc::core::read_block_by_number(*block_cache_, *chain_storage, bn);
// if (!block_with_hash) {
// SILK_DEBUG << "Not found block no. " << bn;
// co_return;
// }
//
// auto rr = co_await core::get_receipts(*tx, *block_with_hash, *chain_storage, workers_);
// SILK_LOG << "Read #" << rr.size() << " receipts from block " << bn;
//
// std::for_each(rr.begin(), rr.end(), [&receipts](const auto& item) {
// receipts[silkworm::to_hex(item.tx_hash, false)] = std::move(item);
// });
//
// const Block extended_block{block_with_hash, false};
// const auto block_size = extended_block.get_block_size();
// const BlockDetails block_details{block_size, block_with_hash->hash, block_with_hash->block.header,
// block_with_hash->block.transactions.size(), block_with_hash->block.ommers,
// block_with_hash->block.withdrawals};
// block_info = BlockInfo{block_with_hash->block.header.number, block_details};
// block_num_changed = true;
// }
//
// if (results.transactions.size() >= page_size && block_num_changed) {
// results.last_page = false;
// break;
// }
//
// auto transaction = co_await chain_storage->read_transaction_by_idx_in_block(bn, txn_index);
// if (!transaction) {
// SILK_LOG << "No transaction found in block " << bn << " for index " << txn_index;
// co_return;
// }
// results.receipts.push_back(std::move(receipts.at(silkworm::to_hex(transaction.value().hash(), false))));
// results.transactions.push_back(std::move(transaction.value()));
// results.blocks.push_back(block_info.value().details);
//
// SILK_LOG << "PageSie " << page_size << ", result size: " << results.transactions.size();
// }
//
// SILK_LOG << "Results"
// << " transactions size: " << results.transactions.size()
// << " receipts size: " << results.receipts.size()
// << " block details size: " << results.blocks.size();

TransactionsWithReceipts results{
.first_page = is_first_page};

while (result_count < page_size && has_more) {
std::vector<TransactionsWithReceipts> transactions_with_receipts_vec;

has_more = co_await trace_blocks(from_to_provider, *tx, address, page_size, result_count, transactions_with_receipts_vec);

for (const auto& item : transactions_with_receipts_vec) {
results.receipts.insert(results.receipts.end(), item.receipts.rbegin(), item.receipts.rend());
results.transactions.insert(results.transactions.end(), item.transactions.rbegin(), item.transactions.rend());
results.blocks.insert(results.blocks.end(), item.blocks.rbegin(), item.blocks.rend());

result_count += item.transactions.size();

if (result_count >= page_size) {
break;
}
}
}

results.last_page = !has_more;
reply = make_json_content(request, results);

} catch (const std::invalid_argument& iv) {
SILK_WARN << "invalid_argument: " << iv.what() << " processing request: " << request.dump();
reply = make_json_content(request, nlohmann::detail::value_t::null);
Expand Down Expand Up @@ -850,6 +947,155 @@ Task<void> OtsRpcApi::handle_ots_search_transactions_after(const nlohmann::json&
co_await tx->close(); // RAII not (yet) available with coroutines
}

Task<TransactionsWithReceipts> OtsRpcApi::collect_transactions_with_receipts(
kv::api::Transaction& tx,
db::chain::CanonicalBodyForStorageProvider& provider,
BlockNum block_number,
const evmc::address& address,
db::kv::api::Timestamp from_timestamp,
bool ascending, uint64_t page_size) {
const auto key = db::code_domain_key(address);
db::kv::api::IndexRangeQuery query_to{
.table = db::table::kTracesToIdx,
.key = key,
.from_timestamp = from_timestamp,
.to_timestamp = -1,
.ascending_order = ascending};
auto paginated_result_to = co_await tx.index_range(std::move(query_to));

SILK_LOG << "TEST *************************************";
auto it_to1 = co_await paginated_result_to.begin();
auto cc{0};
while (const auto value = co_await it_to1.next()) {
const auto txn_id = static_cast<TxnId>(*value);
SILK_LOG << "ITERATE: " << cc++ << ", txn_id: " << txn_id;
if (cc > 10) {
break;
}
}

db::kv::api::IndexRangeQuery query_from{
.table = db::table::kTracesFromIdx,
.key = key,
.from_timestamp = from_timestamp,
.to_timestamp = -1,
.ascending_order = ascending};
auto paginated_result_from = co_await tx.index_range(std::move(query_from));

cc = 0;
auto it_from1 = co_await paginated_result_from.begin();
while (const auto value = co_await it_from1.next()) {
const auto txn_id = static_cast<TxnId>(*value);
SILK_LOG << "ITERATE: " << cc++ << ", txn_id: " << txn_id;
if (cc > 10) {
break;
}
}

SILK_LOG << "collect_transactions_with_receipts **************************************";
auto it_from = co_await paginated_result_from.begin();
auto it_to = co_await paginated_result_to.begin();

// SILK_LOG << "TEST *************************************";
// auto it_to1 = co_await paginated_result_to.begin();
// auto cc{0};
// while (const auto value = co_await it_to1.next()) {
// const auto txn_id = static_cast<TxnId>(*value);
// SILK_LOG << "ITERATE: " << cc++ << ", txn_id: " << txn_id;
// if (cc > 10) {
// break;
// }
// }

TransactionsWithReceipts results{
.first_page = block_number == 0,
.last_page = true};

std::map<std::string, Receipt> receipts;
std::optional<BlockInfo> block_info;
auto block_num_changed = false;

auto it = db::kv::api::set_union(it_from, it_to, ascending);
const auto chain_storage = tx.create_storage();
auto count{0};
while (const auto value = co_await it.next()) {
const auto txn_id = static_cast<TxnId>(*value);
SILK_LOG << "ITERATE: " << count++ << ", txn_id: " << txn_id;
const auto block_number_opt = co_await db::txn::block_num_from_tx_num(tx, txn_id, provider);
if (!block_number_opt) {
SILK_LOG << "No block found for txn_id " << txn_id;
break; // TODO
}
const auto bn = block_number_opt.value();
const auto max_txn_id = co_await db::txn::max_tx_num(tx, bn, provider);
const auto min_txn_id = co_await db::txn::min_tx_num(tx, bn, provider);
const auto txn_index = txn_id - min_txn_id - 1;
SILK_LOG
<< "txn_id: " << txn_id
<< " block_number: " << bn
<< ", txn_index: " << txn_index
<< ", max_txn_id: " << max_txn_id
<< ", min_txn_id: " << min_txn_id
<< ", final txn: " << (txn_id == max_txn_id)
<< ", ascending: " << ascending;

if (txn_id == max_txn_id) {
continue;
}

block_num_changed = false;
if (block_info && block_info.value().number != bn) {
block_info.reset();
}

if (!block_info) {
const auto block_with_hash = co_await rpc::core::read_block_by_number(*block_cache_, *chain_storage, bn);
if (!block_with_hash) {
SILK_DEBUG << "Not found block no. " << bn;
co_return results;
}

auto rr = co_await core::get_receipts(tx, *block_with_hash, *chain_storage, workers_);
SILK_LOG << "Read #" << rr.size() << " receipts from block " << bn;

std::for_each(rr.begin(), rr.end(), [&receipts](const auto& item) {
receipts[silkworm::to_hex(item.tx_hash, false)] = std::move(item);
});

const Block extended_block{block_with_hash, false};
const auto block_size = extended_block.get_block_size();
const BlockDetails block_details{block_size, block_with_hash->hash, block_with_hash->block.header,
block_with_hash->block.transactions.size(), block_with_hash->block.ommers,
block_with_hash->block.withdrawals};
block_info = BlockInfo{block_with_hash->block.header.number, block_details};
block_num_changed = true;
}

if (results.transactions.size() >= page_size && block_num_changed) {
results.last_page = false;
break;
}

auto transaction = co_await chain_storage->read_transaction_by_idx_in_block(bn, txn_index);
if (!transaction) {
SILK_LOG << "No transaction found in block " << bn << " for index " << txn_index;
co_return results;
}
results.receipts.push_back(std::move(receipts.at(silkworm::to_hex(transaction.value().hash(), false))));
results.transactions.push_back(std::move(transaction.value()));
results.blocks.push_back(block_info.value().details);

SILK_LOG << "PageSie " << page_size << ", result size: " << results.transactions.size();
}

SILK_LOG << "Results"
<< " transactions size: " << results.transactions.size()
<< " receipts size: " << results.receipts.size()
<< " block details size: " << results.blocks.size();

co_return results;
}

Task<bool> OtsRpcApi::trace_blocks(
FromToBlockProvider& from_to_provider,
kv::api::Transaction& tx,
Expand Down
Loading
Loading