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/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 6d7d64df..3a9a082d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -52,7 +52,6 @@ abstract class Multistream( val chain: Chain, private val upstreams: MutableList, val caches: Caches, - val postprocessor: RequestPostprocessor ) : Upstream, Lifecycle { companion object { @@ -144,23 +143,6 @@ abstract class Multistream( return FilteredApis(chain, upstreams, matcher, i) } - /** - * Finds an API that executed directly on a remote. - */ - open fun getDirectApi(matcher: Selector.Matcher): Mono> { - val apis = getApiSource(matcher) - apis.request(1) - return Mono.from(apis) - .map(Upstream::getApi) - .map { - RequestPostprocessor.wrap( - it, - postprocessor - ) - } // TODO do it on upstream init, not each time it's called - .switchIfEmpty(Mono.error(Exception("No API available for $chain"))) - } - abstract fun getFeeEstimation(): ChainFees /** diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt deleted file mode 100644 index 69ec2693..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt +++ /dev/null @@ -1,40 +0,0 @@ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import reactor.core.publisher.Mono - -interface RequestPostprocessor { - - fun onReceive(method: String, params: List, json: ByteArray) - - class Empty : RequestPostprocessor { - override fun onReceive(method: String, params: List, json: ByteArray) {} - } - - companion object { - fun wrap( - reader: Reader, - processor: RequestPostprocessor - ): Reader { - return Wrapper(reader, processor) - } - } - - class Wrapper( - private val reader: Reader, - private val processor: RequestPostprocessor - ) : Reader { - - override fun read(key: JsonRpcRequest): Mono { - return reader.read(key) - .doOnNext { - if (it.hasResult()) { - val result = it.getResult() - processor.onReceive(key.method, key.params, result) - } - } - } - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index 3393fd82..1cfe55a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -32,8 +32,8 @@ import reactor.core.publisher.Mono open class BitcoinMultistream( chain: Chain, private val sourceUpstreams: MutableList, - caches: Caches -) : Multistream(chain, sourceUpstreams as MutableList, caches, RequestPostprocessor.Empty()), Lifecycle { + caches: Caches, +) : Multistream(chain, sourceUpstreams as MutableList, caches), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java) @@ -95,6 +95,16 @@ open class BitcoinMultistream( onHeadUpdated(head) return head } + /** + * Finds an API that executed directly on a remote. + */ + open fun getDirectApi(matcher: Selector.Matcher): Mono> { + val apis = getApiSource(matcher) + apis.request(1) + return Mono.from(apis) + .map(Upstream::getApi) + .switchIfEmpty(Mono.error(Exception("No API available for $chain"))) + } override fun getRoutedApi(matcher: Selector.Matcher): Mono> { return Mono.just(callRouter) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt deleted file mode 100644 index 83e5d142..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt +++ /dev/null @@ -1,60 +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.Global -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.DefaultContainer -import io.emeraldpay.dshackle.data.TxId -import io.emeraldpay.dshackle.upstream.RequestPostprocessor -import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson -import org.slf4j.LoggerFactory - -class CacheRequested( - private val caches: Caches -) : RequestPostprocessor { - - companion object { - private val log = LoggerFactory.getLogger(CacheRequested::class.java) - } - - override fun onReceive(method: String, params: List, json: ByteArray) { - try { - if (method == "eth_getTransactionReceipt") { - cacheTxReceipt(params, json) - } - } catch (e: Throwable) { - log.warn("Failed to cache result", e) - } - } - - fun cacheTxReceipt(params: List, json: ByteArray) { - if (params.size != 1) { - return - } - // note: json could be a `null` value - val parsed = Global.objectMapper.readValue(json, TransactionReceiptJson::class.java) ?: return - val value = DefaultContainer( - TxId.from(parsed.transactionHash), - BlockId.from(parsed.blockHash), - parsed.blockNumber, - json, - parsed - ) - caches.cacheReceipt(Caches.Tag.REQUESTED, value) - } -} 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..67f0cf5c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -37,7 +37,7 @@ open class EthereumMultistream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)), EthereumLikeMultistream { +) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { private val log = LoggerFactory.getLogger(EthereumMultistream::class.java) @@ -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..9d797483 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 @@ -37,7 +37,7 @@ open class EthereumPosMultiStream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)), EthereumLikeMultistream { +) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { private val log = LoggerFactory.getLogger(EthereumPosMultiStream::class.java) @@ -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/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index fb36e45b..fc8fe7b1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -180,26 +180,6 @@ class MultistreamSpec extends Specification { !act } - def "Call postprocess after api use"() { - setup: - def request = new JsonRpcRequest("test_foo", [1], 1, null) - - def api = TestingCommons.api() - api.answer("test_foo", [1], "test") - def postprocessor = Mock(RequestPostprocessor) - def up = TestingCommons.upstream(api) - def multistream = new TestMultistream([up], postprocessor) - - when: - def rdr = multistream.getDirectApi(Selector.empty).block(Duration.ofSeconds(1)) - def act = rdr.read(request).block(Duration.ofSeconds(1)) - - then: - act != null - act.hasResult() - act.resultAsProcessedString == "test" - 1 * postprocessor.onReceive("test_foo", [1], "\"test\"".bytes) - } def "Filter upstream matching selector single"() { setup: @@ -341,48 +321,6 @@ class MultistreamSpec extends Specification { .expectComplete() .verify(Duration.ofSeconds(1)) } - - class TestMultistream extends Multistream { - - TestMultistream(List upstreams, @NotNull RequestPostprocessor postprocessor) { - super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor) - } - - @Override - Mono> getRoutedApi(@NotNull Selector.Matcher matcher) { - return null - } - - @Override - Head updateHead() { - return null - } - - @Override - void setHead(@NotNull Head head) { - - } - - @Override - Head getHead() { - return null - } - - @Override - Collection getLabels() { - return null - } - - public T cast(Class selfType) { - return this - } - - @Override - ChainFees getFeeEstimation() { - return null - } - } - class TestEthereumPosMultistream extends EthereumPosMultiStream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy deleted file mode 100644 index bf73f37d..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy +++ /dev/null @@ -1,47 +0,0 @@ -package io.emeraldpay.dshackle.upstream - -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import reactor.core.publisher.Mono -import spock.lang.Specification - -import java.time.Duration - -class RequestPostprocessorSpec extends Specification { - - def "Wrappers calls onReceive for a value"() { - setup: - def request = new JsonRpcRequest("test_foo", [1], 1, null) - def processor = Mock(RequestPostprocessor) - def api = TestingCommons.api() - api.answer("test_foo", [1], "test") - def wrapped = new RequestPostprocessor.Wrapper(api, processor) - - when: - def act = wrapped.read(request).block(Duration.ofSeconds(1)) - - then: - act.hasResult() - act.resultAsProcessedString == "test" - 1 * processor.onReceive("test_foo", [1], "\"test\"".bytes) - } - - def "Wrappers doesn't call onReceive for no value"() { - setup: - def request = new JsonRpcRequest("test_foo", [1], 1, null) - def processor = Mock(RequestPostprocessor) - Reader reader = Mock(Reader) { - 1 * it.read(request) >> Mono.empty() - } - def wrapped = new RequestPostprocessor.Wrapper(reader, processor) - - when: - def act = wrapped.read(request).block(Duration.ofSeconds(1)) - - then: - act == null - 0 * processor.onReceive(_, _, _) - } -} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy deleted file mode 100644 index 2b93c5dc..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy +++ /dev/null @@ -1,58 +0,0 @@ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.data.DefaultContainer -import spock.lang.Specification - -class CacheRequestedSpec extends Specification { - - def "Do nothing if unsupported method"() { - setup: - def caches = Mock(Caches) - CacheRequested instance = new CacheRequested(caches) - when: - instance.onReceive("eth_hashrate", [], '"0x38a"'.bytes) - then: - 0 * caches._(*_) - } - - def "Caches tx receipt"() { - setup: - def caches = Mock(Caches) - CacheRequested instance = new CacheRequested(caches) - def json = '''{ - "blockHash": "0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae", - "blockNumber": "0x213b8a", - "contractAddress": null, - "cumulativeGasUsed": "0x5208", - "gasUsed": "0x5208", - "logs": [], - "transactionHash": "0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197", - "transactionIndex": "0x00" - }'''.bytes - - when: - instance.onReceive("eth_getTransactionReceipt", ["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json) - - then: - 1 * caches.cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer it -> - it.height == 0x213b8a && - it.txId.toHex() == "5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197" && - it.json == json - }) - } - - def "Ignore null"() { - setup: - def caches = Mock(Caches) - CacheRequested instance = new CacheRequested(caches) - def json = 'null'.bytes - - when: - instance.cacheTxReceipt(["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json) - - then: - 0 * caches.cacheReceipt(_, _) - } - -} 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) {