diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index e6c4b9c1..ecd8f8af 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -83,7 +83,10 @@ open class NativeCall( casting[BlockchainType.from(event.chain)]?.let { cast -> multistreamHolder.getUpstream(event.chain).let { up -> val reader = up.cast(cast).getReader() - ethereumCallSelectors.putIfAbsent(event.chain, EthereumCallSelector(reader.heightByHash())) + ethereumCallSelectors.putIfAbsent( + event.chain, + EthereumCallSelector(reader.heightByHash(), up.caches) + ) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt index d98e99b7..65ad6d24 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt @@ -71,8 +71,11 @@ class DefaultEthereumMethods( "eth_getBlockTransactionCountByHash", "eth_getUncleCountByBlockHash", "eth_getBlockByHash", + "eth_getBlockByNumber", "eth_getTransactionByHash", "eth_getTransactionByBlockHashAndIndex", + "eth_getTransactionByBlockNumberAndIndex", + "eth_getTransactionReceipt", "eth_getStorageAt", "eth_getCode", "eth_getUncleByBlockHashAndIndex", @@ -89,9 +92,6 @@ class DefaultEthereumMethods( private val headVerifiedMethods = listOf( "eth_getBlockTransactionCountByNumber", "eth_getUncleCountByBlockNumber", - "eth_getBlockByNumber", - "eth_getTransactionByBlockNumberAndIndex", - "eth_getTransactionReceipt", "eth_getUncleByBlockNumberAndIndex", "eth_feeHistory" ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt index 545e5616..5cb354a5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt @@ -16,11 +16,13 @@ package io.emeraldpay.dshackle.upstream.calls import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.etherjar.hex.HexQuantity +import org.bouncycastle.util.encoders.DecoderException import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.util.Collections @@ -31,7 +33,8 @@ import java.util.Objects * The implementation is specific for Ethereum. */ class EthereumCallSelector( - private val heightReader: Reader + private val heightReader: Reader, + private val caches: Caches ) { companion object { @@ -45,6 +48,11 @@ class EthereumCallSelector( // no "eth_getStorageAt" because it has different structure, and therefore separate logic "eth_call" ).sorted() + + private val GET_BY_HASH_OR_NUMBER_METHODS = setOf( + "eth_getBlockByHash", "eth_getBlockByNumber", + "eth_getTransactionByBlockHashAndIndex", "eth_getTransactionByBlockNumberAndIndex" + ) } private val objectMapper = Global.objectMapper @@ -60,6 +68,8 @@ class EthereumCallSelector( return blockTagSelector(params, 2, head) } else if (method in DefaultEthereumMethods.withFilterIdMethods) { return sameUpstreamMatcher(params) + } else if (method in GET_BY_HASH_OR_NUMBER_METHODS) { + return blockMethodSelector(method, params) } return Mono.empty() } @@ -119,6 +129,32 @@ class EthereumCallSelector( } } + private fun blockMethodSelector(method: String, params: String): Mono { + val list = objectMapper.readerFor(Any::class.java).readValues(params).readAll() + if (list.isEmpty()) { + return Mono.empty() + } + val hashOrNumber = Objects.toString(list[0]) + + return when (method) { + "eth_getTransactionByBlockHashAndIndex", "eth_getBlockByHash" -> blockByHashFromCache(hashOrNumber) + "eth_getTransactionByBlockNumberAndIndex", "eth_getBlockByNumber" -> blockByHeight(hashOrNumber) + else -> Mono.empty() + } + } + + private fun blockByHashFromCache(blockHash: String): Mono { + return try { + caches.getBlocksByHash() + .read(BlockId.from(blockHash)) + .onErrorResume { Mono.empty() } + .map { Selector.HeightMatcher(it.height) } + } catch (e: DecoderException) { + log.warn("Invalid blockHash: $blockHash") + Mono.empty() + } + } + private fun blockByHeight(blockNumber: String): Mono { return try { Mono.just(Selector.HeightMatcher(HexQuantity.from(blockNumber).value.longValueExact())) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index a41cb04e..39fac1bc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -66,8 +66,9 @@ class EthereumDirectReader( } blockByHeightReader = object : Reader { override fun read(key: Long): Mono { + val heightMatcher = Selector.HeightMatcher(key) val request = JsonRpcRequest("eth_getBlockByNumber", listOf(HexQuantity.from(key).toHex(), false)) - return readBlock(request, key.toString()) + return readBlock(request, key.toString(), heightMatcher) } } txReader = object : Reader { @@ -143,8 +144,12 @@ class EthereumDirectReader( } @Suppress("UNCHECKED_CAST") - private fun readBlock(request: JsonRpcRequest, id: String): Mono { - return readWithQuorum(request) + private fun readBlock( + request: JsonRpcRequest, + id: String, + matcher: Selector.Matcher = Selector.empty + ): Mono { + return readWithQuorum(request, matcher) .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id"))) .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200))) .flatMap { blockbytes -> @@ -163,11 +168,11 @@ class EthereumDirectReader( /** * Read from an Upstream applying a Quorum specific for that request */ - private fun readWithQuorum(request: JsonRpcRequest): Mono { + private fun readWithQuorum(request: JsonRpcRequest, matcher: Selector.Matcher = Selector.empty): Mono { return Mono.just(quorumReaderFactory) .map { it.create( - up.getApiSource(Selector.empty), + up.getApiSource(matcher), callMethodsFactory.create().createQuorumFor(request.method), // we do not use Signer for internal requests because it doesn't make much sense null diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy index f81ada4c..ca0e360c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy @@ -15,19 +15,25 @@ */ package io.emeraldpay.dshackle.upstream.calls +import io.emeraldpay.dshackle.cache.BlocksMemCache +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector import reactor.core.publisher.Mono +import reactor.test.StepVerifier import spock.lang.Specification +import java.time.Duration +import java.time.Instant class EthereumCallSelectorSpec extends Specification { def "Get height matcher for latest balance"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { 1 * getCurrentHeight() >> 100 } @@ -39,7 +45,7 @@ class EthereumCallSelectorSpec extends Specification { def "Get height matcher for latest call"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { 1 * getCurrentHeight() >> 100 } @@ -51,7 +57,7 @@ class EthereumCallSelectorSpec extends Specification { def "Get height matcher for latest storageAt"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { 1 * getCurrentHeight() >> 100 } @@ -63,7 +69,7 @@ class EthereumCallSelectorSpec extends Specification { def "Get height matcher for balance on block"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { _ * getCurrentHeight() >> 100 } @@ -78,7 +84,7 @@ class EthereumCallSelectorSpec extends Specification { def heights = Mock(Reader) { 1 * it.read(BlockId.from("0xc90f1c8c125a4d5b90742f16947bdb1d10516f173fd7fc51223d10499de2a812")) >> Mono.just(8606722L) } - EthereumCallSelector callSelector = new EthereumCallSelector(heights) + EthereumCallSelector callSelector = new EthereumCallSelector(heights, Stub(Caches)) def head = Mock(Head) { _ * getCurrentHeight() >> 9128116 } @@ -90,7 +96,7 @@ class EthereumCallSelectorSpec extends Specification { def "No matcher for invalid height"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { _ * getCurrentHeight() >> 100 } @@ -103,7 +109,7 @@ class EthereumCallSelectorSpec extends Specification { def "No matcher for negative height"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { _ * getCurrentHeight() >> 100 } @@ -115,7 +121,7 @@ class EthereumCallSelectorSpec extends Specification { def "No matcher for negative long"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { _ * getCurrentHeight() >> 100 } @@ -128,7 +134,7 @@ class EthereumCallSelectorSpec extends Specification { def "No matcher for pending balance"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Mock(Head) { _ * getCurrentHeight() >> 100 } @@ -140,7 +146,7 @@ class EthereumCallSelectorSpec extends Specification { def "Get height matcher with EIP-1898"() { setup: - EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader), Stub(Caches)) def head = Stub(Head) when: def act = callSelector.getMatcher("eth_call", '["0x0000", {"blockNumber": "0x100"}]', head, false).block() @@ -153,7 +159,7 @@ class EthereumCallSelectorSpec extends Specification { def heights = Mock(Reader) { 1 * it.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(12079192L) } - EthereumCallSelector callSelector = new EthereumCallSelector(heights) + EthereumCallSelector callSelector = new EthereumCallSelector(heights, Stub(Caches)) def head = Stub(Head) when: def act = callSelector.getMatcher("eth_call", @@ -168,7 +174,7 @@ class EthereumCallSelectorSpec extends Specification { def heights = Mock(Reader) { 0 * it.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(12079192L) } - EthereumCallSelector callSelector = new EthereumCallSelector(heights) + EthereumCallSelector callSelector = new EthereumCallSelector(heights, Stub(Caches)) def head = Stub(Head) when: def act = callSelector.getMatcher("eth_call", @@ -183,7 +189,7 @@ class EthereumCallSelectorSpec extends Specification { def heights = Mock(Reader) { 1 * it.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty() } - EthereumCallSelector callSelector = new EthereumCallSelector(heights) + EthereumCallSelector callSelector = new EthereumCallSelector(heights, Stub(Caches)) def head = Mock(Head) { 1 * it.getCurrentHeight() >> 100 } @@ -197,7 +203,7 @@ class EthereumCallSelectorSpec extends Specification { def "Get same matcher for getFilterChanges method"() { setup: - def callSelector = new EthereumCallSelector(Mock(Reader)) + def callSelector = new EthereumCallSelector(Mock(Reader), Stub(Caches)) def head = Mock(Head) expect: @@ -214,7 +220,7 @@ class EthereumCallSelectorSpec extends Specification { def "Get empty matcher for getFilterChanges method without params"() { setup: - def callSelector = new EthereumCallSelector(Mock(Reader)) + def callSelector = new EthereumCallSelector(Mock(Reader), Stub(Caches)) def head = Mock(Head) when: @@ -223,4 +229,82 @@ class EthereumCallSelectorSpec extends Specification { then: act == null } + + def "Get height matcher for getByHash and getTransactionByBlockHash methods"() { + setup: + def hash = "0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32" + def block = new BlockContainer( + 12079192L, BlockId.from(hash), + BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0, "upstream" + ) + def blockByHashCache = Mock(BlocksMemCache) { + 1 * read(BlockId.from(hash)) >> Mono.just(block) + } + def cache = Caches.newBuilder().setBlockByHash(blockByHashCache).build() + def callSelector = new EthereumCallSelector(Stub(Reader), cache) + def head = Stub(Head) + + when: + def act = callSelector.getMatcher( + method, '["0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32", false]', + head, false + ) + + then: + StepVerifier.create(act) + .expectNext(new Selector.HeightMatcher(12079192L)) + .expectComplete() + .verify(Duration.ofSeconds(1)) + + where: + method << ["eth_getTransactionByBlockHashAndIndex", "eth_getBlockByHash"] + } + + def "Get height matcher for getByNumber and getTransactionByBlockNumber methods"() { + setup: + def cache = Stub(Caches) + def callSelector = new EthereumCallSelector(Stub(Reader), cache) + def head = Stub(Head) + + when: + def act = callSelector.getMatcher( + method, '["0xfbfe3b", false]', + head, false + ) + + then: + StepVerifier.create(act) + .expectNext(new Selector.HeightMatcher(16514619L)) + .expectComplete() + .verify(Duration.ofSeconds(1)) + + where: + method << ["eth_getTransactionByBlockNumberAndIndex", "eth_getBlockByNumber"] + } + + def "No height matcher for getByHash method"() { + setup: + def hash = "0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32" + def blockByHashCache = Mock(BlocksMemCache) { + 1 * read(BlockId.from(hash)) >> resultFromCache + } + def cache = Caches.newBuilder().setBlockByHash(blockByHashCache).build() + def callSelector = new EthereumCallSelector(Stub(Reader), cache) + def head = Stub(Head) + + when: + def act = callSelector.getMatcher( + "eth_getBlockByHash", '["0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32", false]', + head, false + ) + + then: + StepVerifier.create(act) + .expectNext() + .expectComplete() + .verify(Duration.ofSeconds(1)) + + where: + resultFromCache << [Mono.empty(), Mono.error(new RuntimeException())] + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index 48f55904..2232caba 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -10,6 +10,7 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.ApiSource import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.Chain @@ -108,7 +109,7 @@ class EthereumDirectReaderSpec extends Specification { transactions = [] } def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) + 1 * getApiSource(new Selector.HeightMatcher(100)) >> Stub(ApiSource) } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)