From 50f6d7b2feca033d04004d65c1d33deff8c5c64b Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sat, 20 Mar 2021 22:34:22 -0400 Subject: [PATCH] problem: doesn't select upstream by available block hash, see EIP-1898 rel: #81 --- .../io/emeraldpay/dshackle/cache/Caches.kt | 26 +++- .../dshackle/cache/CachesFactory.kt | 1 + .../dshackle/cache/HeightByHashAdding.kt | 80 +++++++++++ .../dshackle/cache/HeightByHashCache.kt | 26 ++++ .../dshackle/cache/HeightByHashMemCache.kt | 44 ++++++ .../dshackle/cache/HeightByHashRedisCache.kt | 102 ++++++++++++++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 46 +++++-- .../upstream/calls/EthereumCallSelector.kt | 48 +++++-- .../upstream/ethereum/EthereumReader.kt | 17 ++- .../cache/HeightByHashAddingSpec.groovy | 130 ++++++++++++++++++ .../cache/HeightByHashRedisCacheSpec.groovy | 72 ++++++++++ .../dshackle/rpc/NativeCallSpec.groovy | 13 +- .../calls/EthereumCallSelectorSpec.groovy | 56 ++++++-- 13 files changed, 614 insertions(+), 47 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashAdding.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashCache.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashMemCache.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashRedisCache.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashRedisCacheSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt index 295781c4..9164cd37 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -35,7 +35,8 @@ open class Caches( private val memTxsByHash: TxMemCache, private val redisBlocksByHash: BlocksRedisCache?, private val redisTxsByHash: TxRedisCache?, - private val redisReceipts: ReceiptRedisCache? + private val redisReceipts: ReceiptRedisCache?, + private val redisHeightByHashCache: HeightByHashRedisCache? ) { companion object { @@ -52,6 +53,8 @@ open class Caches( } } + private val memHeightByHash: HeightByHashMemCache = HeightByHashMemCache() + private val blocksByHash: Reader private val txsByHash: Reader private val receiptByHash: Reader @@ -105,6 +108,9 @@ open class Caches( fun cache(tag: Tag, block: BlockContainer) { val job = ArrayList>() + + redisHeightByHashCache?.add(block)?.let(job::add) + if (tag == Tag.LATEST) { //for LATEST data cache it in memory, it may be short living so better to avoid Redis memoizeBlock(block) @@ -147,6 +153,7 @@ open class Caches( */ fun memoizeBlock(block: BlockContainer) { memBlocksByHash.add(block) + memHeightByHash.add(block) val replaced = blocksByHeight.add(block) //evict cached transactions if an existing block was updated replaced?.let { evict(it) } @@ -193,6 +200,14 @@ open class Caches( return receiptByHash } + fun getLastHeightByHash(): Reader { + return memHeightByHash + } + + fun getRedisHeightByHash(): HeightByHashCache? { + return redisHeightByHashCache + } + enum class Tag { /** * Latest data produced by blockchain @@ -212,6 +227,7 @@ open class Caches( private var redisBlocksByHash: BlocksRedisCache? = null private var redisTxsByHash: TxRedisCache? = null private var redisReceiptCache: ReceiptRedisCache? = null + private var redisHeightByHashCache: HeightByHashRedisCache? = null fun setBlockByHash(cache: BlocksMemCache): Builder { blocksByHash = cache @@ -243,6 +259,11 @@ open class Caches( return this } + fun setHeightByHash(cache: HeightByHashRedisCache): Builder { + redisHeightByHashCache = cache + return this + } + fun build(): Caches { if (blocksByHash == null) { blocksByHash = BlocksMemCache() @@ -253,7 +274,8 @@ open class Caches( if (txsByHash == null) { txsByHash = TxMemCache() } - return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash, redisReceiptCache) + return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, + redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache) } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt index 71d32c60..21df05fe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt @@ -96,6 +96,7 @@ class CachesFactory( caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain)) caches.setTxByHash(TxRedisCache(redis.reactive(), chain)) caches.setReceipts(ReceiptRedisCache(redis.reactive(), chain)) + caches.setHeightByHash(HeightByHashRedisCache(redis.reactive(), chain)) } return caches.build() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashAdding.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashAdding.kt new file mode 100644 index 00000000..fce0e617 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashAdding.kt @@ -0,0 +1,80 @@ +/** + * 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.cache + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono + +/** + * Height By Hash reader, that adds to Redis cache missing values. + * + * Tries to read value from current memory cache, then from redis if it's available, and if both empty read from blockchain. + * With the latter case it adds read value back to the redis cache (again, if available). + * + */ +class HeightByHashAdding( + private val mem: Reader, + private val redis: HeightByHashCache?, + private val upstreamReader: Reader +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(HeightByHashAdding::class.java) + } + + constructor(caches: Caches, upstreamReader: Reader) : + this(caches.getLastHeightByHash(), caches.getRedisHeightByHash(), upstreamReader) + + private val delegate: Reader + + init { + if (redis != null) { + delegate = object : Reader { + override fun read(key: BlockId): Mono { + return mem.read(key) + .switchIfEmpty( + Mono.just(key) + .flatMap { redis.read(it) } + ) + .switchIfEmpty( + Mono.just(key) + .flatMap { upstreamReader.read(it) } + .flatMap { redis.add(it).then(Mono.just(it.height)) } + ) + } + } + } else { + delegate = object : Reader { + override fun read(key: BlockId): Mono { + return mem.read(key) + .switchIfEmpty( + Mono.just(key) + .flatMap { upstreamReader.read(it) } + .map { it.height } + ) + } + } + } + } + + override fun read(key: BlockId): Mono { + return delegate.read(key) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashCache.kt new file mode 100644 index 00000000..3dfe08e5 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashCache.kt @@ -0,0 +1,26 @@ +/** + * 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.cache + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader +import reactor.core.publisher.Mono + +interface HeightByHashCache : Reader { + + fun add(block: BlockContainer): Mono +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashMemCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashMemCache.kt new file mode 100644 index 00000000..e63ef853 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashMemCache.kt @@ -0,0 +1,44 @@ +/** + * 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.cache + +import com.github.benmanes.caffeine.cache.Caffeine +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono + +class HeightByHashMemCache( + maxSize: Int = 256 +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(HeightByHashMemCache::class.java) + } + + private val heights = Caffeine.newBuilder() + .maximumSize(maxSize.toLong()) + .build() + + override fun read(key: BlockId): Mono { + return Mono.justOrEmpty(heights.getIfPresent(key)) + } + + fun add(block: BlockContainer) { + heights.put(block.hash, block.height) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashRedisCache.kt new file mode 100644 index 00000000..17f1ce09 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashRedisCache.kt @@ -0,0 +1,102 @@ +/** + * 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.cache + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.grpc.Chain +import io.lettuce.core.api.reactive.RedisReactiveCommands +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.nio.ByteBuffer +import java.nio.ByteOrder +import java.util.concurrent.TimeUnit + +/** + * Cache for height by hash. + * Different from blocks cache because for height we don't really care about eviction and replaced blocks (also a fall-back block + * reader would use full block's cache to find out height). + */ +class HeightByHashRedisCache( + private val redis: RedisReactiveCommands, + private val chain: Chain +) : Reader, HeightByHashCache { + + companion object { + private val log = LoggerFactory.getLogger(HeightByHashRedisCache::class.java) + + private const val MAX_CACHE_TIME_MINUTES = 60L * 4 + } + + override fun read(key: BlockId): Mono { + return redis.get(key(key)) + .flatMap { data -> + Mono.justOrEmpty(fromBytes(data)) as Mono + }.onErrorResume { + log.warn("Failed to read Block Height. ${it.javaClass}:${it.message}") + Mono.empty() + } + } + + override fun add(block: BlockContainer): Mono { + return Mono.just(block) + .flatMap { block -> + // even if block replaced, the mapping hash-long is still valid, so can be cached for long time + // even for fresh blocks + val ttl = TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES) + + val key = key(block.hash) + val value = asBytes(block.height) + redis.setex(key, ttl, value) + } + .doOnError { + log.warn("Failed to save Block Height. ${it.javaClass}:${it.message}") + } + //if failed to cache, just continue without it + .onErrorResume { + Mono.empty() + } + .then() + } + + fun asBytes(value: Long): ByteArray { + val result = ByteArray(8) + val bb = ByteBuffer.allocate(8) + .order(ByteOrder.BIG_ENDIAN) + bb.asLongBuffer() + .put(value) + bb.get(result) + return result + } + + fun fromBytes(value: ByteArray): Long? { + if (value.size != 8) { + return null + } + return ByteBuffer.wrap(value) + .order(ByteOrder.BIG_ENDIAN) + .asLongBuffer() + .get() + } + + /** + * Key in Redis + */ + fun key(hash: BlockId): String { + return "height:${chain.id}:${hash.toHex()}" + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index cb6141e7..a3e3053b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -40,6 +41,7 @@ import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import reactor.core.publisher.* import java.lang.Exception +import java.util.* @Service open class NativeCall( @@ -50,7 +52,18 @@ open class NativeCall( private val objectMapper: ObjectMapper = Global.objectMapper var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() - private val ethereumCallSelector = EthereumCallSelector() + private val ethereumCallSelectors = EnumMap(Chain::class.java) + + init { + multistreamHolder.observeChains().subscribe { chain -> + if (!ethereumCallSelectors.containsKey(chain)) { + multistreamHolder.getUpstream(chain)?.let { up -> + val reader = up.cast(EthereumMultistream::class.java).getReader() + ethereumCallSelectors[chain] = EthereumCallSelector(reader.heightByHash()) + } + } + } + } open fun nativeCall(requestMono: Mono): Flux { return nativeCallResult(requestMono) @@ -121,27 +134,30 @@ open class NativeCall( } fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: Multistream): Flux> { - return Flux.fromIterable(request.itemsList).map { + val chain = Chain.byId(request.chainValue) + return Flux.fromIterable(request.itemsList).flatMap { val method = it.method val params = it.payload.toStringUtf8() // for ethereum the actual block needed for the call may be specified in the call parameters - val callSpecificMather = if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) { - ethereumCallSelector.getMatcher(method, params, upstream.getHead()) + val callSpecificMatcher: Mono = if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) { + ethereumCallSelectors[chain]?.getMatcher(method, params, upstream.getHead()) } else { null + } ?: Mono.empty() + + callSpecificMatcher.defaultIfEmpty(Selector.empty).map { csm -> + val matcher = Selector.Builder() + .withMatcher(csm) + .forMethod(method) + .forLabels(Selector.convertToMatcher(request.selector)) + .build() + + val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum() + callQuorum.init(upstream.getHead()) + + CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params)) } - - val matcher = Selector.Builder() - .withMatcher(callSpecificMather) - .forMethod(method) - .forLabels(Selector.convertToMatcher(request.selector)) - .build() - - val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum() - callQuorum.init(upstream.getHead()) - - CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params)) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt index 8d548a11..d31f52fc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt @@ -16,20 +16,26 @@ package io.emeraldpay.dshackle.upstream.calls import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector import io.infinitape.etherjar.hex.HexQuantity import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono import java.util.* /** * Get a matcher based on a criteria provided with a RPC request. I.e. when the client requests data for "latest", or "0x19f816" block. * The implementation is specific for Ethereum. */ -class EthereumCallSelector { +class EthereumCallSelector( + private val heightReader: Reader +) { companion object { private val log = LoggerFactory.getLogger(EthereumCallSelector::class.java) + // ref https://eth.wiki/json-rpc/API#the-default-block-parameter private val TAG_METHODS = listOf( "eth_getBalance", @@ -46,20 +52,20 @@ class EthereumCallSelector { * @param method JSON RPC name * @param params JSON-encoded list of parameters for the method */ - fun getMatcher(method: String, params: String, head: Head): Selector.Matcher? { + fun getMatcher(method: String, params: String, head: Head): Mono { if (Collections.binarySearch(TAG_METHODS, method) >= 0) { return blockTagSelector(params, 1, head) } else if (method == "eth_getStorageAt") { return blockTagSelector(params, 2, head) } - return null + return Mono.empty() } - private fun blockTagSelector(params: String, pos: Int, head: Head): Selector.Matcher? { + private fun blockTagSelector(params: String, pos: Int, head: Head): Mono { val list = objectMapper.readerFor(Any::class.java).readValues(params).readAll() if (list.size < pos + 1) { log.debug("Tag is not specified. Ignoring") - return null + return Mono.empty() } // integer block number, a string "latest", "earliest" or "pending", or an object with block reference val minHeight: Long? = when (val tag = list[pos].toString()) { @@ -76,15 +82,27 @@ class EthereumCallSelector { } else if (tag.startsWith("{") && list[pos] is Map<*, *>) { // see https://eips.ethereum.org/EIPS/eip-1898 val obj = list[pos] as Map<*, *> - if (obj.containsKey("blockNumber")) { - try { - HexQuantity.from(obj["blockNumber"].toString()).value.toLong() - } catch (t: Throwable) { - log.warn("Invalid blockNumber: $tag") - null + when { + obj.containsKey("blockNumber") -> { + try { + HexQuantity.from(obj["blockNumber"].toString()).value.toLong() + } catch (t: Throwable) { + log.warn("Invalid blockNumber: $tag") + null + } } - } else { - null + obj.containsKey("blockHash") -> { + try { + val blockId = BlockId.from(obj["blockHash"].toString()) + return heightReader.read(blockId) + .switchIfEmpty(Mono.justOrEmpty(head.getCurrentHeight())) + .map { Selector.HeightMatcher(it) } + } catch (t: Throwable) { + log.warn("Invalid blockHash: $tag") + null + } + } + else -> null } } else { log.debug("Invalid tag: $tag") @@ -92,9 +110,9 @@ class EthereumCallSelector { } } return if (minHeight != null && minHeight >= 0) { - Selector.HeightMatcher(minHeight) + Mono.just(Selector.HeightMatcher(minHeight)) } else { - null + Mono.empty() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt index 052e3ac2..8bac7a20 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt @@ -19,6 +19,7 @@ import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache +import io.emeraldpay.dshackle.cache.HeightByHashAdding import io.emeraldpay.dshackle.data.* import io.emeraldpay.dshackle.reader.* import io.emeraldpay.dshackle.upstream.Multistream @@ -77,6 +78,13 @@ open class EthereumReader( private val txHashToId = Function { hash -> TxId.from(hash) } private val idToTxHash = Function { id -> TransactionId.from(id.value) } + private val blocksByIdAsCont = CompoundReader( + caches.getBlocksByHash(), + RekeyingReader(idToBlockHash, directReader.blockReader) + ) + + private val heightByHash = HeightByHashAdding(caches, blocksByIdAsCont) + fun blocksByHashAsCont(): Reader { return CompoundReader( RekeyingReader(blockHashToId, caches.getBlocksByHash()), @@ -99,10 +107,7 @@ open class EthereumReader( } open fun blocksByIdAsCont(): Reader { - return CompoundReader( - caches.getBlocksByHash(), - RekeyingReader(idToBlockHash, directReader.blockReader) - ) + return blocksByIdAsCont } open fun blocksByHeightAsCont(): Reader { @@ -140,6 +145,10 @@ open class EthereumReader( return caches.getReceipts() } + fun heightByHash(): Reader { + return heightByHash + } + override fun isRunning(): Boolean { //TODO should be always running? return up.isRunning diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy new file mode 100644 index 00000000..541a5e61 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy @@ -0,0 +1,130 @@ +/** + * 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.cache + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import reactor.core.publisher.Mono +import spock.lang.Specification +import io.emeraldpay.dshackle.reader.Reader + +import java.time.Instant + +class HeightByHashAddingSpec extends Specification { + + def block = new BlockContainer( + 12079192L, BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"), + BigInteger.ONE, Instant.now(), false, "".bytes, null, [] + ) + + def "use memory if available"() { + setup: + def mem = new HeightByHashMemCache() + def upstream = Mock(Reader) + def reader = new HeightByHashAdding( + mem, null, upstream + ) + + when: + mem.add(block) + def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block() + then: + act == 12079192L + 0 * upstream.read(_) + } + + def "call remote if not in memory and no redis"() { + setup: + def mem = new HeightByHashMemCache() + def upstream = Mock(Reader) + def reader = new HeightByHashAdding( + mem, null, upstream + ) + + when: + def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block() + then: + act == 12079192L + 1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(block) + } + + def "get from redis if not in memory"() { + setup: + def mem = new HeightByHashMemCache() + def upstream = Mock(Reader) + def redis = Mock(HeightByHashCache) + def reader = new HeightByHashAdding( + mem, redis, upstream + ) + + when: + def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block() + then: + act == 12079192L + 0 * upstream.read(_) + 1 * redis.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(12079192L) + } + + def "call remote if not in memory and not in redis, add to redis"() { + setup: + def mem = new HeightByHashMemCache() + def upstream = Mock(Reader) + def redis = Mock(HeightByHashCache) + def reader = new HeightByHashAdding( + mem, redis, upstream + ) + + when: + def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block() + then: + act == 12079192L + 1 * redis.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty() + 1 * redis.add(block) >> Mono.just(true).then() + 1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(block) + } + + def "empty is nowhere found, without redis"() { + setup: + def mem = new HeightByHashMemCache() + def upstream = Mock(Reader) + def reader = new HeightByHashAdding( + mem, null, upstream + ) + + when: + def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block() + then: + act == null + 1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty() + } + + def "empty is nowhere found, with redis"() { + setup: + def mem = new HeightByHashMemCache() + def upstream = Mock(Reader) + def redis = Mock(HeightByHashCache) + def reader = new HeightByHashAdding( + mem, redis, upstream + ) + + when: + def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block() + then: + act == null + 1 * redis.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty() + 1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty() + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashRedisCacheSpec.groovy new file mode 100644 index 00000000..641e8dec --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashRedisCacheSpec.groovy @@ -0,0 +1,72 @@ +/** + * 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.cache + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.test.IntegrationTestingCommons +import io.emeraldpay.grpc.Chain +import io.lettuce.core.api.StatefulRedisConnection +import spock.lang.IgnoreIf +import spock.lang.Specification + +import java.time.Instant + +@IgnoreIf({ IntegrationTestingCommons.isDisabled("redis") }) +class HeightByHashRedisCacheSpec extends Specification { + + def block1 = new BlockContainer( + 12079192L, BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"), + BigInteger.ONE, Instant.now(), false, "".bytes, null, [] + ) + def block2 = new BlockContainer( + 12079193L, BlockId.from("0xd27944b460632699768fbfec3e5d454db590cae43d470b5f42fc4d091e372c25"), + BigInteger.ONE, Instant.now(), false, "".bytes, null, [] + ) + + StatefulRedisConnection redis + HeightByHashRedisCache cache + + def setup() { + redis = IntegrationTestingCommons.redisConnection() + redis.sync().flushdb() + cache = new HeightByHashRedisCache( + redis.reactive(), Chain.ETHEREUM + ) + } + + def "Add and read"() { + when: + cache.add(block1).subscribe() + def act = cache.read(block1.hash).block() + then: + act == 12079192L + } + + def "Add and read multiple"() { + when: + cache.add(block1).subscribe() + cache.add(block2).subscribe() + + def act = cache.read(block1.hash).block() + then: + act == 12079192L + + def act2 = cache.read(block2.hash).block() + then: + act2 == 12079193L + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 2519fbf8..23fe751b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -34,6 +34,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcResponseError +import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.test.StepVerifier import spock.lang.Ignore @@ -210,7 +211,9 @@ class NativeCallSpec extends Specification { def "Returns error for unsupported chain"() { setup: - def upstreams = Mock(MultistreamHolder) + def upstreams = Mock(MultistreamHolder) { + _ * it.observeChains() >> Flux.empty() + } def nativeCall = new NativeCall(upstreams) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() @@ -234,7 +237,9 @@ class NativeCallSpec extends Specification { def "Prepare call"() { setup: - def upstreams = Mock(MultistreamHolder) + def upstreams = Mock(MultistreamHolder) { + _ * it.observeChains() >> Flux.empty() + } def nativeCall = new NativeCall(upstreams) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() @@ -260,7 +265,9 @@ class NativeCallSpec extends Specification { def "Prepare call without payload"() { setup: - def upstreams = Mock(MultistreamHolder) + def upstreams = Mock(MultistreamHolder) { + _ * it.observeChains() >> Flux.empty() + } def nativeCall = new NativeCall(upstreams) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy index 972b7b1e..f4832bd1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy @@ -15,75 +15,115 @@ */ package io.emeraldpay.dshackle.upstream.calls +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Selector +import reactor.core.publisher.Mono import spock.lang.Specification + class EthereumCallSelectorSpec extends Specification { - EthereumCallSelector callSelector = new EthereumCallSelector() - def "Get height matcher for latest balance"() { setup: + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) def head = Mock(Head) { 1 * getCurrentHeight() >> 100 } when: - def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "latest"]', head) + def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "latest"]', head).block() then: act == new Selector.HeightMatcher(100) } def "Get height matcher for latest call"() { setup: + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) def head = Mock(Head) { 1 * getCurrentHeight() >> 100 } when: - def act = callSelector.getMatcher("eth_call", '["0x0000", "latest"]', head) + def act = callSelector.getMatcher("eth_call", '["0x0000", "latest"]', head).block() then: act == new Selector.HeightMatcher(100) } def "Get height matcher for latest storageAt"() { setup: + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) def head = Mock(Head) { 1 * getCurrentHeight() >> 100 } when: - def act = callSelector.getMatcher("eth_getStorageAt", '["0x295a70b2de5e3953354a6a8344e616ed314d7251", "0x0", "latest"]', head) + def act = callSelector.getMatcher("eth_getStorageAt", '["0x295a70b2de5e3953354a6a8344e616ed314d7251", "0x0", "latest"]', head).block() then: act == new Selector.HeightMatcher(100) } def "Get height matcher for balance on block"() { setup: + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) def head = Mock(Head) { _ * getCurrentHeight() >> 100 } when: - def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "0x40"]', head) + def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "0x40"]', head).block() then: act == new Selector.HeightMatcher(0x40) } def "No matcher for pending balance"() { setup: + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) def head = Mock(Head) { _ * getCurrentHeight() >> 100 } when: - def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "pending"]', head) + def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "pending"]', head).block() then: act == null } def "Get height matcher with EIP-1898"() { setup: + EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader)) def head = Stub(Head) when: - def act = callSelector.getMatcher("eth_call", '["0x0000", {"blockNumber": "0x100"}]', head) + def act = callSelector.getMatcher("eth_call", '["0x0000", {"blockNumber": "0x100"}]', head).block() then: act == new Selector.HeightMatcher(0x100) } + + def "Get hash matcher with EIP-1898"() { + setup: + def heights = Mock(Reader) { + 1 * it.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(12079192L) + } + EthereumCallSelector callSelector = new EthereumCallSelector(heights) + def head = Stub(Head) + when: + def act = callSelector.getMatcher("eth_call", + '["0x0000", {"blockHash": "0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"}]', head) + .block() + then: + act == new Selector.HeightMatcher(12079192) + } + + def "Match head if hash matcher for unknown hash"() { + setup: + def heights = Mock(Reader) { + 1 * it.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty() + } + EthereumCallSelector callSelector = new EthereumCallSelector(heights) + def head = Mock(Head) { + 1 * it.getCurrentHeight() >> 100 + } + when: + def act = callSelector.getMatcher("eth_call", + '["0x0000", {"blockHash": "0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"}]', head) + .block() + then: + act == new Selector.HeightMatcher(100) + } }