diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt index cc04b9c5..c92e55c5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt @@ -10,16 +10,16 @@ import reactor.core.publisher.Mono /** * Connects two caches to read through them. First is cache height->hash, second is hash->block. */ -open class BlockByHeight( +open class BlockByHeight( private val heights: Reader, - private val blocks: Reader> -): Reader> { + private val blocks: Reader> +): Reader> { companion object { private val log = LoggerFactory.getLogger(BlockByHeight::class.java) } - override fun read(key: Long): Mono> { + override fun read(key: Long): Mono> { return heights.read(key) .flatMap { blocks.read(it) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt new file mode 100644 index 00000000..1f5f2197 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt @@ -0,0 +1,54 @@ +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.TransactionJson +import org.slf4j.LoggerFactory +import org.springframework.beans.BeanUtils +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +/** + * Reads blocks with full transactions details. Based on data contained in cashes for blocks + * and transactions, i.e. two separate caches that must be provided. + * + * If source block, with just transaction hashes is not available, it returns empty + * If any of the expected block transactions is not available it returns empty + */ +class BlocksWithTxCache( + private val blocks: BlocksMemCache, + private val txes: TxMemCache +): Reader> { + + companion object { + private val log = LoggerFactory.getLogger(BlocksWithTxCache::class.java) + } + + override fun read(key: BlockHash): Mono> { + return blocks.read(key).flatMap { block -> + if (block.transactions == null || block.transactions.isEmpty()) { + // in fact it's not necessary to create a copy, made just for code clarity but may be performance loss + val fullBlock = BlockJson() + BeanUtils.copyProperties(block, fullBlock) + Mono.just(fullBlock) + } else { + Flux.fromIterable(block.transactions) + .map { it.hash } + .flatMap { txes.read(it) } + .collectList() + .flatMap { list -> + if (block.transactions.size != list.size) { + Mono.empty>() + } else { + val fullBlock = BlockJson() + BeanUtils.copyProperties(block, fullBlock) + fullBlock.transactions = list + Mono.just(fullBlock) + } + } + } + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt index d98b4d9f..f0d59e28 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -84,6 +84,14 @@ open class Caches( return txsByHash } + fun getFullBlocks(): Reader> { + return BlocksWithTxCache(blocksByHash, txsByHash) + } + + fun getFullBlocksByHeight(): Reader> { + return BlockByHeight(blocksByHeight, BlocksWithTxCache(blocksByHash, txsByHash)) + } + enum class Tag { /** * Latest data produced by blockchain diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt index bed148dc..addb97a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt @@ -45,8 +45,46 @@ open class CachingEthereumApi( } private val cacheBlocks = caches.getBlocksByHash() - private val cacheHeight = caches.getBlocksByHeight() + private val cacheBlocksByHeight = caches.getBlocksByHeight() private val cacheTx = caches.getTxByHash() + private val cacheFullBlocks = caches.getFullBlocks() + private val cacheFullBlocksByHeight = caches.getFullBlocksByHeight() + + fun readBlockByHash(id: Int, method: String, params: List): Mono { + return if (params.size == 2) { + val includeTransactions = params[1].toString().toBoolean() + val cache = if (includeTransactions) { + cacheFullBlocks + } else { + cacheBlocks + } + Mono.just(params[0]) + .map { BlockHash.from(it as String) } + .flatMap(cache::read) + .transform(converter(id)) + .transform(finalizer()) + } + else Mono.empty() + } + + fun readBlockByNumber(id: Int, method: String, params: List): Mono { + return if (params.size == 2) { + val includeTransactions = params[1].toString().toBoolean() + val cache = if (includeTransactions) { + cacheFullBlocksByHeight + } else { + cacheBlocksByHeight + } + Mono.just(params[0]) + .map { HexQuantity.from(it as String) } + .filter { it.value < BigInteger.valueOf(Long.MAX_VALUE) } + .map { it.value.toLong() } + .flatMap(cache::read) + .transform(converter(id)) + .transform(finalizer()) + } + else Mono.empty() + } override fun execute(id: Int, method: String, params: List): Mono { return when (method) { @@ -54,24 +92,8 @@ open class CachingEthereumApi( head.getFlux().next() .map { HexQuantity.from(it.number).toHex() } .map(toJson(id)) - "eth_getBlockByHash" -> - if (params.size == 2 && (params[1] == "false" || params[1] == false)) - Mono.just(params[0]) - .map { BlockHash.from(it as String) } - .flatMap(cacheBlocks::read) - .transform(converter(id)) - .transform(finalizer()) - 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) - .transform(converter(id)) - .transform(finalizer()) - else Mono.empty() + "eth_getBlockByHash" -> readBlockByHash(id, method, params) + "eth_getBlockByNumber" -> readBlockByNumber(id, method, params) "eth_getTransactionByHash" -> if (params.size == 1) Mono.just(params[0]) diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksWithTxCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksWithTxCacheSpec.groovy new file mode 100644 index 00000000..e2482976 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksWithTxCacheSpec.groovy @@ -0,0 +1,193 @@ +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.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import spock.lang.Specification + +class BlocksWithTxCacheSpec extends Specification { + + // sorted + String hash1 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + String hash3 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" + String hash4 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + + def tx1 = new TransactionJson().with { + it.blockNumber = 100 + it.blockHash = BlockHash.from(hash1) + it.hash = TransactionId.from(hash1) + it.nonce = 1 + it + } + def tx2 = new TransactionJson().with { + it.blockNumber = 100 + it.blockHash = BlockHash.from(hash1) + it.hash = TransactionId.from(hash2) + it.nonce = 2 + it + } + def tx3 = new TransactionJson().with { + it.blockNumber = 101 + it.blockHash = BlockHash.from(hash3) + it.hash = TransactionId.from(hash3) + it.nonce = 3 + it + } + + // no block + def tx4 = new TransactionJson().with { + it.blockNumber = 102 + it.blockHash = BlockHash.from(hash4) + it.hash = TransactionId.from(hash4) + it.nonce = 4 + it + } + + // two transactiosn, tx1 and tx2 + def block1 = new BlockJson().with { + it.number = 100 + it.hash = BlockHash.from(hash1) + it.transactions = [ + new TransactionRefJson(tx1.hash), + new TransactionRefJson(tx2.hash) + ] + it + } + + // one transaction, tx3 + def block2 = new BlockJson().with { + it.number = 101 + it.hash = BlockHash.from(hash3) + it.transactions = [ + new TransactionRefJson(tx3.hash) + ] + it + } + + // no transactions + def block3 = new BlockJson().with { + it.number = 102 + it.hash = BlockHash.from(hash4) + it.transactions = [] + it + } + + + def "Read transactions and return full block"() { + setup: + def txes = new TxMemCache() + def blocks = new BlocksMemCache() + + txes.add(tx1) + txes.add(tx2) + txes.add(tx3) + txes.add(tx4) + blocks.add(block1) + blocks.add(block2) + blocks.add(block3) + + def full = new BlocksWithTxCache(blocks, txes) + + when: + def act = full.read(block1.hash).block() + + then: + act != null + act.hash == BlockHash.from(hash1) + act.number == 100 + act.transactions.size() == 2 + + when: + def transactions = act.transactions.sort { it.hash } + then: + transactions[0] instanceof TransactionJson + transactions[1] instanceof TransactionJson + //verify it didn't update original block + block1.transactions[0] instanceof TransactionRefJson + block1.transactions[1] instanceof TransactionRefJson + with(transactions[0]) { + hash == TransactionId.from(hash1) + nonce == 1 + } + with(transactions[1]) { + hash == TransactionId.from(hash2) + nonce == 2 + } + + // request second block + when: + act = full.read(block2.hash).block() + then: + act != null + act.hash == BlockHash.from(hash3) + act.number == 101 + act.transactions.size() == 1 + with(act.transactions[0]) { + hash == TransactionId.from(hash3) + nonce == 3 + } + } + + def "Read block without transactions"() { + setup: + def txes = new TxMemCache() + def blocks = new BlocksMemCache() + + txes.add(tx1) + txes.add(tx2) + txes.add(tx3) + txes.add(tx4) + blocks.add(block1) + blocks.add(block2) + blocks.add(block3) + + def full = new BlocksWithTxCache(blocks, txes) + + when: + def act = full.read(block3.hash).block() + + then: + act != null + act.hash == BlockHash.from(hash4) + act.number == 102 + act.transactions.size() == 0 + } + + def "Return nothing if no transactions"() { + setup: + def txes = new TxMemCache() + def blocks = new BlocksMemCache() + + txes.add(tx1) + blocks.add(block1) //missing tx2 in cache + + def full = new BlocksWithTxCache(blocks, txes) + + when: + def act = full.read(block1.hash).block() + + then: + act == null + } + + def "Return nothing if no block"() { + setup: + def txes = new TxMemCache() + def blocks = new BlocksMemCache() + + txes.add(tx1) + txes.add(tx2) + txes.add(tx3) + + def full = new BlocksWithTxCache(blocks, txes) + + when: + def act = full.read(block1.hash).block() + + then: + act == null + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy index ff300428..a8747cc6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy @@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.cache.BlockByHeight import io.emeraldpay.dshackle.cache.BlocksMemCache import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.HeightCache +import io.emeraldpay.dshackle.cache.TxMemCache import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead @@ -12,6 +13,7 @@ import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import reactor.test.StepVerifier import spock.lang.Specification @@ -100,4 +102,97 @@ class CachingEthereumApiSpec extends Specification { .verify(Duration.ofSeconds(3)) } + def "Uses base cache when requested, by hash"() { + setup: + def blocksCache = Mock(BlocksMemCache) + def txCache = Mock(TxMemCache) + def head = Mock(EthereumHead.class) + def api = new CachingEthereumApi( + TestingCommons.objectMapper(), + Caches.newBuilder().setBlockByHash(blocksCache).setTxByHash(txCache).build(), + head + ) + def block = new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) + + when: + def act = api.readBlockByHash(1, "eth_getBlockByHash", ["0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58", false]).block() + + then: + act != null + 1 * blocksCache.read(BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) >> Mono.just(block) + 0 * txCache.read(_) + } + + def "Uses full cache when requested, by hash"() { + setup: + def blocksCache = Mock(BlocksMemCache) + def txCache = Mock(TxMemCache) + def head = Mock(EthereumHead.class) + def api = new CachingEthereumApi( + TestingCommons.objectMapper(), + Caches.newBuilder().setBlockByHash(blocksCache).setTxByHash(txCache).build(), + head + ) + def block = new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) + block.transactions = [ + new TransactionRefJson(TransactionId.from("0x0500219f2b147f3013e9030d585e8e5d45401ebd2620a42c879c0d5d1b754073")) + ] + + when: + def act = api.readBlockByHash(1, "eth_getBlockByHash", ["0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58", true]).block() + + then: + act == null + 1 * blocksCache.read(BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) >> Mono.just(block) + 1 * txCache.read(TransactionId.from("0x0500219f2b147f3013e9030d585e8e5d45401ebd2620a42c879c0d5d1b754073")) >> Mono.empty() + } + + def "Uses base cache when requested, by height"() { + setup: + def blocksCache = Mock(BlocksMemCache) + def txCache = Mock(TxMemCache) + def heightCache = Mock(HeightCache) + def head = Mock(EthereumHead.class) + def api = new CachingEthereumApi( + TestingCommons.objectMapper(), + Caches.newBuilder().setBlockByHash(blocksCache).setTxByHash(txCache).setBlockByHeight(heightCache).build(), + head + ) + def block = new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) + + when: + def act = api.readBlockByNumber(1, "eth_getBlockByNumber", ["0x64", false]).block() + + then: + act != null + 1 * heightCache.read(100) >> Mono.just(BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) + 1 * blocksCache.read(BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) >> Mono.just(block) + 0 * txCache.read(_) + } + + def "Uses full cache when requested, by height"() { + setup: + def blocksCache = Mock(BlocksMemCache) + def txCache = Mock(TxMemCache) + def heightCache = Mock(HeightCache) + def head = Mock(EthereumHead.class) + def api = new CachingEthereumApi( + TestingCommons.objectMapper(), + Caches.newBuilder().setBlockByHash(blocksCache).setTxByHash(txCache).setBlockByHeight(heightCache).build(), + head + ) + def block = new BlockJson(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) + block.transactions = [ + new TransactionRefJson(TransactionId.from("0x0500219f2b147f3013e9030d585e8e5d45401ebd2620a42c879c0d5d1b754073")) + ] + + when: + def act = api.readBlockByNumber(1, "eth_getBlockByNumber", ["0x64", true]).block() + + then: + act == null + 1 * heightCache.read(100) >> Mono.just(BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) + 1 * blocksCache.read(BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")) >> Mono.just(block) + 1 * txCache.read(TransactionId.from("0x0500219f2b147f3013e9030d585e8e5d45401ebd2620a42c879c0d5d1b754073")) >> Mono.empty() + } }