From 83649ad63187b907c3d04edd399f1c4c4a6111d4 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 10 Mar 2020 23:45:00 -0400 Subject: [PATCH] solution: Redis cache for blocks --- build.gradle | 3 +- .../dshackle/cache/BlocksRedisCache.kt | 74 +++++++++++++++++++ .../cache/BlocksRedisCacheSpec.groovy | 55 ++++++++++++++ 3 files changed, 131 insertions(+), 1 deletion(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy diff --git a/build.gradle b/build.gradle index d6a33639..4a5120b1 100644 --- a/build.gradle +++ b/build.gradle @@ -67,8 +67,9 @@ dependencies { compile "org.springframework.security:spring-security-config:$springVersion" compile "io.projectreactor:reactor-core:$reactorVersion" compile 'io.projectreactor.addons:reactor-extra:3.2.3.RELEASE' - compile 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.0.M1' + compile 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.2.RELEASE' compile 'com.salesforce.servicelibs:reactor-grpc:0.10.0' + compile 'io.lettuce:lettuce-core:5.2.2.RELEASE' compile 'org.yaml:snakeyaml:1.24' diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt new file mode 100644 index 00000000..3a20e280 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt @@ -0,0 +1,74 @@ +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.BlockHash +import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import io.lettuce.core.api.reactive.RedisReactiveCommands +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.time.Instant +import java.util.concurrent.TimeUnit +import kotlin.math.min + +/** + * Cache blocks in Redis database + */ +class BlocksRedisCache( + private val redis: RedisReactiveCommands, + private val chain: Chain, + private val objectMapper: ObjectMapper +): Reader> { + + companion object { + private val log = LoggerFactory.getLogger(BlocksRedisCache::class.java) + // max caching time is 24 hours + private const val MAX_CACHE_TIME_HOURS = 24L + } + + override fun read(key: BlockHash): Mono> { + return redis.get(key(key)) + .map { data -> + objectMapper.readValue(data, BlockJson::class.java) as BlockJson + }.onErrorResume { + Mono.empty() + } + } + + /** + * Add to cache. + * Note that it returns Mono which must be subscribed to actually save + */ + open fun add(block: BlockJson): Mono { + if (block.timestamp == null || block.hash == null) { + return Mono.empty() + } + return Mono.just(block) + .flatMap { + 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 + //still can be replaced in the blockchain + val age = Instant.now().epochSecond - block.timestamp.epochSecond + val ttl = min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS)) + redis.setex(key(block.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: BlockHash): String { + return "block:${chain.id}:${hash.toHex()}" + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy new file mode 100644 index 00000000..3014ef0b --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy @@ -0,0 +1,55 @@ +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.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson +import io.lettuce.core.RedisClient +import io.lettuce.core.api.StatefulRedisConnection +import spock.lang.IgnoreIf +import spock.lang.Specification + +import java.time.Instant +import java.time.temporal.ChronoUnit + +@IgnoreIf({ + env["DSHACKLE_TEST_ENABLED"] == null || !env["DSHACKLE_TEST_ENABLED"].contains("redis") +}) +class BlocksRedisCacheSpec extends Specification { + + StatefulRedisConnection redis + + String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" + String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" + String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" + String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" + + + def setup() { + RedisClient client = RedisClient.create("redis://localhost:6379"); + StatefulRedisConnection connection = client.connect(); + connection.sync().flushdb() + redis = connection + } + + def "Add and read"() { + setup: + def cache = new BlocksRedisCache( + redis.reactive(), Chain.ETHEREUM, TestingCommons.objectMapper() + ) + 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 = [] + + when: + cache.add(block).subscribe() + def act = cache.read(BlockHash.from(hash1)).block() + then: + act == block + } + +}