From a4f0d441052db732d1f78b789e5a8ef8c307e831 Mon Sep 17 00:00:00 2001 From: Termina1 Date: Fri, 9 Dec 2022 23:04:50 +0200 Subject: [PATCH] changes from upstream: removed RpcReader and use better flow for tx receipt --- .../io/emeraldpay/dshackle/cache/Caches.kt | 2 +- .../dshackle/data/DefaultContainer.kt | 8 +- .../emeraldpay/dshackle/reader/RpcReader.kt | 54 ---- ...reumReader.kt => EthereumCachingReader.kt} | 12 +- .../upstream/ethereum/EthereumDirectReader.kt | 31 ++ .../upstream/ethereum/EthereumFees.kt | 2 +- .../upstream/ethereum/EthereumLegacyFees.kt | 2 +- .../ethereum/EthereumLikeMultistream.kt | 2 +- .../upstream/ethereum/EthereumMultistream.kt | 5 +- .../upstream/ethereum/EthereumPriorityFees.kt | 2 +- .../upstream/ethereum/LocalCallRouter.kt | 4 +- .../ethereum_pos/EthereumPosMultiStream.kt | 4 +- .../dshackle/cache/CachesSpec.groovy | 2 +- .../test/MultistreamHolderMock.groovy | 7 +- ...roovy => EthereumCachingReaderSpec.groovy} | 73 ++++- .../ethereum/EthereumLegacyFeesSpec.groovy | 2 +- .../ethereum/EthereumPriorityFeesSpec.groovy | 8 +- .../ethereum/EthereumReaderSpec.groovy | 281 ------------------ .../ethereum/LocalCallRouterSpec.groovy | 10 +- 19 files changed, 139 insertions(+), 372 deletions(-) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumReader.kt => EthereumCachingReader.kt} (93%) rename src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/{EthereumDirectReaderSpec.groovy => EthereumCachingReaderSpec.groovy} (75%) delete mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt index 8a5b71db..9cf9566d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -92,7 +92,7 @@ open class Caches( open fun cacheReceipt(tag: Tag, data: DefaultContainer) { val currentHeight = head?.getCurrentHeight() if (currentHeight != null && data.height != null && memReceipts.acceptsRecentBlocks(currentHeight - data.height)) { - memReceipts.add(data) + memReceipts.add(data).subscribe() } // TODO move subscription to the caller redisReceipts?.add(data)?.subscribe() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt index 492ece1b..82bc70a7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt @@ -18,11 +18,11 @@ package io.emeraldpay.dshackle.data import org.slf4j.LoggerFactory class DefaultContainer( - val txId: TxId?, - val blockId: BlockId?, - val height: Long?, + val txId: TxId? = null, + val blockId: BlockId? = null, + val height: Long? = null, json: ByteArray, - parsed: T + parsed: T? = null ) : SourceContainer(json, parsed) { companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt deleted file mode 100644 index cd920fb9..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt +++ /dev/null @@ -1,54 +0,0 @@ -/** - * Copyright (c) 2021 EmeraldPay, Inc - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.reader - -import io.emeraldpay.dshackle.upstream.Multistream -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import org.slf4j.LoggerFactory -import reactor.core.publisher.Mono - -/** - * Reader that requests data through upstream RPC using provided JSON RPC request builder - */ -class RpcReader( - private val up: Multistream, - private val paramsBuilder: (T) -> JsonRpcRequest -) : Reader { - - companion object { - private val log = LoggerFactory.getLogger(RpcReader::class.java) - - /** - * Common reader that just passes key as a parameter with the specified method. The key must be serializable to JSON. - * @param method RPC method to use - */ - fun basicRequest(up: Multistream, method: String): RpcReader { - return RpcReader(up) { key -> - JsonRpcRequest(method, listOf(key)) - } - } - } - - override fun read(key: T): Mono { - return up.getDirectApi(Selector.empty) - .flatMap { rdr -> - rdr.read(paramsBuilder(key)).flatMap { - it.requireResult() - } - } - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt similarity index 93% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt index 6f3ea506..ac77713a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt @@ -28,7 +28,6 @@ import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.CompoundReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.RekeyingReader -import io.emeraldpay.dshackle.reader.RpcReader import io.emeraldpay.dshackle.reader.TransformingReader import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Multistream @@ -45,16 +44,16 @@ import org.slf4j.LoggerFactory import java.util.function.Function /** - * Reader for the common operations, that wraps caches + native call with quorum verification + * Reader for the common operations, that use the cache when data is available or a native call with quorum verification */ -open class EthereumReader( +open class EthereumCachingReader( private val up: Multistream, private val caches: Caches, private val callMethodsFactory: Factory ) : Lifecycle { companion object { - private val log = LoggerFactory.getLogger(EthereumReader::class.java) + private val log = LoggerFactory.getLogger(EthereumCachingReader::class.java) } private val objectMapper: ObjectMapper = Global.objectMapper @@ -157,10 +156,9 @@ open class EthereumReader( } fun receipts(): Reader { - // TODO put into cache val requested = RekeyingReader( - { txid: TxId -> txid.toHexWithPrefix() }, - RpcReader.basicRequest(up, "eth_getTransactionReceipt") + { txid: TxId -> TransactionId.from(txid.value) }, + directReader.receiptReader ) return CompoundReader( caches.getReceipts(), 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 429c135c..25fbf56e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -6,7 +6,10 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.DefaultContainer import io.emeraldpay.dshackle.data.TxContainer +import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Multistream @@ -22,6 +25,7 @@ import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson +import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.apache.commons.collections4.Factory import org.slf4j.LoggerFactory @@ -51,6 +55,7 @@ class EthereumDirectReader( val blockByHeightReader: Reader val txReader: Reader val balanceReader: Reader + val receiptReader: Reader init { blockReader = object : Reader { @@ -107,6 +112,32 @@ class EthereumDirectReader( } } } + receiptReader = object : Reader { + override fun read(key: TransactionId): Mono { + val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex())) + return readWithQuorum(request) + .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Receipt not read $key"))) + .doOnNext { json -> + try { + // Caching needs some additional data (ex. Height) to make a decision on how long and where to cache + // So we have to parse the JSON here and extract reference data + val receipt = objectMapper.readValue(json, TransactionReceiptJson::class.java) + caches.cacheReceipt( + Caches.Tag.REQUESTED, + DefaultContainer( + txId = TxId.from(key), + blockId = BlockId.from(receipt.blockHash), + height = receipt.blockNumber, + json = json, + parsed = receipt + ) + ) + } catch (t: Throwable) { + log.warn("Failed to cache Tx Receipt", t) + } + } + } + } } @Suppress("UNCHECKED_CAST") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt index 0cf7c42c..b6f935e2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt @@ -30,7 +30,7 @@ import java.util.function.Function abstract class EthereumFees( upstreams: Multistream, - private val reader: EthereumReader, + private val reader: EthereumCachingReader, heightLimit: Int, ) : AbstractChainFees, TransactionRefJson, TransactionJson>(heightLimit, upstreams, extractTx), ChainFees { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFees.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFees.kt index 7884e248..0afa5c6a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFees.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFees.kt @@ -23,7 +23,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import java.util.function.Function -class EthereumLegacyFees(upstreams: EthereumMultistream, reader: EthereumReader, heightLimit: Int) : +class EthereumLegacyFees(upstreams: EthereumMultistream, reader: EthereumCachingReader, heightLimit: Int) : EthereumFees(upstreams, reader, heightLimit) { companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt index d049fe43..552ad27c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt @@ -7,7 +7,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import reactor.core.publisher.Flux interface EthereumLikeMultistream : Upstream { - fun getReader(): EthereumReader + fun getReader(): EthereumCachingReader fun getSubscribe(): EthereumSubscribe fun getHead(mather: Selector.Matcher): Head diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 899d5358..a1c984a7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -47,8 +47,9 @@ open class EthereumMultistream( private val filteredHeads: MutableMap = ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK) - private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory()) + private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory()) private val subscribe = EthereumSubscribe(this) + private val supportsEIP1559 = when (chain) { Chain.ETHEREUM, Chain.TESTNET_ROPSTEN, Chain.TESTNET_GOERLI, Chain.TESTNET_RINKEBY -> true else -> false @@ -82,7 +83,7 @@ open class EthereumMultistream( return super.isRunning() || reader.isRunning() } - override fun getReader(): EthereumReader { + override fun getReader(): EthereumCachingReader { return reader } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt index 274f6c61..648a66d7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt @@ -24,7 +24,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import java.util.function.Function -class EthereumPriorityFees(upstreams: Multistream, reader: EthereumReader, heightLimit: Int) : +class EthereumPriorityFees(upstreams: Multistream, reader: EthereumCachingReader, heightLimit: Int) : EthereumFees(upstreams, reader, heightLimit) { companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt index c5117a1f..0201405c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt @@ -34,10 +34,10 @@ import java.math.BigInteger * It provides data only if it's available through the router (cached, head, etc). * If data is not available locally then it returns `empty`; at this case the caller should call the remote node for actual data. * - * @see EthereumReader + * @see EthereumCachingReader */ class LocalCallRouter( - private val reader: EthereumReader, + private val reader: EthereumCachingReader, private val methods: CallMethods, private val head: Head ) : Reader { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index bc4364fc..abe8cd0a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -45,7 +45,7 @@ open class EthereumPosMultiStream( private var head: Head? = null - private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory()) + private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory()) private val feeEstimation = EthereumPriorityFees(this, reader, 256) private val subscribe = EthereumSubscribe(this) private val filteredHeads: MutableMap = @@ -77,7 +77,7 @@ open class EthereumPosMultiStream( return super.isRunning() || reader.isRunning() } - override fun getReader(): EthereumReader { + override fun getReader(): EthereumCachingReader { return reader } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy index fa1d0a03..ab303d48 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy @@ -262,6 +262,6 @@ class CachesSpec extends Specification { then: 1 * head.getCurrentHeight() >> 0xccf6e2 1 * receiptMemCache.acceptsRecentBlocks(0) >> true - 1 * receiptMemCache.add(receiptContainer) + 1 * receiptMemCache.add(receiptContainer) >> Mono.empty().then() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 81abb838..777adc05 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -29,9 +29,10 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumReader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Chain + import org.jetbrains.annotations.NotNull class MultistreamHolderMock implements MultistreamHolder { @@ -87,7 +88,7 @@ class MultistreamHolderMock implements MultistreamHolder { static class EthereumMultistreamMock extends EthereumPosMultiStream { - EthereumReader customReader = null + EthereumCachingReader customReader = null CallMethods customMethods = null Head customHead = null @@ -104,7 +105,7 @@ class MultistreamHolderMock implements MultistreamHolder { } @Override - EthereumReader getReader() { + EthereumCachingReader getReader() { if (customReader != null) { return customReader } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy similarity index 75% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReaderSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index d4f17cd0..07d719f2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache +import io.emeraldpay.dshackle.data.DefaultContainer import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.reader.Reader @@ -12,6 +13,7 @@ import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.Chain +import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId @@ -22,7 +24,6 @@ import org.apache.commons.collections4.Factory import reactor.core.publisher.Mono import reactor.test.StepVerifier import spock.lang.Specification - import java.time.Duration import java.time.Instant @@ -171,6 +172,76 @@ class EthereumDirectReaderSpec extends Specification { .verify(Duration.ofSeconds(1)) } + def "Reads tx receipt"() { + setup: + def json = new TransactionReceiptJson().tap { + transactionHash = TransactionId.from(hash1) + blockNumber = 100 + blockHash = BlockHash.from(hash1) + } + def up = Mock(Multistream) { + 1 * getApiSource(_) >> Stub(ApiSource) + } + def calls = Mock(Factory) { + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) + } + EthereumDirectReader reader = new EthereumDirectReader( + up, Caches.default(), new CurrentBlockCache(), calls + ) + reader.quorumReaderFactory = Mock(QuorumReaderFactory) { + 1 * create(_, _, _) >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( + new QuorumRpcReader.Result( + Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList() + ) + ) + } + } + when: + def act = reader.receiptReader.read(TransactionId.from(hash1)) + .block(Duration.ofSeconds(1)) + .with { new String(it) } + then: + act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}' + } + + def "Puts tx receipt in cache after reading"() { + setup: + def json = new TransactionReceiptJson().tap { + transactionHash = TransactionId.from(hash1) + blockNumber = 100 + blockHash = BlockHash.from(hash1) + } + def up = Mock(Multistream) { + 1 * getApiSource(_) >> Stub(ApiSource) + } + def calls = Mock(Factory) { + 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM) + } + def caches = Mock(Caches) { + // note that the Caches needs a Height value, otherwise it's not cached + 1 * cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer data -> data.txId.toHex() == hash1.substring(2) && data.height == 100 }) + } + EthereumDirectReader reader = new EthereumDirectReader( + up, caches, new CurrentBlockCache(), calls + ) + reader.quorumReaderFactory = Mock(QuorumReaderFactory) { + 1 * create(_, _, _) >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( + new QuorumRpcReader.Result( + Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList() + ) + ) + } + } + when: + def act = reader.receiptReader.read(TransactionId.from(hash1)) + .block(Duration.ofSeconds(1)) + .with { new String(it) } + then: + act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}' + } + def "Produce empty on non-existing tx"() { setup: def up = Mock(Multistream) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFeesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFeesSpec.groovy index e8695a35..f2dd9284 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFeesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLegacyFeesSpec.groovy @@ -30,7 +30,7 @@ class EthereumLegacyFeesSpec extends Specification { it.gasPrice = Wei.ofUnits(8, Wei.Unit.GWEI) } - def fees = new EthereumLegacyFees(Stub(EthereumMultistream), Stub(EthereumReader), 10) + def fees = new EthereumLegacyFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10) when: def act = fees.extractFee(block, tx) then: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFeesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFeesSpec.groovy index b1638242..c2fd0a85 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFeesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFeesSpec.groovy @@ -44,7 +44,7 @@ class EthereumPriorityFeesSpec extends Specification { it.maxPriorityFeePerGas = Wei.ofUnits(5.0001, Wei.Unit.GWEI) } - def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumReader), 10) + def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10) when: def act = fees.extractFee(block, tx) then: @@ -66,7 +66,7 @@ class EthereumPriorityFeesSpec extends Specification { it.gasPrice = Wei.from("0x198286458f") } - def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumReader), 10) + def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10) when: def act = fees.extractFee(block, tx) then: @@ -83,7 +83,7 @@ class EthereumPriorityFeesSpec extends Specification { new EthereumFees.EthereumFee(Wei.ofEthers(0.75), Wei.ofEthers(0.5), Wei.ofEthers(0.75), Wei.ofEthers(0.5)), new EthereumFees.EthereumFee(Wei.ofEthers(0.6), Wei.ofEthers(0.2), Wei.ofEthers(0.6), Wei.ofEthers(0.5)), ] - def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumReader), 10) + def fees = new EthereumPriorityFees(Stub(EthereumMultistream), Stub(EthereumCachingReader), 10) when: def act = Flux.fromIterable(inputs) .transform(fees.feeAggregation(ChainFees.Mode.AVG_LAST)) @@ -127,7 +127,7 @@ class EthereumPriorityFeesSpec extends Specification { 1 * getCurrentHeight() >> 13756007 } } - def reader = Mock(EthereumReader) { + def reader = Mock(EthereumCachingReader) { _ * it.blocksByHeightParsed() >> Mock(Reader) { 1 * it.read(13756006) >> Mono.just(block1) 1 * it.read(13756007) >> Mono.just(block2) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy deleted file mode 100644 index c824e25f..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy +++ /dev/null @@ -1,281 +0,0 @@ -/** - * Copyright (c) 2020 EmeraldPay, Inc - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.cache.BlocksMemCache -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.cache.ReceiptRedisCache -import io.emeraldpay.dshackle.cache.TxMemCache -import io.emeraldpay.dshackle.data.BlockContainer -import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.TxContainer -import io.emeraldpay.dshackle.data.TxId -import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock -import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock -import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.Multistream -import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.etherjar.domain.Address -import io.emeraldpay.etherjar.domain.BlockHash -import io.emeraldpay.etherjar.domain.TransactionId -import io.emeraldpay.etherjar.domain.Wei -import io.emeraldpay.etherjar.rpc.json.BlockJson -import io.emeraldpay.etherjar.rpc.json.TransactionJson -import io.emeraldpay.etherjar.rpc.json.TransactionRefJson -import org.apache.commons.collections4.Factory -import org.apache.commons.collections4.functors.ConstantFactory -import reactor.core.publisher.Mono -import spock.lang.Specification - -import java.time.Instant - -class EthereumReaderSpec extends Specification { - - def blockId = BlockId.from("f85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2") - def blockJson = new BlockJson().tap { blockJson -> - blockJson.hash = BlockHash.from(blockId.value) - blockJson.totalDifficulty = BigInteger.ONE - blockJson.number = 101 - blockJson.timestamp = Instant.ofEpochSecond(100000000) - blockJson.transactions = [] - blockJson.uncles = [] - } - def txId = BlockId.from("a38e7b4d456777c94b46c61a1e4cf52fbdd92acc4444719d1fad77005698c221") - def txJson = new TransactionJson().tap { json -> - json.hash = TransactionId.from(txId.value) - json.blockHash = blockJson.hash - json.blockNumber = blockJson.number - } - Factory calls = ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)) - - def "Block by Id reads from cache"() { - setup: - def memCache = Mock(BlocksMemCache) { - 1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson)) - } - def caches = Caches.newBuilder() - .setBlockByHash(memCache) - .build() - def reader = new EthereumReader(Stub(Multistream), caches, calls) - - when: - def act = reader.blocksByIdParsed().read(blockId).block() - - then: - act == blockJson - } - - def "Block by Id reads from api if cache is empty"() { - setup: - def memCache = Mock(BlocksMemCache) { - 1 * read(blockId) >> Mono.empty() - } - def caches = Caches.newBuilder() - .setBlockByHash(memCache) - .build() - def api = TestingCommons.api() - api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) - - def upstream = TestingCommons.multistream(api) - def reader = new EthereumReader(upstream, caches, calls) - - when: - def act = reader.blocksByIdParsed().read(blockId).block() - - then: - act == blockJson - } - - def "Block by Id reads from api if cache failed"() { - setup: - def memCache = Mock(BlocksMemCache) { - 1 * read(blockId) >> Mono.error(new IllegalStateException("Test error")) - } - def caches = Caches.newBuilder() - .setBlockByHash(memCache) - .build() - def api = TestingCommons.api() - api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) - - def upstream = TestingCommons.multistream(api) - def reader = new EthereumReader(upstream, caches, calls) - - when: - def act = reader.blocksByIdParsed().read(blockId).block() - - then: - act == blockJson - } - - def "Block by Hash reads from cache"() { - setup: - def memCache = Mock(BlocksMemCache) { - 1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson)) - } - def caches = Caches.newBuilder() - .setBlockByHash(memCache) - .build() - def reader = new EthereumReader(Stub(Multistream), caches, calls) - - when: - def act = reader.blocksByHashParsed().read(blockJson.hash).block() - - then: - act == blockJson - } - - def "Block by Hash reads from api if cache is empty"() { - setup: - def memCache = Mock(BlocksMemCache) { - 1 * read(blockId) >> Mono.empty() - } - def caches = Caches.newBuilder() - .setBlockByHash(memCache) - .build() - def api = TestingCommons.api() - api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) - def upstream = TestingCommons.multistream(api) - def reader = new EthereumReader(upstream, caches, calls) - - when: - def act = reader.blocksByHashParsed().read(blockJson.hash).block() - - then: - act == blockJson - } - - def "Tx by Hash reads from cache"() { - setup: - def memCache = Mock(TxMemCache) { - 1 * read(txId) >> Mono.just(TxContainer.from(txJson)) - } - def caches = Caches.newBuilder() - .setTxByHash(memCache) - .build() - def reader = new EthereumReader(Stub(Multistream), caches, calls) - - when: - def act = reader.txByHash().read(txJson.hash).block() - - then: - act == txJson - } - - def "Tx by Hash reads from api if cache is empty"() { - setup: - def memCache = Mock(TxMemCache) { - 1 * read(txId) >> Mono.empty() - } - def caches = Caches.newBuilder() - .setTxByHash(memCache) - .build() - - def api = TestingCommons.api() - api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson) - def upstream = TestingCommons.multistream(api) - def reader = new EthereumReader(upstream, caches, calls) - - when: - def act = reader.txByHash().read(txJson.hash).block() - - then: - act == txJson - } - - def "Caches balance until block mined"() { - setup: - def api = TestingCommons.api() - // no height - api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0x10") - // height 101 + 1 => 102 => 0x66 - api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "0x66"], "0xff") - EthereumPosRpcUpstreamMock upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api) - def upstreams = TestingCommons.multistream(upstream) - def reader = new EthereumReader(upstreams, Caches.default(), calls) - reader.start() - - when: - def act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block() - - then: - act == Wei.from("0x10") - - when: - //now it should use cached value, without actual request - act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block() - - then: - act == Wei.from("0x10") - - when: - //move head forward, which should erase cache - def block2 = blockJson.copy().tap { - it.number++ - it.totalDifficulty = BigInteger.TWO - } - upstream.nextBlock(BlockContainer.from(block2)) - Thread.sleep(50) - act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block() - - then: - act == Wei.from("0xff") - } - - def "Read receipt from upstream if cache is empty"() { - setup: - def api = TestingCommons.api() - api.answerOnce("eth_getTransactionReceipt", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"], [ - transactionHash: "0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2" - ]) - EthereumPosRpcUpstreamMock upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api) - def upstreams = TestingCommons.multistream(upstream) - def reader = new EthereumReader(upstreams, Caches.default(), calls) - reader.start() - - when: - def act = reader.receipts().read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")).block() - - then: - act != null - new String(act) == '{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}' - } - - def "Read receipt from cache if available"() { - setup: - def api = TestingCommons.api() - EthereumPosRpcUpstreamMock upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM, api) - def upstreams = TestingCommons.multistream(upstream) - def receiptCache = Mock(ReceiptRedisCache) { - 1 * it.read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")) >> - Mono.just('{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}'.bytes) - } - def cashes = Caches.newBuilder() - .setReceipts(receiptCache) - .build() - def reader = new EthereumReader(upstreams, cashes, calls) - reader.start() - - when: - def act = reader.receipts().read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")).block() - - then: - act != null - new String(act) == '{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}' - api.calls.get() == 0 - } -} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy index c936b8b3..12e6cb44 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy @@ -23,7 +23,7 @@ class LocalCallRouterSpec extends Specification { setup: def methods = new DefaultEthereumMethods(Chain.ETHEREUM) def router = new LocalCallRouter( - new EthereumReader( + new EthereumCachingReader( TestingCommons.multistream(TestingCommons.api()), Caches.default(), ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)) @@ -41,7 +41,7 @@ class LocalCallRouterSpec extends Specification { setup: def methods = new DefaultEthereumMethods(Chain.ETHEREUM) def router = new LocalCallRouter( - new EthereumReader( + new EthereumCachingReader( TestingCommons.multistream(TestingCommons.api()), Caches.default(), ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM)) @@ -61,7 +61,7 @@ class LocalCallRouterSpec extends Specification { def head = Mock(Head) { 1 * getCurrentHeight() >> 101L } - def reader = Mock(EthereumReader) { + def reader = Mock(EthereumCachingReader) { _ * blocksByIdAsCont() >> new EmptyReader<>() _ * txByHashAsCont() >> new EmptyReader<>() 1 * blocksByHeightAsCont() >> Mock(Reader) { @@ -87,7 +87,7 @@ class LocalCallRouterSpec extends Specification { def "getBlockByNumber with earliest uses 0 block"() { setup: def head = Stub(Head) {} - def reader = Mock(EthereumReader) { + def reader = Mock(EthereumCachingReader) { _ * blocksByIdAsCont() >> new EmptyReader<>() _ * txByHashAsCont() >> new EmptyReader<>() 1 * blocksByHeightAsCont() >> Mock(Reader) { @@ -113,7 +113,7 @@ class LocalCallRouterSpec extends Specification { def "getBlockByNumber fetches the block"() { setup: def head = Stub(Head) {} - def reader = Mock(EthereumReader) { + def reader = Mock(EthereumCachingReader) { _ * blocksByIdAsCont() >> new EmptyReader<>() _ * txByHashAsCont() >> new EmptyReader<>() 1 * blocksByHeightAsCont() >> Mock(Reader) {