From 35cb35fe0a554abc78c6dfe8a68da984e8061685 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sat, 14 Mar 2020 22:25:27 -0400 Subject: [PATCH] solution: Redis caching --- README.adoc | 9 +- docs/09-caching.adoc | 62 +++++++++++- gradle.properties | 2 +- .../kotlin/io/emeraldpay/dshackle/Config.kt | 12 ++- .../dshackle/cache/BlocksRedisCache.kt | 26 +++-- .../dshackle/cache/BlocksWithTxCache.kt | 6 +- .../io/emeraldpay/dshackle/cache/Caches.kt | 77 ++++++++++++--- .../dshackle/cache/CachesFactory.kt | 86 +++++++++++++++++ .../emeraldpay/dshackle/cache/TxRedisCache.kt | 13 ++- .../dshackle/config/EnvVariables.kt | 19 ++++ .../dshackle/config/UpstreamsConfigReader.kt | 13 +-- .../dshackle/reader/CompoundReader.kt | 16 ++-- .../dshackle/upstream/CachingEthereumApi.kt | 3 + .../dshackle/upstream/CurrentUpstreams.kt | 6 +- .../upstream/ethereum/DirectEthereumApi.kt | 7 +- .../cache/BlocksRedisCacheSpec.groovy | 52 ++++++++++ .../dshackle/cache/TxRedisCacheSpec.groovy | 29 ++++++ .../dshackle/config/EnvVariablesSpec.groovy | 29 ++++++ .../config/UpstreamsConfigReaderSpec.groovy | 21 ---- .../dshackle/reader/CompoundReaderSpec.groovy | 95 +++++++++++++++++++ .../dshackle/test/TestingCommons.groovy | 6 ++ .../upstream/CurrentUpstreamsSpec.groovy | 8 +- src/test/resources/upstreams-bitcoin.yaml | 15 +++ 23 files changed, 529 insertions(+), 83 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/EnvVariables.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/config/EnvVariablesSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/reader/CompoundReaderSpec.groovy create mode 100644 src/test/resources/upstreams-bitcoin.yaml diff --git a/README.adoc b/README.adoc index 8931a4fe..4e7be3ca 100644 --- a/README.adoc +++ b/README.adoc @@ -43,14 +43,13 @@ image::call-schema.png[alt="Call Schema",width=100%,align="center"] == Roadmap -- [ ] Redis caching +- [ ] JSON RPC emulation, in addition to gRPC protocol +- [ ] *Support Bitcoin RPC* +- [ ] Access to ERC-20 tokens on asset level +- [ ] Subscription to bitcoind notification over gRPC (instead of ZeroMQ) - [ ] Prometheus monitoring - [ ] BIP-32 Pubkey -- [ ] *Support Bitcoin RPC* -- [ ] Subscription to bitcoind notification over GRPC (instead of ZeroMQ) -- [ ] JSON RPC emulation, in addition to GRPC protocol - [ ] Lightweight sidecar node connector -- [ ] Access to ERC-20 tokens on asset level - [ ] External logging - [ ] Configurable upstream roles diff --git a/docs/09-caching.adoc b/docs/09-caching.adoc index caf800df..25ffce51 100644 --- a/docs/09-caching.adoc +++ b/docs/09-caching.adoc @@ -1,9 +1,67 @@ == Caching -=== In memory cache +Dshackle can be configured to cache blockchain data. It can be a _hot_ in-memory cache, and optional _cold_ Redis-based +cache. + +Dshackle has enough information to effectively cache data and evict outdated values. If some data has been removed from +the blockchain, for example when block was replaced with another block at the same height, then the old values are +immediately evicted from the caches. + +=== In-memory cache Dshackle keeps latest blocks in memory (by default 64 blocks) === Redis cache -TBD \ No newline at end of file +Dshackle can optionally cache blocks and transactions in Redis cache. The values are cached up to 1 hour, but +fresh blocks and transactions are cached for shorter period. + +It makes sense to reuse the same Redis cache between multiple instances of the Dshackle. + +.Basic config (dshackle.yaml) +[source, yaml] +---- +cache: + redis: + enabled: true +---- + +.Full config (dshackle.yaml) +[source, yaml] +---- +cache: + redis: + enabled: true + host: 127.0.0.1 + port: 6379 + db: 0 + password: passw0rd! +---- + + +.Options +|=== +| Name | Default Value | Description + +| enabled +| false +| Set to `true` if Dshackle should use Redis for caching + +| host +| 127.0.0.1 +| Redis host + +| port +| 6379 +| Redis port + +| db +| 0 +| Redis database + +| password +| -- +| Password if Redis requires authentication. The value can be read from Environment variable, to do that + specify it as `${REDIS_PASSWORD}`, where REDIS_PASSWORD is the name of the variable + +|=== \ No newline at end of file diff --git a/gradle.properties b/gradle.properties index a4d1d6f4..7ea35874 100644 --- a/gradle.properties +++ b/gradle.properties @@ -15,7 +15,7 @@ springVersion=5.1.4.RELEASE reactorVersion=3.2.9.RELEASE # Our Libs -etherjarVersion=0.9.0-SNAPSHOT +etherjarVersion=0.10.0-SNAPSHOT # Testing spockVersion=1.2-groovy-2.5 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt index fe90a9b8..8e42d4c0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt @@ -19,25 +19,22 @@ import com.fasterxml.jackson.core.Version import com.fasterxml.jackson.databind.DeserializationFeature import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.module.SimpleModule +import io.lettuce.core.AbstractRedisClient +import io.lettuce.core.cluster.RedisClusterClient import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Qualifier -import org.springframework.context.ApplicationContext import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration -import org.springframework.context.annotation.Import import org.springframework.core.env.Environment import org.springframework.scheduling.annotation.EnableAsync import org.springframework.scheduling.annotation.EnableScheduling -import org.springframework.scheduling.annotation.Scheduled import reactor.core.scheduler.Scheduler import reactor.core.scheduler.Schedulers import java.io.File -import java.lang.IllegalStateException import java.text.SimpleDateFormat import java.util.* import java.util.concurrent.Executors -import kotlin.system.exitProcess @Configuration @EnableScheduling @@ -82,4 +79,9 @@ open class Config( open fun fileResolver(): FileResolver { return FileResolver(configDir()) } + + @Bean + open fun redisClient(): AbstractRedisClient { + return RedisClusterClient.create("redis://password@localhost:6379/0"); + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt index 5640d093..359c7d00 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt @@ -24,8 +24,9 @@ class BlocksRedisCache( companion object { private val log = LoggerFactory.getLogger(BlocksRedisCache::class.java) - // max caching time is 24 hours - private const val MAX_CACHE_TIME_HOURS = 24L + private const val MAX_CACHE_TIME_MINUTES = 60L + // doesn't make sense to cached in redis short living objects + private const val MIN_CACHE_TIME_SECONDS = 10 } override fun read(key: BlockHash): Mono> { @@ -37,23 +38,36 @@ class BlocksRedisCache( } } + fun evict(id: BlockHash): Mono { + return Mono.just(id) + .flatMap { + redis.del(key(it)) + } + .then() + } + /** * Add to cache. * Note that it returns Mono which must be subscribed to actually save */ - open fun add(block: BlockJson): Mono { + fun add(block: BlockJson): Mono { if (block.timestamp == null || block.hash == null) { return Mono.empty() } return Mono.just(block) .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 //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) + val ttl = min(age, TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES)) + if (ttl > MIN_CACHE_TIME_SECONDS) { + redis.setex(key(block.hash), ttl, data) + } else { + Mono.empty() + } } .doOnError { log.warn("Failed to save to Redis: ${it.message}") @@ -68,7 +82,7 @@ class BlocksRedisCache( /** * Key in Redis */ - open fun key(hash: BlockHash): String { + fun key(hash: BlockHash): String { return "block:${chain.id}:${hash.toHex()}" } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt index 1f5f2197..6cf7a9ae 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksWithTxCache.kt @@ -2,8 +2,10 @@ 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 import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import org.springframework.beans.BeanUtils import reactor.core.publisher.Flux @@ -17,8 +19,8 @@ import reactor.core.publisher.Mono * If any of the expected block transactions is not available it returns empty */ class BlocksWithTxCache( - private val blocks: BlocksMemCache, - private val txes: TxMemCache + private val blocks: Reader>, + private val txes: Reader ): Reader> { companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt index f0d59e28..e63027a5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.cache +import io.emeraldpay.dshackle.reader.CompoundReader import io.emeraldpay.dshackle.reader.Reader import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId @@ -7,11 +8,16 @@ import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionJson import io.infinitape.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.publisher.TopicProcessor open class Caches( - private val blocksByHash: BlocksMemCache, + private val memBlocksByHash: BlocksMemCache, private val blocksByHeight: HeightCache, - private val txsByHash: TxMemCache + private val memTxsByHash: TxMemCache, + private val redisBlocksByHash: BlocksRedisCache?, + private val redisTxsByHash: TxRedisCache? ) { companion object { @@ -28,6 +34,22 @@ open class Caches( } } + private val blocksByHash: Reader> + private val txsByHash: Reader + + init { + blocksByHash = if (redisBlocksByHash == null) { + memBlocksByHash + } else { + CompoundReader(memBlocksByHash, redisBlocksByHash) + } + txsByHash = if (redisTxsByHash == null) { + memTxsByHash + } else { + CompoundReader(memTxsByHash, redisTxsByHash) + } + } + /** * Cache data that was just requested */ @@ -40,32 +62,51 @@ open class Caches( } fun cache(tag: Tag, tx: TransactionJson) { - txsByHash.add(tx) + //do not cache transactions that are not in a block yet + if (tx.blockHash == null) { + return + } + memTxsByHash.add(tx) + memBlocksByHash.get(tx.blockHash)?.let { block -> + redisTxsByHash?.add(tx, block) + } } fun cache(tag: Tag, block: BlockJson) { + val job = ArrayList>() if (tag == Tag.LATEST) { - blocksByHash.add(block) + //for LATEST data cache in memory, it will be short living so better to avoid Redis + memBlocksByHash.add(block) val replaced = blocksByHeight.add(block) //evict cached transactions if an existing block was updated replaced?.let { replacedBlockHash -> var evicted = false - blocksByHash.get(replacedBlockHash)?.let { block -> - txsByHash.evict(block) + redisBlocksByHash?.evict(replacedBlockHash) + memBlocksByHash.get(replacedBlockHash)?.let { block -> + memTxsByHash.evict(block) + redisTxsByHash?.evict(block) evicted = true } if (!evicted) { - txsByHash.evict(replacedBlockHash) + memTxsByHash.evict(replacedBlockHash) } } } else if (tag == Tag.REQUESTED) { - // if block with transactions was requests cache only transactions - block.transactions.forEach { tx -> - if (tx is TransactionJson) { - cache(Tag.REQUESTED, tx) + //shouldn't cache block json with transactions, separate txes and blocks with refs + val blockOnly = block.withoutTransactionDetails() + memBlocksByHash.add(blockOnly) + redisBlocksByHash?.add(blockOnly)?.let(job::add) + + // now cache only transactions + val transactions = block.transactions.filterIsInstance() + if (transactions.isNotEmpty()) { + transactions.forEach { cache(Tag.REQUESTED, it) } + if (redisTxsByHash != null) { + job.add(Flux.fromIterable(transactions).flatMap { redisTxsByHash.add(it, block) }.then()) } } } + Flux.fromIterable(job).flatMap { it }.subscribe() //TODO move out to a caller } fun getBlocksByHash(): Reader> { @@ -107,12 +148,19 @@ open class Caches( private var blocksByHash: BlocksMemCache? = null private var blocksByHeight: HeightCache? = null private var txsByHash: TxMemCache? = null + private var redisBlocksByHash: BlocksRedisCache? = null + private var redisTxsByHash: TxRedisCache? = null fun setBlockByHash(cache: BlocksMemCache): Builder { blocksByHash = cache return this } + fun setBlockByHash(cache: BlocksRedisCache): Builder { + redisBlocksByHash = cache + return this + } + fun setBlockByHeight(cache: HeightCache): Builder { blocksByHeight = cache return this @@ -123,6 +171,11 @@ open class Caches( return this } + fun setTxByHash(cache: TxRedisCache): Builder { + redisTxsByHash = cache + return this + } + fun build(): Caches { if (blocksByHash == null) { blocksByHash = BlocksMemCache() @@ -133,7 +186,7 @@ open class Caches( if (txsByHash == null) { txsByHash = TxMemCache() } - return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!) + return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash) } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt new file mode 100644 index 00000000..169665e7 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt @@ -0,0 +1,86 @@ +package io.emeraldpay.dshackle.cache + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.config.EnvVariables +import io.emeraldpay.grpc.Chain +import io.lettuce.core.RedisClient +import io.lettuce.core.RedisURI +import io.lettuce.core.api.StatefulRedisConnection +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.beans.factory.annotation.Value +import org.springframework.core.env.Environment +import org.springframework.stereotype.Repository +import java.util.* +import java.util.concurrent.ConcurrentHashMap +import javax.annotation.PostConstruct +import kotlin.collections.HashMap + + +@Repository +class CachesFactory( + @Autowired private val objectMapper: ObjectMapper, + @Autowired private val env: Environment +) { + + companion object { + private val log = LoggerFactory.getLogger(CachesFactory::class.java) + private const val CONFIG_PREFIX = "cache.redis" + } + + private var redis: StatefulRedisConnection? = null + private val all = EnumMap(io.emeraldpay.grpc.Chain::class.java) + + @PostConstruct + fun init() { + if (!env.getProperty("${CONFIG_PREFIX}.enabled", Boolean::class.java, false)) { + return + } + val address = env.getProperty("${CONFIG_PREFIX}.host", "127.0.0.1") + val port = env.getProperty("${CONFIG_PREFIX}.port", Int::class.java, 6379) + + var uri = RedisURI.builder() + .withHost(address) + .withPort(port) + + env.getProperty("${CONFIG_PREFIX}.db", Int::class.java)?.let { value -> + uri = uri.withDatabase(value) + } + + //log URI _before_ adding a password, to avoid leaking it to the log + log.info("Use Redis cache at: ${uri.build().toURI()}") + + env.getProperty("${CONFIG_PREFIX}.password")?.let { value -> + uri = uri.withPassword(value) + } + + val client = RedisClient.create(uri.build()) + val ping = client.connect().sync().ping() + if (ping != "PONG") { + throw IllegalStateException("Redis connection is not configured. Response: $ping") + } + redis = client.connect() + } + + private fun initCache(chain: Chain): Caches { + val caches = Caches.newBuilder() + redis?.let { redis -> + caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain, objectMapper)) + caches.setTxByHash(TxRedisCache(redis.reactive(), chain, objectMapper)) + } + return caches.build() + } + + fun getCaches(chain: Chain): Caches { + val existing = all[chain] + if (existing == null) { + synchronized(all) { + if (!all.containsKey(chain)) { + all[chain] = initCache(chain) + } + } + return getCaches(chain) + } + return existing + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt index ae73d92c..478db4df 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt @@ -39,7 +39,7 @@ class TxRedisCache( } } - open fun evict(block: BlockJson): Mono { + fun evict(block: BlockJson): Mono { return Mono.just(block) .map { block -> block.transactions.map { @@ -50,8 +50,15 @@ class TxRedisCache( }.then() } + fun evict(id: TransactionId): Mono { + return Mono.just(id) + .flatMap { + redis.del(key(it)) + } + .then() + } - open fun add(tx: TransactionJson, block: BlockJson): Mono { + fun add(tx: TransactionJson, block: BlockJson): Mono { if (tx.blockHash == null || block.hash == null || tx.blockHash != block.hash || block.timestamp == null) { return Mono.empty() } @@ -78,7 +85,7 @@ class TxRedisCache( /** * Key in Redis */ - open fun key(hash: TransactionId): String { + fun key(hash: TransactionId): String { return "tx:${chain.id}:${hash.toHex()}" } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/EnvVariables.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/EnvVariables.kt new file mode 100644 index 00000000..08c8b4a9 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/EnvVariables.kt @@ -0,0 +1,19 @@ +package io.emeraldpay.dshackle.config + +/** + * Update configuration value from environment variables. Format: ${ENV_VAR_NAME} + */ +class EnvVariables { + + companion object { + private val envRegex = Regex("\\$\\{(\\w+?)}") + } + + fun postProcess(value: String): String { + return envRegex.replace(value) { m -> + m.groups[1]?.let { g -> + System.getProperty(g.value) ?: System.getenv(g.value) ?: "" + } ?: "" + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 0339401b..48a47a05 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -32,7 +32,7 @@ import java.time.Duration class UpstreamsConfigReader { private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java) - private val envRegex = Regex("\\$\\{(\\w+?)}") + private val envVariables = EnvVariables() fun read(input: InputStream): UpstreamsConfig { val yaml = Yaml() @@ -268,13 +268,13 @@ class UpstreamsConfigReader { private fun getListOfString(mappingNode: MappingNode?, key: String): List? { return getList(mappingNode, key)?.value ?.map { it.value } - ?.map(this::postProcess) + ?.map(envVariables::postProcess) } private fun getValueAsString(mappingNode: MappingNode?, key: String): String? { return getValue(mappingNode, key)?.let { return@let it.value - }?.let(this::postProcess) + }?.let(envVariables::postProcess) } private fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? { @@ -305,11 +305,4 @@ class UpstreamsConfigReader { } } - fun postProcess(value: String): String { - return envRegex.replace(value) { m -> - m.groups[1]?.let { g -> - System.getProperty(g.value) ?: System.getenv(g.value) ?: "" - } ?: "" - } - } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt index 65d10285..001a6c66 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt @@ -15,24 +15,22 @@ */ package io.emeraldpay.dshackle.reader +import reactor.core.publisher.Flux import reactor.core.publisher.Mono +/** + * Composition of multiple readers. Reader returns first value returned by any of the source readers. + */ class CompoundReader( - private val readers: Collection> + private vararg val readers: Reader ): Reader { override fun read(key: K): Mono { if (readers.isEmpty()) { return Mono.empty() } - var result = readers.first().read(key) - if (readers.size == 1) { - return result - } - readers.stream().skip(1).forEach { - result = result.switchIfEmpty(it.read(key)) - } - return result + return Flux.fromIterable(readers.asIterable()) + .flatMap { it.read(key) }.next() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt index addb97a2..4893069e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt @@ -38,6 +38,9 @@ open class CachingEthereumApi( companion object { private val log = LoggerFactory.getLogger(CachingEthereumApi::class.java) + /** + * Create caching API with empty memory-only cache + */ @JvmStatic fun empty(): CachingEthereumApi { return CachingEthereumApi(ObjectMapper(), Caches.default(), EmptyEthereumHead()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt index b9232f58..72eaac65 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesEnabled +import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.QuorumBasedMethods @@ -35,7 +36,8 @@ import kotlin.concurrent.withLock @Repository class CurrentUpstreams( - @Autowired private val objectMapper: ObjectMapper + @Autowired private val objectMapper: ObjectMapper, + @Autowired private val cachesFactory: CachesFactory ): Upstreams { private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java) @@ -55,7 +57,7 @@ class CurrentUpstreams( log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed") } else { if (current == null) { - val created = ChainUpstreams(chain, ArrayList(), Caches.default(), objectMapper) + val created = ChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) if (up is CachesEnabled) { up.setCaches(created.caches) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt index 11b2d9eb..d472b9f4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt @@ -101,7 +101,12 @@ open class DirectEthereumApi( return rpcClient.execute(callMapping(method, params)) .timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout"))) .doOnNext { value -> - caches?.cacheRequested(value) + try { + caches?.cacheRequested(value) + } catch (e: Throwable) { + //ignore all caching errors, client shouldn't have problems because of them + log.warn("Uncaught caching exception", e) + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy index 3014ef0b..931678fe 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy @@ -52,4 +52,56 @@ class BlocksRedisCacheSpec extends Specification { act == block } + def "Evict existing block"() { + 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(hash2) + block.transactions = [] + block.uncles = [] + + when: + cache.add(block).subscribe() + def act = cache.read(BlockHash.from(hash2)).block() + then: + act == block + + when: + cache.evict(block.hash).subscribe() + act = cache.read(BlockHash.from(hash2)).block() + + then: + act == null + } + + def "Evict non-existing block"() { + 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(hash2) + block.transactions = [] + block.uncles = [] + + when: + cache.add(block).subscribe() + def act = cache.read(BlockHash.from(hash2)).block() + then: + act == block + + when: + cache.evict(BlockHash.from(hash3)).subscribe() + act = cache.read(BlockHash.from(hash2)).block() + + then: + act == block + } + } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy index e9d982cb..c97b24a6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy @@ -54,6 +54,35 @@ class TxRedisCacheSpec extends Specification { act == tx } + def "Evict single tx"() { + setup: + def block = new BlockJson() + block.number = 100 + block.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS) + block.hash = BlockHash.from(hash2) + block.transactions = [] + block.uncles = [] + + def tx = new TransactionJson() + tx.hash = TransactionId.from(hash3) + 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(tx.hash).block() + then: + act == tx + + when: + cache.evict(tx.hash).subscribe() + act = cache.read(tx.hash).block() + then: + act == null + } + def "Evict all by block data"() { when: def block1 = new BlockJson() diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/EnvVariablesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/EnvVariablesSpec.groovy new file mode 100644 index 00000000..6834a032 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/config/EnvVariablesSpec.groovy @@ -0,0 +1,29 @@ +package io.emeraldpay.dshackle.config + +import spock.lang.Specification + +class EnvVariablesSpec extends Specification { + + EnvVariables reader = new EnvVariables() + + def "Post process for usual strings"() { + expect: + s == reader.postProcess(s) + where: + s << ["", "a", "13143", "/etc/client1.myservice.com.key", "true", "1a68f20154fc258fe4149c199ad8f281"] + } + + def "Post process replaces from env"() { + setup: + System.setProperty("id", "1") + System.setProperty("HOME", "/home/user") + System.setProperty("PASSWORD", "1a68f20154fc258fe4149c199ad8f281") + expect: + replaced == reader.postProcess(orig) + where: + orig | replaced + "p_\${id}" | "p_1" + "home: \${HOME}" | "home: /home/user" + "\${PASSWORD}" | "1a68f20154fc258fe4149c199ad8f281" + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 045af98a..0c904d25 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -134,27 +134,6 @@ class UpstreamsConfigReaderSpec extends Specification { } } - def "Post process for usual strings"() { - expect: - s == reader.postProcess(s) - where: - s << ["", "a", "13143", "/etc/client1.myservice.com.key", "true", "1a68f20154fc258fe4149c199ad8f281"] - } - - def "Post process replaces from env"() { - setup: - System.setProperty("id", "1") - System.setProperty("HOME", "/home/user") - System.setProperty("PASSWORD", "1a68f20154fc258fe4149c199ad8f281") - expect: - replaced == reader.postProcess(orig) - where: - orig | replaced - "p_\${id}" | "p_1" - "home: \${HOME}" | "home: /home/user" - "\${PASSWORD}" | "1a68f20154fc258fe4149c199ad8f281" - } - def "Parse config without defaults"() { setup: def config = this.class.getClassLoader().getResourceAsStream("upstreams-no-defaults.yaml") diff --git a/src/test/groovy/io/emeraldpay/dshackle/reader/CompoundReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/reader/CompoundReaderSpec.groovy new file mode 100644 index 00000000..0b559028 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/reader/CompoundReaderSpec.groovy @@ -0,0 +1,95 @@ +package io.emeraldpay.dshackle.reader + +import reactor.core.publisher.Mono +import reactor.test.StepVerifier +import spock.lang.Specification + +import java.time.Duration + +class CompoundReaderSpec extends Specification { + + def reader1 = new Reader() { + @Override + Mono read(String key) { + return Mono.just("test-1").delaySubscription(Duration.ofMillis(100)) + } + } + def reader2 = new Reader() { + @Override + Mono read(String key) { + return Mono.just("test-2").delaySubscription(Duration.ofMillis(200)) + } + } + def reader3 = new Reader() { + @Override + Mono read(String key) { + return Mono.just("test-3").delaySubscription(Duration.ofMillis(300)) + } + } + + def reader1Empty = new Reader() { + @Override + Mono read(String key) { + return Mono.empty().delaySubscription(Duration.ofMillis(100)) + } + } + + def "Return empty when no readers"() { + setup: + def reader = new CompoundReader() + when: + def act = reader.read("test") + then: + StepVerifier.create(act) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Return first"() { + setup: + def reader = new CompoundReader(reader1, reader2, reader3) + when: + def act = reader.read("test") + then: + StepVerifier.create(act) + .expectNext("test-1") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Return second"() { + setup: + def reader = new CompoundReader(reader3, reader2) + when: + def act = reader.read("test") + then: + StepVerifier.create(act) + .expectNext("test-2") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Return third"() { + setup: + def reader = new CompoundReader(reader3, reader2, reader1) + when: + def act = reader.read("test") + then: + StepVerifier.create(act) + .expectNext("test-1") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Ignore empty"() { + setup: + def reader = new CompoundReader(reader3, reader1Empty, reader2, reader1Empty) + when: + def act = reader.read("test") + then: + StepVerifier.create(act) + .expectNext("test-2") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 918d01f6..d7e7158f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.DeserializationFeature import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.module.SimpleModule import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.upstream.AggregatedUpstream import io.emeraldpay.dshackle.upstream.ChainUpstreams import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods @@ -28,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.ReactorRpcClient +import org.springframework.core.env.StandardEnvironment import java.text.SimpleDateFormat @@ -73,4 +75,8 @@ class TestingCommons { static AggregatedUpstream aggregatedUpstream(EthereumUpstream up) { return new ChainUpstreams(Chain.ETHEREUM, [up], Caches.default(), objectMapper()) } + + static CachesFactory emptyCaches() { + return new CachesFactory(objectMapper(), new StandardEnvironment()) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy index d5e09d4b..0c9ead4b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/CurrentUpstreamsSpec.groovy @@ -11,7 +11,7 @@ class CurrentUpstreamsSpec extends Specification { def "add upstream"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper()) + def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) when: current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED)) @@ -22,7 +22,7 @@ class CurrentUpstreamsSpec extends Specification { def "add multiple upstreams"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper()) + def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(ReactorRpcClient))) def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) @@ -38,7 +38,7 @@ class CurrentUpstreamsSpec extends Specification { def "remove upstream"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper()) + def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api(Stub(ReactorRpcClient))) def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) @@ -56,7 +56,7 @@ class CurrentUpstreamsSpec extends Specification { def "available after adding"() { setup: - def current = new CurrentUpstreams(TestingCommons.objectMapper()) + def current = new CurrentUpstreams(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(Stub(ReactorRpcClient))) when: diff --git a/src/test/resources/upstreams-bitcoin.yaml b/src/test/resources/upstreams-bitcoin.yaml new file mode 100644 index 00000000..13587264 --- /dev/null +++ b/src/test/resources/upstreams-bitcoin.yaml @@ -0,0 +1,15 @@ +version: v1 + +defaultOptions: + - chains: + - bitcoin + options: + min-peers: 3 + +upstreams: + - id: local + chain: bitcoin + connection: + bitcoin: + rpc: + url: "http://localhost:8545" \ No newline at end of file