problem: redis cached makes JSON parsing only to extract metadata, ineffective

solution: encode in protobuf with metadata included

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()  {

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

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<String, String>,
+        private val redis: RedisReactiveCommands<String, ByteArray>,
-        private val chain: Chain,
+        private val chain: Chain
-        private val objectMapper: ObjectMapper
 ) : Reader<BlockId, BlockContainer> {
 
     companion object {
@@ -34,13 +34,56 @@ class BlocksRedisCache(
     override fun read(key: BlockId): Mono<BlockContainer> {
         return redis.get(key(key))
                 .map { data ->
-                    val block = objectMapper.readValue(data, BlockJson::class.java)
+                    fromProto(data)
-                    BlockContainer.from(block, objectMapper)
                 }.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<Void> {
         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 {

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<String, String>? = null
+    private var redis: StatefulRedisConnection<String, ByteArray>? = null
     private val all = EnumMap<Chain, Caches>(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.setBlockByHash(BlocksRedisCache(redis.reactive(), chain))
-            caches.setTxByHash(TxRedisCache(redis.reactive(), chain, objectMapper))
+            caches.setTxByHash(TxRedisCache(redis.reactive(), chain))
         }
         return caches.build()
     }

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<String, String>,
+        private val redis: RedisReactiveCommands<String, ByteArray>,
-        private val chain: Chain,
+        private val chain: Chain
-        private val objectMapper: ObjectMapper
 ) : Reader<TxId, TxContainer> {
 
     companion object {
@@ -35,14 +35,46 @@ class TxRedisCache(
     override fun read(key: TxId): Mono<TxContainer> {
         return redis.get(key(key))
                 .map { data ->
-                    val json = data
+                    fromProto(data)
-                    val tx = objectMapper.readValue(json, TransactionJson::class.java)
-                    TxContainer.from(tx, objectMapper)
                 }.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<Void> {
         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 {

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;
+}
+

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<String, String> redis
+    StatefulRedisConnection<String, byte[]> 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()
+        redis = IntegrationTestingCommons.redisConnection()
-        StatefulRedisConnection<String, String> connection = client.connect();
+        redis.sync().flushdb()
-        connection.sync().flushdb()
+        cache = new BlocksRedisCache(
-        redis = connection
+                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<TransactionRefJson>()
         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<TransactionRefJson>()
         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<TransactionRefJson>()
         block.number = 100
         block.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS)

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<String, byte[]> redis = IntegrationTestingCommons.redisConnection()
-        StatefulRedisConnection<String, String> connection = client.connect();
+        redis.sync().flushdb()
-        connection.sync().flushdb()
+        cache = new TxRedisCache(redis.reactive(), Chain.ETHEREUM)
-        StatefulRedisConnection<String, String> redis = connection
+    }
-        cache = new TxRedisCache(redis.reactive(), Chain.ETHEREUM, TestingCommons.objectMapper())
+
+    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"() {

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<String, byte[]> redisConnection() {
+        return redis().connect(RedisCodec.of(StringCodec.ASCII, ByteArrayCodec.INSTANCE))
+    }
 }