diff --git a/build.gradle b/build.gradle index 2910c498..19582c0d 100644 --- a/build.gradle +++ b/build.gradle @@ -10,6 +10,7 @@ buildscript { } dependencies { classpath 'org.jetbrains.kotlin:kotlin-gradle-plugin:1.3.70' + classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.12' } } @@ -25,6 +26,7 @@ plugins { id 'org.springframework.boot' version '2.1.4.RELEASE' id 'io.spring.dependency-management' version '1.0.6.RELEASE' id 'com.palantir.git-version' version '0.12.2' + id "com.google.protobuf" version "0.8.12" } @@ -165,6 +167,15 @@ jar { afterEvaluate { distZip.dependsOn(jar) compileKotlin.dependsOn(generateVersion) + generateProto.dependsOn(clean) +} + +protobuf { + protoc { artifact = "com.google.protobuf:protoc:${protocVersion}" } + plugins { + } + generateProtoTasks { + } } task generateVersion() { diff --git a/gradle.properties b/gradle.properties index 3c39a100..c5ae5013 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,12 +1,12 @@ # Languages groovyVersion=2.5.5 kotlinVersion=1.3.11 +protocVersion=3.7.1 # Main Libs slf4jVersion=1.7.25 jacksonVersion=2.9.8 grpcVersion=1.20.0 -protocVersion=3.7.1 protobufVersion=3.7.1 # Core diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt index dc542440..e74f2d11 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt @@ -1,15 +1,16 @@ package io.emeraldpay.dshackle.cache -import com.fasterxml.jackson.databind.ObjectMapper +import com.google.protobuf.ByteString import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.proto.CachesProto import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.json.BlockJson import io.lettuce.core.api.reactive.RedisReactiveCommands -import org.apache.commons.codec.binary.Base64 import org.slf4j.LoggerFactory import reactor.core.publisher.Mono +import java.math.BigInteger import java.time.Instant import java.util.concurrent.TimeUnit import kotlin.math.min @@ -18,9 +19,8 @@ import kotlin.math.min * Cache blocks in Redis database */ class BlocksRedisCache( - private val redis: RedisReactiveCommands, - private val chain: Chain, - private val objectMapper: ObjectMapper + private val redis: RedisReactiveCommands, + private val chain: Chain ) : Reader { companion object { @@ -34,13 +34,56 @@ class BlocksRedisCache( override fun read(key: BlockId): Mono { return redis.get(key(key)) .map { data -> - val block = objectMapper.readValue(data, BlockJson::class.java) - BlockContainer.from(block, objectMapper) + fromProto(data) }.onErrorResume { Mono.empty() } } + fun toProto(value: BlockContainer): ByteArray { + if (value.full) { + throw IllegalArgumentException("Full Block is not supposed to be cached") + } + val meta = CachesProto.BlockMeta.newBuilder() + .setHash(ByteString.copyFrom(value.hash.value)) + .setHeight(value.height) + .setDifficulty(ByteString.copyFrom(value.difficulty.toByteArray())) + .setTimestamp(value.timestamp.toEpochMilli()) + + value.transactions.forEach { + meta.addTxHashes(ByteString.copyFrom(it.value)) + } + + return CachesProto.ValueContainer.newBuilder() + .setType(CachesProto.ValueContainer.ValueType.BLOCK) + .setValue(ByteString.copyFrom(value.json!!)) + .setBlockMeta(meta) + .build() + .toByteArray() + } + + fun fromProto(msg: ByteArray): BlockContainer { + val value = CachesProto.ValueContainer.parseFrom(msg) + if (value.type != CachesProto.ValueContainer.ValueType.BLOCK) { + throw IllegalArgumentException("Expect BLOCK value, receive ${value.type}") + } + if (!value.hasBlockMeta()) { + throw IllegalArgumentException("Container doesn't have Block Meta") + } + val meta = value.blockMeta + return BlockContainer( + meta.height, + BlockId(meta.hash.toByteArray()), + BigInteger(meta.difficulty.toByteArray()), + Instant.ofEpochMilli(meta.timestamp), + false, + value.value.toByteArray(), + meta.txHashesList.map { + TxId(it.toByteArray()) + } + ) + } + fun evict(id: BlockId): Mono { return Mono.just(id) .flatMap { @@ -59,20 +102,21 @@ class BlocksRedisCache( } return Mono.just(block) .flatMap { block -> - val data = String(block.json!!) //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.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES)) if (ttl > MIN_CACHE_TIME_SECONDS) { - redis.setex(key(block.hash), ttl, data) + val key = key(block.hash) + val value = toProto(block) + redis.setex(key, ttl, value) } else { Mono.empty() } } .doOnError { - log.warn("Failed to save to Redis: ${it.message}") + log.warn("Failed to save Block to Redis: ${it.message}") } //if failed to cache, just continue without it .onErrorResume { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt index ee2eb8b6..b9cd9426 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt @@ -2,20 +2,18 @@ package io.emeraldpay.dshackle.cache import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.config.CacheConfig -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 io.lettuce.core.codec.ByteArrayCodec +import io.lettuce.core.codec.RedisCodec +import io.lettuce.core.codec.StringCodec 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 @@ -29,7 +27,7 @@ class CachesFactory( private const val CONFIG_PREFIX = "cache.redis" } - private var redis: StatefulRedisConnection? = null + private var redis: StatefulRedisConnection? = null private val all = EnumMap(io.emeraldpay.grpc.Chain::class.java) @PostConstruct @@ -56,15 +54,15 @@ class CachesFactory( if (ping != "PONG") { throw IllegalStateException("Redis connection is not configured. Response: $ping") } - redis = client.connect() + redis = client.connect(RedisCodec.of(StringCodec.ASCII, ByteArrayCodec.INSTANCE)) } private fun initCache(chain: Chain): Caches { val caches = Caches.newBuilder() .setObjectMapper(objectMapper) redis?.let { redis -> - caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain, objectMapper)) - caches.setTxByHash(TxRedisCache(redis.reactive(), chain, objectMapper)) + caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain)) + caches.setTxByHash(TxRedisCache(redis.reactive(), chain)) } return caches.build() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt index aa5e711d..ee4163a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt @@ -1,14 +1,15 @@ package io.emeraldpay.dshackle.cache import com.fasterxml.jackson.databind.ObjectMapper +import com.google.protobuf.ByteString import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.grpc.Chain -import io.infinitape.etherjar.rpc.json.TransactionJson +import io.emeraldpay.dshackle.proto.CachesProto import io.lettuce.core.api.reactive.RedisReactiveCommands -import org.apache.commons.codec.binary.Base64 import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import reactor.util.function.Tuples @@ -20,9 +21,8 @@ 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 + private val redis: RedisReactiveCommands, + private val chain: Chain ) : Reader { companion object { @@ -35,14 +35,46 @@ class TxRedisCache( override fun read(key: TxId): Mono { return redis.get(key(key)) .map { data -> - val json = data - val tx = objectMapper.readValue(json, TransactionJson::class.java) - TxContainer.from(tx, objectMapper) + fromProto(data) }.onErrorResume { Mono.empty() } } + fun toProto(value: TxContainer): ByteArray { + val meta = CachesProto.TxMeta.newBuilder() + .setHash(ByteString.copyFrom(value.hash.value)) + .setHeight(value.height) + + value.blockId?.value?.let { + meta.setBlockHash(ByteString.copyFrom(it)) + } + + return CachesProto.ValueContainer.newBuilder() + .setType(CachesProto.ValueContainer.ValueType.TX) + .setValue(ByteString.copyFrom(value.json!!)) + .setTxMeta(meta) + .build() + .toByteArray() + } + + fun fromProto(msg: ByteArray): TxContainer { + val value = CachesProto.ValueContainer.parseFrom(msg) + if (value.type != CachesProto.ValueContainer.ValueType.TX) { + throw IllegalArgumentException("Expect TX value, receive ${value.type}") + } + if (!value.hasTxMeta()) { + throw IllegalArgumentException("Container doesn't have Tx Meta") + } + val meta = value.txMeta + return TxContainer( + meta.height, + TxId(meta.hash.toByteArray()), + BlockId(meta.blockHash.toByteArray()), + value.value.toByteArray() + ) + } + fun evict(block: BlockContainer): Mono { return Mono.just(block) .map { block -> @@ -68,16 +100,18 @@ class TxRedisCache( } return Mono.just(Tuples.of(tx, block)) .flatMap { - val data = String(it.t1.json!!) + val key = key(it.t1.hash) + val value = toProto(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) + //store + redis.setex(key, ttl, value) } .doOnError { - log.warn("Failed to save to Redis: ${it.message}") + log.warn("Failed to save TX to Redis: ${it.message}", it) } //if failed to cache, just continue without it .onErrorResume { diff --git a/src/main/proto/cache.proto b/src/main/proto/cache.proto new file mode 100644 index 00000000..9edbe6a3 --- /dev/null +++ b/src/main/proto/cache.proto @@ -0,0 +1,40 @@ +syntax = "proto3"; +package emerald.dshackle; +option java_package = "io.emeraldpay.dshackle.proto"; +option java_outer_classname = "CachesProto"; + +message ValueContainer { + ValueType type = 1; + Compression compression = 2; + bytes value = 3; + + oneof meta_type { + BlockMeta block_meta = 4; + TxMeta tx_meta = 5; + } + + enum ValueType { + UNKNOWN = 0; + BLOCK = 1; + TX = 2; + } + + enum Compression { + NONE = 0; + } +} + +message BlockMeta { + uint64 height = 1; + bytes hash = 2; + bytes difficulty = 3; + uint64 timestamp = 4; + repeated bytes tx_hashes = 5; +} + +message TxMeta { + uint64 height = 1; + bytes hash = 2; + bytes block_hash = 3; +} + diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy index 029c4d47..a6de791f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.cache import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.test.IntegrationTestingCommons import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.grpc.Chain @@ -20,7 +21,8 @@ import java.time.temporal.ChronoUnit @IgnoreIf({ IntegrationTestingCommons.isDisabled("redis") }) class BlocksRedisCacheSpec extends Specification { - StatefulRedisConnection redis + StatefulRedisConnection redis + BlocksRedisCache cache String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" @@ -30,17 +32,43 @@ class BlocksRedisCacheSpec extends Specification { ObjectMapper objectMapper = TestingCommons.objectMapper() def setup() { - RedisClient client = IntegrationTestingCommons.redis() - StatefulRedisConnection connection = client.connect(); - connection.sync().flushdb() - redis = connection + redis = IntegrationTestingCommons.redisConnection() + redis.sync().flushdb() + cache = new BlocksRedisCache( + redis.reactive(), Chain.ETHEREUM + ) + } + + def "Decode encoded"() { + setup: + BlockContainer cont = new BlockContainer( + 100, + BlockId.from(hash3), + BigInteger.valueOf(10515), + Instant.ofEpochSecond(10501050), + false, + "test".bytes, + [TxId.from(hash2), TxId.from(hash1)] + ) + + when: + def enc = cache.toProto(cont) + def dec = cache.fromProto(enc) + + then: + dec.height == 100 + dec.hash.toHex() == hash3 + dec.difficulty.toString() == "10515" + dec.timestamp == Instant.ofEpochSecond(10501050) + dec.json == "test".bytes + dec.transactions.size() == 2 + dec.transactions[0].toHex() == hash2 + dec.transactions[1].toHex() == hash1 + dec == cont } 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) @@ -59,9 +87,6 @@ class BlocksRedisCacheSpec extends Specification { 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) @@ -86,9 +111,6 @@ class BlocksRedisCacheSpec extends Specification { 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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy index ca0d73de..cadcdda8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.cache import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.test.IntegrationTestingCommons @@ -15,6 +16,9 @@ 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 io.lettuce.core.codec.ByteArrayCodec +import io.lettuce.core.codec.RedisCodec +import io.lettuce.core.codec.StringCodec import spock.lang.IgnoreIf import spock.lang.Specification @@ -33,11 +37,29 @@ class TxRedisCacheSpec extends Specification { def objectMapper = TestingCommons.objectMapper() def setup() { - RedisClient client = IntegrationTestingCommons.redis() - StatefulRedisConnection connection = client.connect(); - connection.sync().flushdb() - StatefulRedisConnection redis = connection - cache = new TxRedisCache(redis.reactive(), Chain.ETHEREUM, TestingCommons.objectMapper()) + StatefulRedisConnection redis = IntegrationTestingCommons.redisConnection() + redis.sync().flushdb() + cache = new TxRedisCache(redis.reactive(), Chain.ETHEREUM) + } + + def "Decode encoded"() { + setup: + TxContainer cont = new TxContainer( + 2000, + TxId.from(hash1), + BlockId.from(hash2), + "test".bytes + ) + when: + def enc = cache.toProto(cont) + def dec = cache.fromProto(enc) + + then: + dec.height == 2000 + dec.hash.toHex() == hash1 + dec.blockId.toHex() == hash2 + dec.json == "test".bytes + dec == cont } def "Add and read"() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/IntegrationTestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/IntegrationTestingCommons.groovy index acdeac36..6a35db73 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/IntegrationTestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/IntegrationTestingCommons.groovy @@ -1,6 +1,10 @@ package io.emeraldpay.dshackle.test import io.lettuce.core.RedisClient +import io.lettuce.core.api.StatefulRedisConnection +import io.lettuce.core.codec.ByteArrayCodec +import io.lettuce.core.codec.RedisCodec +import io.lettuce.core.codec.StringCodec class IntegrationTestingCommons { @@ -22,4 +26,7 @@ class IntegrationTestingCommons { return RedisClient.create("redis://${host}:${port}") } + static StatefulRedisConnection redisConnection() { + return redis().connect(RedisCodec.of(StringCodec.ASCII, ByteArrayCodec.INSTANCE)) + } }