From 0ef46b2980f9753cc29b6bd3dbad5509efbdebd2 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 4 May 2020 21:51:42 -0400 Subject: [PATCH] problem: different places to fetch from upstream solution: single access reader for Ethereum upstreams --- .../dshackle/cache/CurrentBlockCache.kt | 24 ++ .../dshackle/reader/CompoundReader.kt | 14 +- .../dshackle/reader/RekeyingReader.kt | 43 ++++ .../dshackle/reader/TransformingReader.kt | 39 +++ .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 27 +- .../emeraldpay/dshackle/rpc/TrackBitcoinTx.kt | 1 - .../dshackle/rpc/TrackEthereumAddress.kt | 18 +- .../dshackle/rpc/TrackEthereumTx.kt | 38 +-- .../dshackle/upstream/ChainUpstreams.kt | 1 - .../dshackle/upstream/CurrentUpstreams.kt | 4 +- .../emeraldpay/dshackle/upstream/Upstream.kt | 6 +- .../{BitcoinData.kt => BitcoinReader.kt} | 4 +- .../upstream/bitcoin/BitcoinUpstream.kt | 4 +- ...eams.kt => AggregatedEthereumUpstreams.kt} | 24 +- .../dshackle/upstream/ethereum/EthereumApi.kt | 9 + .../upstream/ethereum/EthereumReader.kt | 172 +++++++++++++ .../dshackle/reader/RekeyingReaderSpec.groovy | 40 +++ .../reader/TransformingReaderSpec.groovy | 40 +++ .../dshackle/rpc/TrackBitcoinTxSpec.groovy | 8 +- .../rpc/TrackEthereumAddressSpec.groovy | 10 +- .../dshackle/rpc/TrackEthereumTxSpec.groovy | 8 +- .../dshackle/test/TestingCommons.groovy | 6 +- .../dshackle/test/UpstreamsMock.groovy | 34 ++- .../upstream/AggregatedUpstreamSpec.groovy | 4 +- .../ethereum/EthereumReaderSpec.groovy | 237 ++++++++++++++++++ 25 files changed, 748 insertions(+), 67 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/CurrentBlockCache.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/RekeyingReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/TransformingReader.kt rename src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/{BitcoinData.kt => BitcoinReader.kt} (92%) rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumChainUpstreams.kt => AggregatedEthereumUpstreams.kt} (85%) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/reader/RekeyingReaderSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/reader/TransformingReaderSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/CurrentBlockCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/CurrentBlockCache.kt new file mode 100644 index 00000000..6b11d8e3 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CurrentBlockCache.kt @@ -0,0 +1,24 @@ +package io.emeraldpay.dshackle.cache + +import io.emeraldpay.dshackle.reader.Reader +import reactor.core.publisher.Mono +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicReference + +class CurrentBlockCache : Reader { + + private val cache = AtomicReference(ConcurrentHashMap()) + + override fun read(key: K): Mono { + return Mono.justOrEmpty(cache.get()[key]) + } + + fun put(key: K, data: D) { + cache.get()[key] = data + } + + fun evict() { + cache.set(ConcurrentHashMap()) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt index 83256fbe..26a425af 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt @@ -16,8 +16,11 @@ */ package io.emeraldpay.dshackle.reader +import io.emeraldpay.dshackle.Defaults +import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import java.time.Duration /** * Composition of multiple readers. Reader returns first value returned by any of the source readers. @@ -26,12 +29,21 @@ class CompoundReader( private vararg val readers: Reader ): Reader { + companion object { + private val log = LoggerFactory.getLogger(CompoundReader::class.java) + } + override fun read(key: K): Mono { if (readers.isEmpty()) { return Mono.empty() } return Flux.fromIterable(readers.asIterable()) - .flatMap { it.read(key) }.next() + .flatMap { rdr -> + rdr.read(key) + .timeout(Defaults.timeoutInternal, Mono.empty()) + .doOnError { t -> log.warn("Failed to read from $rdr", t) } + .onErrorResume { Mono.empty() } + }.next() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/RekeyingReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/RekeyingReader.kt new file mode 100644 index 00000000..7e37d1ad --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RekeyingReader.kt @@ -0,0 +1,43 @@ +/** + * 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.reader + +import reactor.core.publisher.Mono +import java.util.function.Function + +/** + * Reader wrapper that maps the input key from ne value to another (ex. convert from Long to String) + */ +class RekeyingReader( + /** + * Mapping between original Key and Key supported by the reader + */ + private val rekey: Function, + /** + * Actual reader + */ + private val reader: Reader +) : Reader { + + override fun read(key: K): Mono { + return Mono.just(key) + .map(rekey) + .flatMap { + reader.read(it) + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/TransformingReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/TransformingReader.kt new file mode 100644 index 00000000..4a82a0a8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/TransformingReader.kt @@ -0,0 +1,39 @@ +/** + * 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.reader + +import reactor.core.publisher.Mono +import java.util.function.Function + +/** + * Reader wrapper that transforms output of the reader to a different format + */ +class TransformingReader( + /** + * Actual reader + */ + private val reader: Reader, + /** + * Result transformation + */ + private val transformer: Function +) : Reader { + + override fun read(key: K): Mono { + return reader.read(key).map(transformer) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 8be7ed84..f05388c8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -51,24 +51,39 @@ class BlockchainRpc( override fun subscribeTxStatus(request: Mono): Flux { return request.flatMapMany { request -> val chain = Chain.byId(request.chainValue) - trackTx.find { it.isSupported(chain) }?.subscribe(request) - ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) + try { + trackTx.find { it.isSupported(chain) }?.subscribe(request) + ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) + } catch (t: Throwable) { + log.error("Internal error during Tx Subscription", t) + Flux.error(IllegalStateException("Internal Error")) + } } } override fun subscribeBalance(requestMono: Mono): Flux { return requestMono.flatMapMany { request -> val chain = Chain.byId(request.asset.chainValue) - trackAddress.find { it.isSupported(chain) }?.subscribe(request) - ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) + try { + trackAddress.find { it.isSupported(chain) }?.subscribe(request) + ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) + } catch (t: Throwable) { + log.error("Internal error during Balance Subscription", t) + Flux.error(IllegalStateException("Internal Error")) + } } } override fun getBalance(requestMono: Mono): Flux { return requestMono.flatMapMany { request -> val chain = Chain.byId(request.asset.chainValue) - trackAddress.find { it.isSupported(chain) }?.getBalance(request) - ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) + try { + trackAddress.find { it.isSupported(chain) }?.getBalance(request) + ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) + } catch (t: Throwable) { + log.error("Internal error during Balance Request", t) + Flux.error(IllegalStateException("Internal Error")) + } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt index abf75491..c2e8e40f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt @@ -128,7 +128,6 @@ class TrackBitcoinTx( } fun loadMempool(upstream: BitcoinUpstream, txid: String): Mono { - println("access: ${upstream.getData()}") val mempool = upstream.getData().getMempool().get() return mempool.map { if (it.contains(txid)) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt index 9edd3459..ccc87424 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt @@ -21,15 +21,12 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.SilentException -import io.emeraldpay.dshackle.upstream.AggregatedUpstream -import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi +import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.Wei -import io.infinitape.etherjar.rpc.Commands -import io.infinitape.etherjar.rpc.json.BlockTag import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -90,6 +87,11 @@ class TrackEthereumAddress( } } + fun getUpstream(chain: Chain): AggregatedEthereumUpstreams { + return upstreams.getUpstream(chain)?.cast(AggregatedEthereumUpstreams::class.java, EthereumApi::class.java) + ?: throw SilentException.UnsupportedBlockchain(chain) + } + private fun initAddress(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) if (!upstreams.isAvailable(chain)) { @@ -120,10 +122,10 @@ class TrackEthereumAddress( } fun getBalance(addr: TrackedAddress): Mono { - val up = upstreams.getUpstream(addr.chain) as AggregatedUpstream? - ?: return Mono.error(SilentException.UnsupportedBlockchain(addr.chain)) - return up.getApi(Selector.empty) - .flatMap { api -> api.executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) } + return getUpstream(addr.chain) + .getReader() + .balance() + .read(addr.address) .timeout(Defaults.timeout) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt index 8b709535..f4df3d70 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt @@ -23,15 +23,13 @@ import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.TxId -import io.emeraldpay.dshackle.upstream.AggregatedUpstream -import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi +import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionJson @@ -72,8 +70,7 @@ class TrackEthereumTx( override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux { val base = prepareTracking(request) - val up = upstreams.getUpstream(base.chain)?.castApi(EthereumApi::class.java) - ?: return Flux.empty() + val up = getUpstream(base.chain) return update(base) .defaultIfEmpty(base) .flatMapMany { @@ -82,9 +79,18 @@ class TrackEthereumTx( .map(this@TrackEthereumTx::asProto) .subscribeOn(scheduler) } + .doOnError { t -> + log.error("Subscription error", t) + } } - fun subscribe(base: TxDetails, up: Upstream): Flux { + + fun getUpstream(chain: Chain): AggregatedEthereumUpstreams { + return upstreams.getUpstream(chain)?.cast(AggregatedEthereumUpstreams::class.java, EthereumApi::class.java) + ?: throw SilentException.UnsupportedBlockchain(chain) + } + + fun subscribe(base: TxDetails, up: AggregatedEthereumUpstreams): Flux { var latestTx = base val untilFound = Mono.just(latestTx) @@ -151,11 +157,9 @@ class TrackEthereumTx( private fun update(tx: TxDetails): Mono { val initialStatus = tx.status - val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream? - ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain)) - val execution = upstream.getApi(Selector.empty) - .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) } - return execution + val upstream = getUpstream(tx.chain) + return upstream.getReader() + .txByHash().read(tx.txid) .onErrorResume(RpcException::class.java) { t -> log.warn("Upstream error, ignoring. {}", t.rpcMessage) Mono.empty() @@ -198,10 +202,12 @@ class TrackEthereumTx( } private fun loadWeight(tx: TxDetails): Mono { - val upstream = upstreams.getUpstream(tx.chain) as AggregatedUpstream? - ?: return Mono.error(SilentException.UnsupportedBlockchain(tx.chain)) - return upstream.getApi(Selector.empty) - .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) } + val upstream = getUpstream(tx.chain) + if (tx.status.blockHash == null) { + return Mono.empty() + } + return upstream.getReader() + .blocksByHash().read(tx.status.blockHash) .map { block -> setBlockDetails(tx, block) }.doOnError { t -> @@ -209,7 +215,7 @@ class TrackEthereumTx( } } - fun updateFromBlock(upstream: Upstream, tx: TxDetails, blockTx: TransactionJson): Mono { + fun updateFromBlock(upstream: AggregatedEthereumUpstreams, tx: TxDetails, blockTx: TransactionJson): Mono { return if (blockTx.blockNumber != null && blockTx.blockHash != null && blockTx.blockHash != ZERO_BLOCK) { val updated = tx.withStatus( blockHash = blockTx.blockHash, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 362bad82..7b5b33c0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt index ecf48ddb..84f733df 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt @@ -28,7 +28,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams +import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory @@ -65,7 +65,7 @@ class CurrentUpstreams( .cast(EthereumUpstream::class.java, EthereumApi::class.java) as Upstream val current = chainMapping[chain] as ChainUpstreams? val factory = Callable { - EthereumChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams + AggregatedEthereumUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams } processUpdate(change, up, current, factory) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index 83d80735..9a3005ce 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -21,12 +21,12 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import reactor.core.publisher.Flux import reactor.core.publisher.Mono -interface Upstream { +interface Upstream { fun isAvailable(): Boolean fun getStatus(): UpstreamAvailability fun observeStatus(): Flux fun getHead(): Head - fun getApi(matcher: Selector.Matcher): Mono + fun getApi(matcher: Selector.Matcher): Mono fun getOptions(): UpstreamsConfig.Options fun setLag(lag: Long) fun getLag(): Long @@ -34,6 +34,6 @@ interface Upstream { fun getMethods(): CallMethods fun getId(): String - fun castApi(apiType: Class): Upstream + fun castApi(apiType: Class): Upstream fun , TA : UpstreamApi> cast(selfType: Class, apiType: Class): T } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinData.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt similarity index 92% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinData.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt index a5e4ecbf..d736a41a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinData.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt @@ -19,13 +19,13 @@ import io.emeraldpay.dshackle.upstream.Head import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle -open class BitcoinData( +open class BitcoinReader( api: DirectBitcoinApi, head: Head ) : Lifecycle { companion object { - private val log = LoggerFactory.getLogger(BitcoinData::class.java) + private val log = LoggerFactory.getLogger(BitcoinReader::class.java) } private val mempool = CachingMempoolData(api, head) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt index e7783be6..14441e09 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt @@ -42,7 +42,7 @@ open class BitcoinUpstream( private val head: Head = createHead() private var validatorSubscription: Disposable? = null - private val data = BitcoinData(api, head) + private val data = BitcoinReader(api, head) private fun createHead(): Head { return BitcoinRpcHead( @@ -51,7 +51,7 @@ open class BitcoinUpstream( ) } - open fun getData(): BitcoinData { + open fun getData(): BitcoinReader { return data } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/AggregatedEthereumUpstreams.kt similarity index 85% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/AggregatedEthereumUpstreams.kt index 545c1161..72e9f8c1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/AggregatedEthereumUpstreams.kt @@ -24,7 +24,7 @@ import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle -class EthereumChainUpstreams( +open class AggregatedEthereumUpstreams( chain: Chain, val upstreams: MutableList, caches: Caches, @@ -32,11 +32,13 @@ class EthereumChainUpstreams( ) : ChainUpstreams(chain, upstreams as MutableList>, caches, objectMapper) { companion object { - private val log = LoggerFactory.getLogger(EthereumChainUpstreams::class.java) + private val log = LoggerFactory.getLogger(AggregatedEthereumUpstreams::class.java) } private var head: Head? = null + private val reader: EthereumReader = EthereumReader(this, this.caches, objectMapper) + init { this.init() } @@ -48,6 +50,24 @@ class EthereumChainUpstreams( super.init() } + override fun start() { + super.start() + reader.start() + } + + override fun stop() { + super.stop() + reader.stop() + } + + override fun isRunning(): Boolean { + return super.isRunning() || reader.isRunning + } + + open fun getReader(): EthereumReader { + return reader + } + override fun getHead(): Head { return head!! } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt index 6f9ebca0..b3e12e63 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamApi import io.infinitape.etherjar.rpc.* @@ -37,6 +38,14 @@ abstract class EthereumApi( private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) var upstream: Upstream? = null + fun reader(): Reader, RS> { + return object : Reader, RS> { + override fun read(key: RpcCall): Mono { + return this@EthereumApi.executeAndConvert(key) + } + } + } + fun execute(rpcCall: RpcCall): Mono { return execute(0, rpcCall.method, rpcCall.params as List) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt new file mode 100644 index 00000000..5a540bbb --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt @@ -0,0 +1,172 @@ +/** + * 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 com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled +import io.emeraldpay.dshackle.cache.CurrentBlockCache +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.reader.* +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream +import io.infinitape.etherjar.domain.Address +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.domain.Wei +import io.infinitape.etherjar.rpc.Commands +import io.infinitape.etherjar.rpc.RpcCall +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.BlockTag +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable +import reactor.core.publisher.Mono +import reactor.util.retry.Retry +import java.time.Duration +import java.util.concurrent.TimeoutException +import java.util.function.Function + +open class EthereumReader( + private val up: Upstream, + private val caches: Caches, + private val objectMapper: ObjectMapper +) : Lifecycle { + + companion object { + private val log = LoggerFactory.getLogger(EthereumReader::class.java) + } + + private var headListener: Disposable? = null + private val balanceCache = CurrentBlockCache() + + private val extractBlock = Function> { block -> + objectMapper + .readValue(block.json, BlockJson::class.java) + .withoutTransactionDetails() + } + + private val extractTx = Function { tx -> + objectMapper + .readValue(tx.json, TransactionJson::class.java) + } + + private val blocksDirect: Reader> + private val txDirect: Reader + private val balanceDirect: Reader + + private val idToBlockHash = Function { id -> BlockHash.from(id.value) } + private val blockHashToId = Function { hash -> BlockId.from(hash) } + + private val txHashToId = Function { hash -> TxId.from(hash) } + + init { + blocksDirect = object : Reader> { + override fun read(key: BlockHash): Mono> { + return up.getApi(Selector.empty).flatMap { api -> + api.executeAndConvert(Commands.eth().getBlock(key)) + .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $key"))) + .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))) + .doOnNext { block -> + caches.cache(Caches.Tag.REQUESTED, BlockContainer.from(block, objectMapper)) + } + } + } + } + txDirect = object : Reader { + override fun read(key: TransactionId): Mono { + return up.getApi(Selector.empty).flatMap { api -> + api.executeAndConvert(Commands.eth().getTransaction(key)) + .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key"))) + .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))) + .doOnNext { tx -> + if (tx.blockNumber != null && tx.blockHash != null) { + caches.cache(Caches.Tag.REQUESTED, TxContainer.from(tx, objectMapper)) + } + } + } + } + } + balanceDirect = object : Reader { + override fun read(key: Address): Mono { + return up.getApi(Selector.empty).flatMap { api -> + api.executeAndConvert(Commands.eth().getBalance(key, BlockTag.LATEST)) + .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key"))) + .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))) + .doOnNext { value -> + balanceCache.put(key, value) + } + } + } + } + } + + fun blocksById(): Reader> { + return CompoundReader( + TransformingReader(caches.getBlocksByHash(), extractBlock), + RekeyingReader(idToBlockHash, blocksDirect) + ) + } + + fun blocksByHash(): Reader> { + return CompoundReader( + RekeyingReader( + blockHashToId, + TransformingReader(caches.getBlocksByHash(), extractBlock) + ), + blocksDirect + ) + } + + fun txByHash(): Reader { + return CompoundReader( + RekeyingReader( + txHashToId, + TransformingReader(caches.getTxByHash(), extractTx) + ), + txDirect + ) + } + + fun balance(): Reader { + return CompoundReader( + balanceCache, balanceDirect + ) + } + + override fun isRunning(): Boolean { + return this.headListener != null + } + + override fun start() { + this.headListener = up.getHead().getFlux().subscribe { + balanceCache.evict() + } + } + + override fun stop() { + val headListener = this.headListener + this.headListener = null + headListener?.dispose() + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/reader/RekeyingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/reader/RekeyingReaderSpec.groovy new file mode 100644 index 00000000..2fc08ada --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/reader/RekeyingReaderSpec.groovy @@ -0,0 +1,40 @@ +/** + * 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.reader + +import reactor.core.publisher.Mono +import spock.lang.Specification + +import java.util.function.Function + +class RekeyingReaderSpec extends Specification { + + def "Simple rekey"() { + setup: + Reader r = new Reader() { + @Override + Mono read(Long key) { + return Mono.just(key * 2) + } + } + Function f = { String s -> Long.parseLong(s) } + when: + def rekey = new RekeyingReader(f, r) + def act = rekey.read("4").block() + then: + act == 8 + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/reader/TransformingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/reader/TransformingReaderSpec.groovy new file mode 100644 index 00000000..b889b5ab --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/reader/TransformingReaderSpec.groovy @@ -0,0 +1,40 @@ +/** + * 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.reader + +import reactor.core.publisher.Mono +import spock.lang.Specification + +import java.util.function.Function + +class TransformingReaderSpec extends Specification { + + def "Transform"() { + Reader r = new Reader() { + @Override + Mono read(String key) { + return Mono.just(key + "2") + } + } + Function f = { String s -> Long.parseLong(s) } + when: + def transform = new TransformingReader(r, f) + def act = transform.read("4").block() + then: + act == 42L + } + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy index d018ee09..ed15dfe5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy @@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinData +import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinReader import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.bitcoin.CachingMempoolData @@ -46,7 +46,7 @@ class TrackBitcoinTxSpec extends Specification { ]) } BitcoinUpstream upstream = Mock(BitcoinUpstream) { - _ * getData() >> Mock(BitcoinData) { + _ * getData() >> Mock(BitcoinReader) { _ * getMempool() >> mempoolAccess } } @@ -72,7 +72,7 @@ class TrackBitcoinTxSpec extends Specification { ]) } BitcoinUpstream upstream = Mock(BitcoinUpstream) { - _ * getData() >> Mock(BitcoinData) { + _ * getData() >> Mock(BitcoinReader) { _ * getMempool() >> mempoolAccess } } @@ -231,7 +231,7 @@ class TrackBitcoinTxSpec extends Specification { BitcoinUpstream upstream = Mock(BitcoinUpstream) { _ * getApi(_) >> Mono.just(api) _ * getHead() >> head - _ * getData() >> Mock(BitcoinData) { + _ * getData() >> Mock(BitcoinReader) { _ * getMempool() >> mempoolAccess } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy index 757d58c3..53408786 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumAddressSpec.groovy @@ -19,14 +19,18 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.UpstreamsMock import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.ethereum.EthereumReader import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.ReactorRpcClient +import io.infinitape.etherjar.rpc.RpcCall import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.TopicProcessor import reactor.core.scheduler.Schedulers @@ -111,12 +115,12 @@ class TrackEthereumAddressSpec extends Specification { def flux = trackAddress.subscribe(req) then: StepVerifier.create(flux) - .expectNext(exp1) + .expectNext(exp1).as("First block") .then { upstreamMock.nextBlock(BlockContainer.from(block2, TestingCommons.objectMapper())) } - .expectNext(exp2) + .expectNext(exp2).as("Second block") .thenCancel() - .verify(Duration.ofSeconds(3)) + .verify(Duration.ofSeconds(1)) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index e8da61a6..aa0b7fb0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy @@ -27,8 +27,7 @@ import io.emeraldpay.dshackle.test.UpstreamsMock import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams -import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs +import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId @@ -39,7 +38,6 @@ import io.infinitape.etherjar.rpc.json.TransactionRefJson import reactor.core.publisher.Flux import reactor.test.StepVerifier import reactor.test.scheduler.VirtualTimeScheduler -import spock.lang.Ignore import spock.lang.Specification import java.time.Duration @@ -68,7 +66,7 @@ class TrackEthereumTxSpec extends Specification { } def blockHeadJson = new BlockJson().with { - it.hash = BlockHash.from("0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22") + it.hash = BlockHash.from("0x552d882f16f34a9dba2ccdc05c0a6a27c22a0e65cbc1b52a8ca60562112c6060") it.timestamp = Instant.ofEpochMilli(156400200000) it.number = 108 it.totalDifficulty = BigInteger.valueOf(800) @@ -123,7 +121,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api(Stub(ReactorRpcClient)) def upstreamMock = TestingCommons.upstream(apiMock) Upstreams upstreams = new UpstreamsMock(Chain.ETHEREUM, upstreamMock) - ((EthereumChainUpstreams) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) { + ((AggregatedEthereumUpstreams) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) { _ * getFlux() >> Flux.empty() } TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index e0b3c67a..867f63bc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -25,15 +25,13 @@ import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.CacheConfig import io.emeraldpay.dshackle.upstream.AggregatedUpstream -import io.emeraldpay.dshackle.upstream.ChainUpstreams import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams +import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.ReactorRpcClient -import org.springframework.core.env.StandardEnvironment import java.text.SimpleDateFormat @@ -77,7 +75,7 @@ class TestingCommons { } static AggregatedUpstream aggregatedUpstream(EthereumUpstream up) { - return new EthereumChainUpstreams(Chain.ETHEREUM, [up], Caches.default(objectMapper()), objectMapper()) + return new AggregatedEthereumUpstreams(Chain.ETHEREUM, [up], Caches.default(objectMapper()), objectMapper()) } static CachesFactory emptyCaches() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy index e59c16bc..4f1f6ecb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy @@ -16,13 +16,15 @@ */ package io.emeraldpay.dshackle.test +import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.upstream.AggregatedUpstream -import io.emeraldpay.dshackle.upstream.ChainUpstreams import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstreams -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams +import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams +import io.emeraldpay.dshackle.upstream.ethereum.EthereumReader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux @@ -30,7 +32,7 @@ import reactor.core.publisher.Flux class UpstreamsMock implements Upstreams { private Map target = [:] - private Map upstreams = [:] + private Map upstreams = [:] UpstreamsMock(Chain chain, Upstream up) { addUpstream(chain, up) @@ -40,15 +42,20 @@ class UpstreamsMock implements Upstreams { addUpstream(chain2, up2) } - AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { + AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull EthereumUpstream up) { if (!upstreams.containsKey(chain)) { - upstreams[chain] = new EthereumChainUpstreams(chain, [up], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) + upstreams[chain] = new AggregatedEthereumUpstreamsMock(chain, [up], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) + upstreams[chain].start() } else { upstreams[chain].addUpstream(up) } return upstreams[chain] } + void setReader(@NotNull Chain chain, EthereumReader reader) { + upstreams[chain].customReader = reader + } + @Override AggregatedUpstream getUpstream(@NotNull Chain chain) { return upstreams[chain] @@ -78,4 +85,21 @@ class UpstreamsMock implements Upstreams { return upstreams.containsKey(chain) } + static class AggregatedEthereumUpstreamsMock extends AggregatedEthereumUpstreams { + + EthereumReader customReader = null + + AggregatedEthereumUpstreamsMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches, @NotNull ObjectMapper objectMapper) { + super(chain, upstreams, caches, objectMapper) + } + + @Override + EthereumReader getReader() { + if (customReader != null) { + return customReader + } + return super.getReader() + } + } + } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy index 80f80901..0414a71c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AggregatedUpstreamSpec.groovy @@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi -import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstreams +import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams import io.emeraldpay.grpc.Chain import spock.lang.Specification @@ -32,7 +32,7 @@ class AggregatedUpstreamSpec extends Specification { setup: def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, Stub(DirectEthereumApi), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new EthereumChainUpstreams(Chain.ETHEREUM, [up1, up2], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) + def aggr = new AggregatedEthereumUpstreams(Chain.ETHEREUM, [up1, up2], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy new file mode 100644 index 00000000..18f5afa4 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy @@ -0,0 +1,237 @@ +/** + * 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.TxMemCache +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.TxContainer +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.Upstream +import io.infinitape.etherjar.domain.Address +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.domain.Wei +import io.infinitape.etherjar.rpc.ReactorRpcClient +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +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 + } + + def "Block by Id reads from cache"() { + setup: + def memCache = Mock(BlocksMemCache) { + 1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson, TestingCommons.objectMapper())) + } + def caches = Caches.newBuilder() + .setBlockByHash(memCache) + .setObjectMapper(TestingCommons.objectMapper()) + .build() + def reader = new EthereumReader(Stub(Upstream), caches, TestingCommons.objectMapper()) + + when: + def act = reader.blocksById().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) + .setObjectMapper(TestingCommons.objectMapper()) + .build() + def rpcClient = Stub(ReactorRpcClient) + def api = TestingCommons.api(rpcClient) + api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) + + def upstream = TestingCommons.aggregatedUpstream(api) + def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) + + when: + def act = reader.blocksById().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) + .setObjectMapper(TestingCommons.objectMapper()) + .build() + def rpcClient = Stub(ReactorRpcClient) + def api = TestingCommons.api(rpcClient) + api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) + + def upstream = TestingCommons.aggregatedUpstream(api) + def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) + + when: + def act = reader.blocksById().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, TestingCommons.objectMapper())) + } + def caches = Caches.newBuilder() + .setBlockByHash(memCache) + .setObjectMapper(TestingCommons.objectMapper()) + .build() + def reader = new EthereumReader(Stub(Upstream), caches, TestingCommons.objectMapper()) + + when: + def act = reader.blocksByHash().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) + .setObjectMapper(TestingCommons.objectMapper()) + .build() + def rpcClient = Stub(ReactorRpcClient) + def api = TestingCommons.api(rpcClient) + api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) + def upstream = TestingCommons.aggregatedUpstream(api) + def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) + + when: + def act = reader.blocksByHash().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, TestingCommons.objectMapper())) + } + def caches = Caches.newBuilder() + .setTxByHash(memCache) + .setObjectMapper(TestingCommons.objectMapper()) + .build() + def reader = new EthereumReader(Stub(Upstream), caches, TestingCommons.objectMapper()) + + 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) + .setObjectMapper(TestingCommons.objectMapper()) + .build() + + def rpcClient = Stub(ReactorRpcClient) + def api = TestingCommons.api(rpcClient) + api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson) + def upstream = TestingCommons.aggregatedUpstream(api) + def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) + + when: + def act = reader.txByHash().read(txJson.hash).block() + + then: + act == txJson + } + + def "Caches balance until block mined"() { + setup: + def rpcClient = Stub(ReactorRpcClient) + def api = TestingCommons.api(rpcClient) + api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0x10") + api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0xff") + def upstream = TestingCommons.upstream(api) + def reader = new EthereumReader(upstream, Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) + 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, TestingCommons.objectMapper())) + act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block() + + then: + act == Wei.from("0xff") + } +}