diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt new file mode 100644 index 00000000..b3b10681 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt @@ -0,0 +1,27 @@ +package io.emeraldpay.dshackle.cache + +import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono + +/** + * Connects two caches to read through them. First is cache height->hash, second is hash->block. + */ +class BlockByHeight( + private val heights: Reader, + private val blocks: Reader> +): Reader> { + + companion object { + private val log = LoggerFactory.getLogger(BlockByHeight::class.java) + } + + override fun read(key: Long): Mono> { + return heights.read(key) + .flatMap { blocks.read(it) } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt index f44b7e36..dd52b152 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.cache +import io.emeraldpay.dshackle.reader.Reader import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson @@ -25,13 +26,13 @@ import java.util.concurrent.ConcurrentLinkedQueue class BlocksMemCache( val maxSize: Int = 64 -) { +): Reader> { private val mapping = ConcurrentHashMap>() private val queue = ConcurrentLinkedQueue() - fun get(hash: BlockHash): Mono> { - return Mono.justOrEmpty(mapping[hash]) + override fun read(key: BlockHash): Mono> { + return Mono.justOrEmpty(mapping[key]) } fun add(block: BlockJson) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt new file mode 100644 index 00000000..7aaf20c6 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt @@ -0,0 +1,38 @@ +package io.emeraldpay.dshackle.cache + +import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.util.concurrent.ConcurrentHashMap + +/** + * Memory cache for blocks heights, keeps mapping height->hash. + */ +class HeightCache( + val maxSize: Int = 256 +): Reader { + + companion object { + private val log = LoggerFactory.getLogger(HeightCache::class.java) + } + + private val heights = ConcurrentHashMap() + + override fun read(key: Long): Mono { + return Mono.justOrEmpty(heights[key]) + } + + fun add(block: BlockJson) { + heights[block.number] = block.hash + + // evict old numbers if full + var dropHeight = block.number - maxSize + while (heights.size > maxSize && dropHeight < block.number) { + heights.remove(dropHeight) + dropHeight++ + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt deleted file mode 100644 index 7c1e4925..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt +++ /dev/null @@ -1,32 +0,0 @@ -/** - * Copyright (c) 2019 ETCDEV GmbH - * - * 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.cache.BlocksMemCache -import io.infinitape.etherjar.domain.BlockHash -import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson -import reactor.core.publisher.Mono - -class BlockCacheReader( - val cache: BlocksMemCache -): Reader> { - - override fun read(key: BlockHash): Mono> { - return cache.get(key) - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt index cf7e4bc6..73b95184 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt @@ -16,18 +16,11 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.cache.BlockByHeight import io.emeraldpay.dshackle.cache.BlocksMemCache +import io.emeraldpay.dshackle.cache.HeightCache import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.reader.BlockCacheReader -import io.emeraldpay.dshackle.reader.CompoundReader -import io.emeraldpay.dshackle.reader.Reader -import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead -import io.infinitape.etherjar.domain.BlockHash -import io.infinitape.etherjar.domain.TransactionId -import io.infinitape.etherjar.rpc.json.BlockJson -import io.infinitape.etherjar.rpc.json.TransactionRefJson -import org.reactivestreams.Publisher import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux @@ -42,11 +35,9 @@ abstract class AggregatedUpstream( val objectMapper: ObjectMapper ): Upstream, Lifecycle { - private val blocksCache = BlocksMemCache() private var cacheSubscription: Disposable? = null - private val blockReader: Reader> = CompoundReader( - listOf(BlockCacheReader(blocksCache)) - ) + private val blockReaderByHash = BlocksMemCache() + private val blockReaderByHeight = HeightCache() var cache: CachingEthereumApi = CachingEthereumApi.empty() private val reconfigLock = ReentrantLock() private var callMethods: CallMethods? = null @@ -118,9 +109,10 @@ abstract class AggregatedUpstream( reconfigLock.withLock { cacheSubscription?.dispose() cacheSubscription = head.getFlux().subscribe { - blocksCache.add(it) + blockReaderByHash.add(it) + blockReaderByHeight.add(it) } - cache = CachingEthereumApi(objectMapper, blockReader, head) + cache = CachingEthereumApi(objectMapper, blockReaderByHash, BlockByHeight(blockReaderByHeight, blockReaderByHash), head) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt index 65ee5c5f..7c86f4b4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt @@ -29,11 +29,13 @@ import io.infinitape.etherjar.rpc.json.ResponseJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import reactor.core.publisher.Mono +import java.math.BigInteger import java.util.function.Function open class CachingEthereumApi( private val objectMapper: ObjectMapper, private val cache: Reader>, + private val cacheHeight: Reader>, private val head: EthereumHead ): EthereumApi(objectMapper) { @@ -42,7 +44,7 @@ open class CachingEthereumApi( @JvmStatic fun empty(): CachingEthereumApi { - return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyEthereumHead()) + return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyReader(), EmptyEthereumHead()) } } @@ -63,6 +65,21 @@ open class CachingEthereumApi( Mono.empty() } else Mono.empty() + "eth_getBlockByNumber" -> + if (params.size == 2 && (params[1] == "false" || params[1] == false)) + Mono.just(params[0]) + .map { HexQuantity.from(it as String) } + .filter { + it.value < BigInteger.valueOf(Long.MAX_VALUE) + } + .map { it.value.toLong() } + .flatMap(cacheHeight::read) + .map(toJson(id)) + .onErrorResume { t -> + log.warn("Error during read from cache", t) + Mono.empty() + } + else Mono.empty() else -> Mono.empty() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlockByHeightSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlockByHeightSpec.groovy new file mode 100644 index 00000000..275c4d9f --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlockByHeightSpec.groovy @@ -0,0 +1,138 @@ +package io.emeraldpay.dshackle.cache + +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import spock.lang.Specification + +class BlockByHeightSpec extends Specification { + + String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + + def "Fetch with all data available"() { + setup: + def blocks = new BlocksMemCache() + def heights = new HeightCache() + + def block = new BlockJson() + block.number = 100 + block.hash = BlockHash.from(hash1) + + blocks.add(block) + heights.add(block) + + def blocksByHeight = new BlockByHeight(heights, blocks) + + when: + def act = blocksByHeight.read(100).block() + then: + act == block + } + + def "Fetch correct blocks if multiple"() { + setup: + def blocks = new BlocksMemCache() + def heights = new HeightCache() + + def block1 = new BlockJson() + block1.number = 100 + block1.hash = BlockHash.from(hash1) + def block2 = new BlockJson() + block2.number = 101 + block2.hash = BlockHash.from(hash2) + + blocks.add(block1) + heights.add(block1) + blocks.add(block2) + heights.add(block2) + + def blocksByHeight = new BlockByHeight(heights, blocks) + + when: + def act = blocksByHeight.read(100).block() + then: + act == block1 + + when: + act = blocksByHeight.read(101).block() + then: + act == block2 + } + + def "Fetch last block if updated"() { + setup: + def blocks = new BlocksMemCache() + def heights = new HeightCache() + + def block1 = new BlockJson() + block1.number = 100 + block1.hash = BlockHash.from(hash1) + def block2 = new BlockJson() + block2.number = 100 + block2.hash = BlockHash.from(hash2) + + blocks.add(block1) + heights.add(block1) + blocks.add(block2) + heights.add(block2) + + def blocksByHeight = new BlockByHeight(heights, blocks) + + when: + def act = blocksByHeight.read(100).block() + then: + act == block2 + } + + def "Fetch nothing if block expired"() { + setup: + def blocks = new BlocksMemCache() + def heights = new HeightCache() + + def block = new BlockJson() + block.number = 100 + block.hash = BlockHash.from(hash1) + + // add only to heights + heights.add(block) + + def blocksByHeight = new BlockByHeight(heights, blocks) + + when: + def act = blocksByHeight.read(100).block() + then: + act == null + } + + def "Fetch nothing if height expired"() { + setup: + def blocks = new BlocksMemCache() + def heights = new HeightCache() + + def block = new BlockJson() + block.number = 100 + block.hash = BlockHash.from(hash1) + + // add only to blocks + blocks.add(block) + + def blocksByHeight = new BlockByHeight(heights, blocks) + + when: + def act = blocksByHeight.read(100).block() + then: + act == null + } + + def "Fetch nothing if both empty"() { + setup: + def blocks = new BlocksMemCache() + def heights = new HeightCache() + def blocksByHeight = new BlockByHeight(heights, blocks) + when: + def act = blocksByHeight.read(100).block() + then: + act == null + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy index 3b2eb0b0..eacfc945 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.cache import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson import spock.lang.Specification class BlocksMemCacheSpec extends Specification { @@ -30,13 +31,13 @@ class BlocksMemCacheSpec extends Specification { def "Add and read"() { setup: def cache = new BlocksMemCache() - def block = new BlockJson() + def block = new BlockJson() block.number = 100 block.hash = BlockHash.from(hash1) when: cache.add(block) - def act = cache.get(BlockHash.from(hash1)).block() + def act = cache.read(BlockHash.from(hash1)).block() then: act == block } @@ -48,20 +49,21 @@ class BlocksMemCacheSpec extends Specification { when: [hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i -> - def block = new BlockJson() + def block = new BlockJson() block.number = 100 + i block.hash = BlockHash.from(hash) cache.add(block) } - def act1 = cache.get(BlockHash.from(hash1)).block() - def act2 = cache.get(BlockHash.from(hash2)).block() - def act3 = cache.get(BlockHash.from(hash3)).block() - def act4 = cache.get(BlockHash.from(hash4)).block() + def act1 = cache.read(BlockHash.from(hash1)).block() + def act2 = cache.read(BlockHash.from(hash2)).block() + def act3 = cache.read(BlockHash.from(hash3)).block() + def act4 = cache.read(BlockHash.from(hash4)).block() then: act2.hash.toHex() == hash2 act3.hash.toHex() == hash3 act4.hash.toHex() == hash4 act1 == null } + } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/HeightCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightCacheSpec.groovy new file mode 100644 index 00000000..ff4be861 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightCacheSpec.groovy @@ -0,0 +1,61 @@ +package io.emeraldpay.dshackle.cache + +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import spock.lang.Specification + +class HeightCacheSpec extends Specification { + + String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" + String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" + + def "Add and read"() { + setup: + def cache = new HeightCache() + + when: + [hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i -> + def block = new BlockJson() + block.number = 100 + i + block.hash = BlockHash.from(hash) + cache.add(block) + } + + def act1 = cache.read(100).block() + def act2 = cache.read(101).block() + def act3 = cache.read(102).block() + def act4 = cache.read(103).block() + then: + act1.toHex() == hash1 + act2.toHex() == hash2 + act3.toHex() == hash3 + act4.toHex() == hash4 + } + + def "Keeps only configured amount"() { + setup: + def cache = new HeightCache(3) + [hash1] + + when: + [hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i -> + def block = new BlockJson() + block.number = 100 + i + block.hash = BlockHash.from(hash) + cache.add(block) + } + + def act1 = cache.read(100).block() + def act2 = cache.read(101).block() + def act3 = cache.read(102).block() + def act4 = cache.read(103).block() + then: + act1 == null + act2.toHex() == hash2 + act3.toHex() == hash3 + act4.toHex() == hash4 + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy index bc4b1344..03f1c5e1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy @@ -1,7 +1,8 @@ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.cache.BlockByHeight import io.emeraldpay.dshackle.cache.BlocksMemCache -import io.emeraldpay.dshackle.reader.BlockCacheReader +import io.emeraldpay.dshackle.cache.HeightCache import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead @@ -23,6 +24,7 @@ class CachingEthereumApiSpec extends Specification { def api = new CachingEthereumApi( TestingCommons.objectMapper(), new EmptyReader>(), + new EmptyReader<>(), head ) 1 * head.getFlux() >> Flux.just(new BlockJson(number: 100)) @@ -42,6 +44,7 @@ class CachingEthereumApiSpec extends Specification { def api = new CachingEthereumApi( TestingCommons.objectMapper(), new EmptyReader>(), + new EmptyReader<>(), head ) when: @@ -53,13 +56,14 @@ class CachingEthereumApiSpec extends Specification { .verify(Duration.ofSeconds(3)) } - def "Return block when cached"() { + def "Return block by hash when cached"() { setup: def cache = new BlocksMemCache(); def head = Mock(EthereumHead.class) def api = new CachingEthereumApi( TestingCommons.objectMapper(), - new BlockCacheReader(cache), + cache, + new EmptyReader<>(), head ) cache.add(new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58"))) @@ -74,4 +78,29 @@ class CachingEthereumApiSpec extends Specification { .verify(Duration.ofSeconds(3)) } + def "Return block by height when cached"() { + setup: + def blocksCache = new BlocksMemCache() + def heightCache = new HeightCache() + def head = Mock(EthereumHead.class) + def api = new CachingEthereumApi( + TestingCommons.objectMapper(), + blocksCache, + new BlockByHeight(heightCache, blocksCache), + head + ) + def block = new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) + heightCache.add(block) + blocksCache.add(block) + + when: + def act = api.execute(1, "eth_getBlockByNumber", ["0x64", false]).map { new String(it)} + + then: + StepVerifier.create(act) + .expectNext('{"jsonrpc":"2.0","id":1,"result":{"number":"0x64","hash":"0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58","transactions":[],"uncles":[]}}') + .expectComplete() + .verify(Duration.ofSeconds(3)) + } + }