From 46cf08c6a23b93fa0f60b46b09049ce7f37b1e39 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 11 Mar 2020 23:02:30 -0400 Subject: [PATCH] solution: Redis cache for transactions --- .../dshackle/cache/BlocksRedisCache.kt | 2 +- .../emeraldpay/dshackle/cache/TxRedisCache.kt | 84 +++++++++++++ .../dshackle/cache/TxRedisCacheSpec.groovy | 111 ++++++++++++++++++ 3 files changed, 196 insertions(+), 1 deletion(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt index 3a20e280..5640d093 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt @@ -46,7 +46,7 @@ class BlocksRedisCache( return Mono.empty() } return Mono.just(block) - .flatMap { + .flatMap { block -> val data = objectMapper.writeValueAsString(block) //default caching time is age of the block, i.e. block create hour ago //keep for hour, but block create 10 seconds ago cache for 10 seconds, as it diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt new file mode 100644 index 00000000..ae73d92c --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt @@ -0,0 +1,84 @@ +package io.emeraldpay.dshackle.cache + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.grpc.Chain +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 io.lettuce.core.api.reactive.RedisReactiveCommands +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import reactor.util.function.Tuples +import java.time.Instant +import java.util.concurrent.TimeUnit +import kotlin.math.min + +/** + * Cache transactions in Redis, up to 24 hours. + */ +class TxRedisCache( + private val redis: RedisReactiveCommands, + private val chain: Chain, + private val objectMapper: ObjectMapper +): Reader { + + companion object { + private val log = LoggerFactory.getLogger(TxRedisCache::class.java) + // max caching time is 24 hours + private const val MAX_CACHE_TIME_HOURS = 24L + } + + override fun read(key: TransactionId): Mono { + return redis.get(key(key)) + .map { data -> + objectMapper.readValue(data, TransactionJson::class.java) as TransactionJson + }.onErrorResume { + Mono.empty() + } + } + + open fun evict(block: BlockJson): Mono { + return Mono.just(block) + .map { block -> + block.transactions.map { + key(it.hash) + }.toTypedArray() + }.flatMap { keys -> + redis.del(*keys) + }.then() + } + + + open fun add(tx: TransactionJson, block: BlockJson): Mono { + if (tx.blockHash == null || block.hash == null || tx.blockHash != block.hash || block.timestamp == null) { + return Mono.empty() + } + return Mono.just(Tuples.of(tx, block)) + .flatMap { + val data = objectMapper.writeValueAsString(it.t1) + //default caching time is age of the block, i.e. block create hour ago + //keep for hour, but block create 10 seconds ago cache for 10 seconds, as it + //still can be replaced in the blockchain + val age = Instant.now().epochSecond - it.t2.timestamp.epochSecond + val ttl = min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS)) + redis.setex(key(it.t1.hash), ttl, data) + } + .doOnError { + log.warn("Failed to save to Redis: ${it.message}") + } + //if failed to cache, just continue without it + .onErrorResume { + Mono.empty() + } + .then() + } + + /** + * Key in Redis + */ + open fun key(hash: TransactionId): String { + return "tx:${chain.id}:${hash.toHex()}" + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy new file mode 100644 index 00000000..e9d982cb --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy @@ -0,0 +1,111 @@ +package io.emeraldpay.dshackle.cache + +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.domain.Wei +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import io.lettuce.core.RedisClient +import io.lettuce.core.api.StatefulRedisConnection +import spock.lang.Specification + +import java.time.Instant +import java.time.temporal.ChronoUnit + +class TxRedisCacheSpec extends Specification { + + String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" + String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" + TxRedisCache cache + + def setup() { + RedisClient client = RedisClient.create("redis://localhost:6379"); + StatefulRedisConnection connection = client.connect(); + connection.sync().flushdb() + StatefulRedisConnection redis = connection + cache = new TxRedisCache(redis.reactive(), Chain.ETHEREUM, TestingCommons.objectMapper()) + } + + def "Add and read"() { + setup: + def block = new BlockJson() + block.number = 100 + block.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS) + block.hash = BlockHash.from(hash1) + block.transactions = [] + block.uncles = [] + + def tx = new TransactionJson() + tx.hash = TransactionId.from(hash1) + tx.blockHash = block.hash + tx.blockNumber = block.number + tx.value = Wei.ofEthers(1.234) + tx.nonce = 0 + + when: + cache.add(tx, block).subscribe() + def act = cache.read(TransactionId.from(hash1)).block() + then: + act == tx + } + + def "Evict all by block data"() { + when: + def block1 = new BlockJson() + block1.hash = BlockHash.from(hash1) + block1.number = 100 + block1.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS) + block1.transactions = [ + new TransactionRefJson(TransactionId.from(hash1)), + new TransactionRefJson(TransactionId.from(hash2)), + ] + def block2 = new BlockJson() + block2.hash = BlockHash.from(hash2) + block2.number = 101 + block2.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS) + block2.transactions = [ + new TransactionRefJson(TransactionId.from(hash3)), + new TransactionRefJson(TransactionId.from(hash4)), + ] + + [hash1, hash2].eachWithIndex{ String hash, int i -> + def tx = new TransactionJson() + tx.blockNumber = block1.number + tx.blockHash = block1.hash + tx.hash = TransactionId.from(hash) + tx.value = Wei.ofEthers(i) + tx.nonce = 0 + cache.add(tx, block1).subscribe() + } + [hash3, hash4].eachWithIndex{ String hash, int i -> + def tx = new TransactionJson() + tx.blockNumber = block2.number + tx.blockHash = block2.hash + tx.hash = TransactionId.from(hash) + tx.value = Wei.ofEthers(i) + tx.nonce = 0 + cache.add(tx, block2).subscribe() + } + + + cache.evict(block1).subscribe() + + def act1 = cache.read(TransactionId.from(hash1)).block() + def act2 = cache.read(TransactionId.from(hash2)).block() + def act3 = cache.read(TransactionId.from(hash3)).block() + def act4 = cache.read(TransactionId.from(hash4)).block() + + then: + act1 == null + act2 == null + act3 != null + act3.hash.toHex() == hash3 + act4 != null + act4.hash.toHex() == hash4 + } +}