From d3a3b6606a994319290c2daacfbbc95407e450bf Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 20 Jul 2020 20:34:37 -0400 Subject: [PATCH] problem: doesn't cache transactions requested for a block --- .../io/emeraldpay/dshackle/cache/Caches.kt | 15 ++-- .../dshackle/cache/OnTxRedisCache.kt | 2 +- .../emeraldpay/dshackle/cache/TxRedisCache.kt | 4 +- .../dshackle/cache/CachesSpec.groovy | 75 +++++++++++++++---- 4 files changed, 71 insertions(+), 25 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt index 4f0bdd39..4d71d470 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -97,9 +97,10 @@ open class Caches( return } memTxsByHash.add(tx) - memBlocksByHash.get(tx.blockId)?.let { block -> - redisTxsByHash?.add(tx, block) - } + //TODO move subscription to the caller + getBlocksByHash().read(tx.blockId).flatMap { block -> + redisTxsByHash?.add(tx, block) ?: Mono.empty() + }.subscribe() } fun cache(tag: Tag, block: BlockContainer) { @@ -142,11 +143,11 @@ open class Caches( val transactions = plainTransactions.map { tx -> TxContainer.from(tx) } - transactions.forEach { - cache(Tag.REQUESTED, it) - } if (redisTxsByHash != null) { - job.add(Flux.fromIterable(transactions).flatMap { redisTxsByHash.add(it, block) }.then()) + job.add(Flux.fromIterable(transactions) + .doOnNext { memTxsByHash.add(it) } + .flatMap { redisTxsByHash.add(it, block) } + .then()) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt index 1a40c656..74ff6a56 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt @@ -117,7 +117,7 @@ abstract class OnTxRedisCache( } } - fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono { + open fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono { return Mono.just(id) .flatMap { val key = key(it) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt index e5eaa14d..1591d4ac 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt @@ -35,7 +35,7 @@ import kotlin.math.min /** * Cache transactions in Redis, up to 24 hours. */ -class TxRedisCache( +open class TxRedisCache( private val redis: RedisReactiveCommands, private val chain: Chain ) : Reader, @@ -74,7 +74,7 @@ class TxRedisCache( ) } - fun add(tx: TxContainer, block: BlockContainer): Mono { + open fun add(tx: TxContainer, block: BlockContainer): Mono { return super.add(tx.hash, tx, block, tx.height) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy index 510d0835..39896d55 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.cache import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global 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.infinitape.etherjar.domain.BlockHash @@ -25,6 +26,7 @@ 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 reactor.core.publisher.Mono import spock.lang.Specification import java.time.Instant @@ -145,14 +147,6 @@ class CachesSpec extends Specification { def "Cache txes of a requested block"() { setup: - TxMemCache txCache = Mock() - HeightCache heightCache = Mock() - BlocksMemCache blocksCache = Mock() - def caches = Caches.newBuilder() - .setTxByHash(txCache) - .setBlockByHeight(heightCache) - .setBlockByHash(blocksCache) - .build() def tx1 = new TransactionJson().with { hash = TransactionId.from(hash1) @@ -166,20 +160,71 @@ class CachesSpec extends Specification { blockNumber = 100 it } + BlockContainer block = new BlockJson().with { block -> + block.number = 100 + block.hash = BlockHash.from(hash1) + block.totalDifficulty = BigInteger.ONE + block.transactions = [tx1, tx2] + block.timestamp = Instant.now() + BlockContainer.from(block) + } - def block = new BlockJson() - block.number = 100 - block.hash = BlockHash.from(hash1) - block.totalDifficulty = BigInteger.ONE - block.transactions = [tx1, tx2] - block.timestamp = Instant.now() - block = BlockContainer.from(block) + TxMemCache txCache = Mock() + HeightCache heightCache = Mock() + BlocksMemCache blocksCache = Mock() { + _ * add(block) + _ * read(block.hash) >> Mono.just(block) + } + TxRedisCache txRedisCache = Mock() + def caches = Caches.newBuilder() + .setTxByHash(txCache) + .setBlockByHeight(heightCache) + .setBlockByHash(blocksCache) + .setTxByHash(txRedisCache) + .build() when: caches.cache(Caches.Tag.REQUESTED, block) then: 1 * txCache.add(TxContainer.from(tx1)) 1 * txCache.add(TxContainer.from(tx2)) + 1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then() + 1 * txRedisCache.add(TxContainer.from(tx2), block) >> Mono.just(1).then() + } + + def "Cache tx with redis"() { + setup: + + def tx1 = new TransactionJson().with { + hash = TransactionId.from(hash1) + blockHash = BlockHash.from(hash1) + blockNumber = 100 + it + } + BlockContainer block = new BlockJson().with { block -> + block.number = 100 + block.hash = BlockHash.from(hash1) + block.totalDifficulty = BigInteger.ONE + block.transactions = [tx1] + block.timestamp = Instant.now() + BlockContainer.from(block) + } + TxMemCache txCache = Mock() + HeightCache heightCache = Mock() + BlocksMemCache blocksCache = Mock() + TxRedisCache txRedisCache = Mock() + def caches = Caches.newBuilder() + .setTxByHash(txCache) + .setBlockByHeight(heightCache) + .setBlockByHash(blocksCache) + .setTxByHash(txRedisCache) + .build() + + when: + caches.cache(Caches.Tag.REQUESTED, TxContainer.from(tx1)) + then: + 1 * blocksCache.read(block.hash) >> Mono.just(block) + 1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then() } }