rpcdaemon: erigon_getLatestLogs for e3 (#2611)

.github/workflows/rpc-integration-tests.yml

@@ -33,7 +33,7 @@ jobs:
       - name: Checkout RPC Tests Repository & Install Requirements
         run: |
           rm -rf ${{runner.workspace}}/rpc-tests
-          git -c advice.detachedHead=false clone --depth 1 --branch v1.26.0 https://github.com/erigontech/rpc-tests ${{runner.workspace}}/rpc-tests
+          git -c advice.detachedHead=false clone --depth 1 --branch v1.27.0 https://github.com/erigontech/rpc-tests ${{runner.workspace}}/rpc-tests
           cd ${{runner.workspace}}/rpc-tests
           pip3 install -r requirements.txt --break-system-packages
 

silkworm/core/common/util.hpp

@@ -58,6 +58,11 @@ inline bool is_valid_hex(std::string_view s) {
     return std::regex_match(s.data(), kHexRegex);
 }
 
+inline bool is_valid_dec(std::string_view s) {
+    static const std::regex kHexRegex("^[0-9]+$");
+    return std::regex_match(s.data(), kHexRegex);
+}
+
 inline bool is_valid_hash(std::string_view s) {
     if (s.length() != 2 + kHashLength * 2) {
         return false;

silkworm/rpc/commands/erigon_api.cpp

@@ -331,10 +331,12 @@ Task<void> ErigonRpcApi::handle_erigon_get_latest_logs(const nlohmann::json& req
         co_return;
     }
 
-    LogFilterOptions options{true};
+    LogFilterOptions options{true, true};
     if (params.size() > 1) {
         options = params[1].get<LogFilterOptions>();
+
         options.add_timestamp = true;
+        options.overwrite_log_index = true;
     }
 
     if (options.log_count != 0 && options.block_count != 0) {
@@ -368,6 +370,7 @@ Task<void> ErigonRpcApi::handle_erigon_get_latest_logs(const nlohmann::json& req
             co_await tx->close();  // RAII not (yet) available with coroutines
             co_return;
         }
+        SILK_DEBUG << "start: " << start << " end: " << end;
 
         std::vector<Log> logs;
         co_await logs_walker.get_logs(start, end, filter.addresses, filter.topics, options, true, logs);

silkworm/rpc/core/block_reader.cpp

@@ -150,6 +150,9 @@ Task<std::pair<BlockNum, bool>> BlockReader::get_block_num(const std::string& bl
     } else if (is_valid_hex(block_id)) {
         block_num = static_cast<BlockNum>(std::stol(block_id, nullptr, 16));
         check_if_latest = latest_required;
+    } else if (is_valid_dec(block_id)) {
+        block_num = static_cast<BlockNum>(std::stol(block_id, nullptr, 10));
+        check_if_latest = latest_required;
     } else {
         throw std::invalid_argument("get_block_num::Invalid Block Id");
     }

silkworm/rpc/core/block_reader_test.cpp

@@ -152,7 +152,8 @@ TEST_CASE("get_block_num latest_required", "[rpc][core][blocks]") {
     SECTION("block_num in dec") {
         static const std::string kBlockIdDec = "67890";
         auto result = boost::asio::co_spawn(pool, block_reader.get_block_num(kBlockIdDec, /*latest_required=*/false), boost::asio::use_future);
-        REQUIRE_THROWS(result.get());
+        auto [block_num, ignore] = result.get();
+        CHECK(block_num == 67890);
     }
 
     SECTION("block_num in hex & latest true") {

silkworm/rpc/core/logs_walker.cpp

@@ -28,7 +28,6 @@
 #include <silkworm/infra/common/log.hpp>
 #include <silkworm/rpc/core/cached_chain.hpp>
 #include <silkworm/rpc/core/receipts.hpp>
-#include <silkworm/rpc/ethdb/bitmap.hpp>
 #include <silkworm/rpc/ethdb/cbor.hpp>
 #include <silkworm/rpc/ethdb/kv/backend_providers.hpp>
 
@@ -93,6 +92,11 @@ Task<void> LogsWalker::get_logs(BlockNum start,
     db::kv::api::PaginatedStream<db::kv::api::Timestamp> paginated_stream;
     if (!topics.empty()) {
         for (auto sub_topic = topics.begin(); sub_topic < topics.end(); ++sub_topic) {
+            if (sub_topic->empty()) {
+                continue;
+            }
+
+            db::kv::api::PaginatedStream<db::kv::api::Timestamp> union_stream;
             for (auto it = sub_topic->begin(); it < sub_topic->end(); ++it) {
                 SILK_DEBUG << "topic: " << to_hex(*it) << ", from_timestamp: " << from_timestamp << ", to_timestamp: "
                            << to_timestamp;
@@ -104,8 +108,13 @@ Task<void> LogsWalker::get_logs(BlockNum start,
                     .to_timestamp = to_timestamp,
                     .ascending_order = asc_order};
                 auto paginated_result = co_await tx_.index_range(std::move(query));
-                paginated_stream = db::kv::api::set_union(std::move(paginated_stream), co_await paginated_result.begin());
+                union_stream = db::kv::api::set_union(std::move(union_stream), co_await paginated_result.begin());
+            }
+            if (!paginated_stream) {
+                paginated_stream = std::move(union_stream);
+                continue;
             }
+            paginated_stream = db::kv::api::set_intersection(std::move(paginated_stream), std::move(union_stream));
         }
     }
     if (!addresses.empty()) {
@@ -139,6 +148,7 @@ Task<void> LogsWalker::get_logs(BlockNum start,
     uint64_t log_count{0};
     Logs filtered_chunk_logs;
 
+    uint64_t block_timestamp;
     auto itr = db::txn::make_txn_nums_stream(std::move(paginated_stream), asc_order, tx_, provider);
     while (const auto tnx_nums = co_await itr->next()) {
         if (tnx_nums->final_txn) {
@@ -152,10 +162,9 @@ Task<void> LogsWalker::get_logs(BlockNum start,
                 SILK_DEBUG << "Not found block no.  " << tnx_nums->block_num;
                 break;
             }
+            block_timestamp = block_with_hash->block.header.timestamp;
             receipts = co_await core::get_receipts(tx_, *block_with_hash, *chain_storage, workers_);
             SILK_DEBUG << "Read #" << receipts.size() << " receipts from block " << tnx_nums->block_num;
-
-            ++block_count;
         }
         const auto transaction = co_await chain_storage->read_transaction_by_idx_in_block(tnx_nums->block_num, tnx_nums->txn_index);
         if (!transaction) {
@@ -166,16 +175,32 @@ Task<void> LogsWalker::get_logs(BlockNum start,
         SILK_DEBUG << "Got transaction: block_num: " << tnx_nums->block_num << ", txn_index: " << tnx_nums->txn_index;
 
         SILKWORM_ASSERT(tnx_nums->txn_index < receipts.size());
-        const auto& receipt = receipts.at(tnx_nums->txn_index);
+        auto& receipt = receipts.at(tnx_nums->txn_index);
 
-        SILK_DEBUG << "#rawLogs: " << receipt.logs.size();
+        // ERIGON3 compatibility: erigon_getLatestLogs overwrites log index
+        if (options.overwrite_log_index) {
+            uint32_t log_index{0};
+            for (auto& log : receipt.logs) {
+                log.index = log_index++;
+            }
+        }
+
+        SILK_DEBUG << "blockNum: " << tnx_nums->block_num << ", #rawLogs: " << receipt.logs.size();
         filtered_chunk_logs.clear();
         filter_logs(receipt.logs, addresses, topics, filtered_chunk_logs, options.log_count == 0 ? 0 : options.log_count - log_count);
         SILK_DEBUG << "filtered #logs: " << filtered_chunk_logs.size();
-
+        if (filtered_chunk_logs.empty()) {
+            continue;
+        }
+        ++block_count;
         log_count += filtered_chunk_logs.size();
         SILK_DEBUG << "log_count: " << log_count;
 
+        if (options.add_timestamp) {
+            for (auto& log : filtered_chunk_logs) {
+                log.timestamp = block_timestamp;
+            }
+        }
         logs.insert(logs.end(), filtered_chunk_logs.begin(), filtered_chunk_logs.end());
 
         if (options.log_count != 0 && options.log_count <= log_count) {

silkworm/rpc/types/filter.hpp

@@ -39,6 +39,7 @@ struct Filter {
 
 struct LogFilterOptions {
     bool add_timestamp{false};
+    bool overwrite_log_index{false};
     std::uint64_t log_count{0};
     std::uint64_t block_count{0};
     bool ignore_topics_order{false};