solution: caching for eth_getTransactionReceipt

fix: #26

src/main/kotlin/io/emeraldpay/dshackle/Global.kt

@@ -20,6 +20,8 @@ import com.fasterxml.jackson.databind.DeserializationFeature
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.module.SimpleModule
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.infinitape.etherjar.rpc.json.TransactionReceiptJson
+import io.infinitape.etherjar.rpc.json.TransactionReceiptJsonDeserializer
 import java.text.SimpleDateFormat
 import java.util.*
 

src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt

@@ -27,61 +27,33 @@ 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
 
 /**
  * Cache blocks in Redis database
  */
 class BlocksRedisCache(
-        private val redis: RedisReactiveCommands<String, ByteArray>,
+        redis: RedisReactiveCommands<String, ByteArray>,
-        private val chain: Chain
+        chain: Chain
-) : Reader<BlockId, BlockContainer> {
+) : Reader<BlockId, BlockContainer>,
+        OnBlockRedisCache<BlockContainer>(redis, chain, CachesProto.ValueContainer.ValueType.BLOCK) {
 
     companion object {
         private val log = LoggerFactory.getLogger(BlocksRedisCache::class.java)
-        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: BlockId): Mono<BlockContainer> {
+    override fun buildMeta(block: BlockContainer): CachesProto.BlockMeta.Builder {
-        return redis.get(key(key))
+        val meta = super.buildMeta(block)
-                .map { data ->
+        block.transactions.forEach {
-                    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 meta
-        return CachesProto.ValueContainer.newBuilder()
-                .setType(CachesProto.ValueContainer.ValueType.BLOCK)
-                .setValue(ByteString.copyFrom(value.json!!))
-                .setBlockMeta(meta)
-                .build()
-                .toByteArray()
     }
 
-    fun fromProto(msg: ByteArray): BlockContainer {
+    override fun serializeValue(value: BlockContainer): ByteArray {
-        val value = CachesProto.ValueContainer.parseFrom(msg)
+        return value.json!!
-        if (value.type != CachesProto.ValueContainer.ValueType.BLOCK) {
+    }
-            throw IllegalArgumentException("Expect BLOCK value, receive ${value.type}")
+
-        }
+    override fun deserializeValue(value: CachesProto.ValueContainer): BlockContainer {
         if (!value.hasBlockMeta()) {
             throw IllegalArgumentException("Container doesn't have Block Meta")
         }
@@ -100,51 +72,14 @@ class BlocksRedisCache(
         )
     }
 
-    fun evict(id: BlockId): Mono<Void> {
-        return Mono.just(id)
-                .flatMap {
-                    redis.del(key(it))
-                }
-                .then()
-    }
-
-    /**
-     * Add to cache.
-     * Note that it returns Mono<Void> which must be subscribed to actually save
-     */
     fun add(block: BlockContainer): Mono<Void> {
         if (block.timestamp == null || block.hash == null) {
             return Mono.empty()
         }
-        return Mono.just(block)
+        if (block.full) {
-                .flatMap { block ->
+            return Mono.error(IllegalArgumentException("Full Block is not supposed to be cached"))
-                    //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
+        return super.add(block, block)
-                    //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) {
-                        val key = key(block.hash)
-                        val value = toProto(block)
-                        redis.setex(key, ttl, value)
-                    } else {
-                        Mono.empty()
-                    }
-                }
-                .doOnError {
-                    log.warn("Failed to save Block to Redis: ${it.message}")
-                }
-                //if failed to cache, just continue without it
-                .onErrorResume {
-                    Mono.empty()
-                }
-                .then()
     }
 
-    /**
-     * Key in Redis
-     */
-    fun key(hash: BlockId): String {
-        return "block:${chain.id}:${hash.toHex()}"
-    }
 }

src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt

@@ -15,17 +15,16 @@
  */
 package io.emeraldpay.dshackle.cache
 
-import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
-import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.*
-import io.emeraldpay.dshackle.data.BlockId
-import io.emeraldpay.dshackle.data.TxContainer
-import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.CompoundReader
+import io.emeraldpay.dshackle.reader.EmptyReader
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumFullBlocksReader
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionJson
+import io.infinitape.etherjar.rpc.json.TransactionReceiptJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
@@ -35,7 +34,8 @@ open class Caches(
         private val blocksByHeight: HeightCache,
         private val memTxsByHash: TxMemCache,
         private val redisBlocksByHash: BlocksRedisCache?,
-        private val redisTxsByHash: TxRedisCache?
+        private val redisTxsByHash: TxRedisCache?,
+        private val redisReceipts: ReceiptRedisCache?
 ) {
 
     companion object {
@@ -54,6 +54,7 @@ open class Caches(
 
     private val blocksByHash: Reader<BlockId, BlockContainer>
     private val txsByHash: Reader<TxId, TxContainer>
+    private val receiptByHash: Reader<TxId, ByteArray>
 
     init {
         blocksByHash = if (redisBlocksByHash == null) {
@@ -66,6 +67,12 @@ open class Caches(
         } else {
             CompoundReader(memTxsByHash, redisTxsByHash)
         }
+        receiptByHash = redisReceipts ?: EmptyReader()
+    }
+
+    fun setHead(head: Head) {
+        redisTxsByHash?.head = head
+        redisReceipts?.head = head
     }
 
     /**
@@ -79,6 +86,11 @@ open class Caches(
         }
     }
 
+    open fun cacheReceipt(tag: Tag, data: DefaultContainer<TransactionReceiptJson>) {
+        //TODO move subscription to the caller
+        redisReceipts?.add(data)?.subscribe()
+    }
+
     fun cache(tag: Tag, tx: TxContainer) {
         //do not cache transactions that are not in a block yet
         if (tx.blockId == null) {
@@ -166,6 +178,10 @@ open class Caches(
         return BlockByHeight(blocksByHeight, EthereumFullBlocksReader(blocksByHash, txsByHash))
     }
 
+    fun getReceipts(): Reader<TxId, ByteArray> {
+        return receiptByHash
+    }
+
     enum class Tag {
         /**
          * Latest data produced by blockchain
@@ -184,6 +200,7 @@ open class Caches(
         private var txsByHash: TxMemCache? = null
         private var redisBlocksByHash: BlocksRedisCache? = null
         private var redisTxsByHash: TxRedisCache? = null
+        private var redisReceiptCache: ReceiptRedisCache? = null
 
         fun setBlockByHash(cache: BlocksMemCache): Builder {
             blocksByHash = cache
@@ -210,6 +227,11 @@ open class Caches(
             return this
         }
 
+        fun setReceipts(cache: ReceiptRedisCache): Builder {
+            redisReceiptCache = cache
+            return this
+        }
+
         fun build(): Caches {
             if (blocksByHash == null) {
                 blocksByHash = BlocksMemCache()
@@ -220,7 +242,7 @@ open class Caches(
             if (txsByHash == null) {
                 txsByHash = TxMemCache()
             }
-            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash)
+            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash, redisReceiptCache)
         }
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt

@@ -15,7 +15,6 @@
  */
 package io.emeraldpay.dshackle.cache
 
-import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.config.CacheConfig
 import io.emeraldpay.grpc.Chain
 import io.lettuce.core.RedisClient
@@ -96,6 +95,7 @@ class CachesFactory(
         redis?.let { redis ->
             caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain))
             caches.setTxByHash(TxRedisCache(redis.reactive(), chain))
+            caches.setReceipts(ReceiptRedisCache(redis.reactive(), chain))
         }
         return caches.build()
     }

src/main/kotlin/io/emeraldpay/dshackle/cache/OnBlockRedisCache.kt

@@ -0,0 +1,142 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.cache
+
+import com.google.protobuf.ByteString
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.proto.CachesProto
+import io.emeraldpay.dshackle.proto.CachesProto.ValueContainer
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.grpc.Chain
+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
+
+abstract class OnBlockRedisCache<T>(
+        private val redis: RedisReactiveCommands<String, ByteArray>,
+        private val chain: Chain,
+        private val valueType: ValueContainer.ValueType
+) : Reader<BlockId, T> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(OnBlockRedisCache::class.java)
+
+        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 = 3
+    }
+
+    private val prefix: String = when (valueType) {
+        ValueContainer.ValueType.BLOCK -> "block"
+        else -> throw IllegalStateException("No prefix for value type $valueType")
+    }
+
+    fun toProto(block: BlockContainer, value: T): ValueContainer {
+        return ValueContainer.newBuilder()
+                .setType(valueType)
+                .setValue(ByteString.copyFrom(serializeValue(value)))
+                .setBlockMeta(buildMeta(block))
+                .build()
+    }
+
+    open fun buildMeta(block: BlockContainer): CachesProto.BlockMeta.Builder {
+        return CachesProto.BlockMeta.newBuilder()
+                .setHash(ByteString.copyFrom(block.hash.value))
+                .setHeight(block.height)
+                .setDifficulty(ByteString.copyFrom(block.difficulty.toByteArray()))
+                .setTimestamp(block.timestamp.toEpochMilli())
+    }
+
+    abstract fun serializeValue(value: T): ByteArray
+
+    fun fromProto(msg: ByteArray): T {
+        val value = ValueContainer.parseFrom(msg)
+        if (value.type != valueType) {
+            val error = "Expected $valueType value, received ${value.type}"
+            log.warn(error)
+            throw IllegalArgumentException(error)
+        }
+        return deserializeValue(value)
+    }
+
+    abstract fun deserializeValue(value: ValueContainer): T
+
+    /**
+     * Key in Redis
+     */
+    fun key(hash: BlockId): String {
+        return "${prefix}:${chain.id}:${hash.toHex()}"
+    }
+
+    /**
+     * Add to cache.
+     * Note that it returns Mono<Void> which must be subscribed to actually save
+     */
+    open fun add(block: BlockContainer, value: T): Mono<Void> {
+        return Mono.just(block)
+                .flatMap { block ->
+                    val ttl = cachingTime(block.timestamp!!)
+                    if (ttl > MIN_CACHE_TIME_SECONDS) {
+                        val key = key(block.hash)
+                        val proto = toProto(block, value)
+                        redis.setex(key, ttl, proto.toByteArray())
+                    } else {
+                        Mono.empty()
+                    }
+                }
+                .doOnError {
+                    log.warn("Failed to save Block to Redis: ${it.message}")
+                }
+                //if failed to cache, just continue without it
+                .onErrorResume {
+                    Mono.empty()
+                }
+                .then()
+    }
+
+    /**
+     * Calculate time to cache the value
+     */
+    fun cachingTime(blockTime: Instant): Long {
+        //default caching time is age of the block, i.e. block create hour ago
+        //keep for hour, but block created 10 seconds ago cache only for 10 seconds, because it
+        //still can be replaced in the blockchain
+        val age = Instant.now().epochSecond - blockTime.epochSecond
+        return min(age, TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES))
+    }
+
+    fun evict(id: BlockId): Mono<Void> {
+        return Mono.just(id)
+                .flatMap {
+                    redis.del(key(it))
+                }
+                .then()
+    }
+
+    override fun read(key: BlockId): Mono<T> {
+        return redis.get(key(key))
+                .map { data ->
+                    fromProto(data)
+                }.onErrorResume {
+                    Mono.empty()
+                }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt

@@ -0,0 +1,168 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.cache
+
+import com.google.protobuf.ByteString
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.proto.CachesProto
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.grpc.Chain
+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
+
+abstract class OnTxRedisCache<T>(
+        private val redis: RedisReactiveCommands<String, ByteArray>,
+        private val chain: Chain,
+        private val valueType: CachesProto.ValueContainer.ValueType
+) : Reader<TxId, T> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(OnTxRedisCache::class.java)
+
+        // max caching time is 24 hours
+        const val MAX_CACHE_TIME_HOURS = 24L
+        const val MIN_CACHE_TIME_SECONDS = 30L
+        const val BLOCK_TIME_SECONDS = 10L
+    }
+
+    private val prefix: String = when (valueType) {
+        CachesProto.ValueContainer.ValueType.TX -> "tx"
+        CachesProto.ValueContainer.ValueType.TX_RECEIPT -> "tx-receipt"
+        else -> throw IllegalStateException("No prefix for value type $valueType")
+    }
+
+    var head: Head? = null
+
+    /**
+     * Key in Redis
+     */
+    fun key(hash: TxId): String {
+        return "${prefix}:${chain.id}:${hash.toHex()}"
+    }
+
+    fun evict(block: BlockContainer): Mono<Void> {
+        return Mono.just(block)
+                .map { block ->
+                    block.transactions.map {
+                        key(it)
+                    }.toTypedArray()
+                }.flatMap { keys ->
+                    redis.del(*keys)
+                }.then()
+    }
+
+    fun evict(id: TxId): Mono<Void> {
+        return Mono.just(id)
+                .flatMap {
+                    redis.del(key(it))
+                }
+                .then()
+    }
+
+    fun toProto(id: TxId, value: T): ByteArray {
+        val meta = buildMeta(id, value)
+
+        return CachesProto.ValueContainer.newBuilder()
+                .setType(valueType)
+                .setValue(ByteString.copyFrom(serializeValue(value)))
+                .setTxMeta(meta)
+                .build()
+                .toByteArray()
+    }
+
+    open fun buildMeta(id: TxId, value: T): CachesProto.TxMeta.Builder {
+        return CachesProto.TxMeta.newBuilder()
+                .setHash(ByteString.copyFrom(id.value))
+    }
+
+    abstract fun serializeValue(value: T): ByteArray
+
+    fun fromProto(msg: ByteArray): T {
+        val value = CachesProto.ValueContainer.parseFrom(msg)
+        if (value.type != valueType) {
+            val error = "Expected $valueType value, received ${value.type}"
+            log.warn(error)
+            throw IllegalArgumentException(error)
+        }
+        return deserializeValue(value)
+    }
+
+    abstract fun deserializeValue(value: CachesProto.ValueContainer): T
+
+    override fun read(key: TxId): Mono<T> {
+        return redis.get(key(key))
+                .map { data ->
+                    fromProto(data)
+                }.onErrorResume {
+                    Mono.empty()
+                }
+    }
+
+    fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono<Void> {
+        return Mono.just(id)
+                .flatMap {
+                    val key = key(it)
+                    val encodedValue = toProto(it, value)
+                    val ttl = if (block?.timestamp != null) {
+                        cachingTime(block.timestamp)
+                    } else {
+                        cachingTime(blockHeight)
+                    }
+                    //store
+                    redis.setex(key, ttl, encodedValue)
+                }
+                .doOnError {
+                    log.warn("Failed to save TX to Redis: ${it.message}", it)
+                }
+                //if failed to cache, just continue without it
+                .onErrorResume {
+                    Mono.empty()
+                }
+                .then()
+    }
+
+    /**
+     * Calculate time to cache the value, based on block time
+     */
+    fun cachingTime(blockTime: Instant): Long {
+        //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 - blockTime.epochSecond
+        return min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS))
+    }
+
+    /**
+     * Calculate time to cache the value, based on block height
+     */
+    fun cachingTime(blockHeight: Long?): Long {
+        if (blockHeight == null) {
+            return MIN_CACHE_TIME_SECONDS
+        }
+        val headHeight = head?.getCurrentHeight() ?: return MIN_CACHE_TIME_SECONDS
+        val confirmations = headHeight - blockHeight
+        if (confirmations <= 0) {
+            return MIN_CACHE_TIME_SECONDS
+        }
+        return min(confirmations * BLOCK_TIME_SECONDS, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS))
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt

@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.cache
+
+import io.emeraldpay.dshackle.data.DefaultContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.proto.CachesProto
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.rpc.json.TransactionReceiptJson
+import io.lettuce.core.api.reactive.RedisReactiveCommands
+import reactor.core.publisher.Mono
+
+class ReceiptRedisCache(
+        redis: RedisReactiveCommands<String, ByteArray>,
+        chain: Chain
+) : OnTxRedisCache<ByteArray>(redis, chain, CachesProto.ValueContainer.ValueType.TX_RECEIPT) {
+
+    override fun deserializeValue(value: CachesProto.ValueContainer): ByteArray {
+        return value.value.toByteArray()
+    }
+
+    override fun serializeValue(value: ByteArray): ByteArray {
+        return value
+    }
+
+    fun add(json: DefaultContainer<TransactionReceiptJson>): Mono<Void> {
+        return super.add(json.txId!!, json.json!!, null, json.height)
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt

@@ -38,28 +38,15 @@ import kotlin.math.min
 class TxRedisCache(
         private val redis: RedisReactiveCommands<String, ByteArray>,
         private val chain: Chain
-) : Reader<TxId, TxContainer> {
+) : Reader<TxId, TxContainer>,
+        OnTxRedisCache<TxContainer>(redis, chain, CachesProto.ValueContainer.ValueType.TX) {
 
     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: TxId): Mono<TxContainer> {
+    override fun buildMeta(id: TxId, value: TxContainer): CachesProto.TxMeta.Builder {
-        return redis.get(key(key))
+        val meta = super.buildMeta(id, value)
-                .map { data ->
-                    fromProto(data)
-                }.onErrorResume {
-                    Mono.empty()
-                }
-    }
-
-    fun toProto(value: TxContainer): ByteArray {
-        val meta = CachesProto.TxMeta.newBuilder()
-                .setHash(ByteString.copyFrom(value.hash.value))
-
         value.height?.let {
             meta.setHeight(it)
         }
@@ -67,20 +54,14 @@ class TxRedisCache(
         value.blockId?.value?.let {
             meta.setBlockHash(ByteString.copyFrom(it))
         }
-
+        return meta
-        return CachesProto.ValueContainer.newBuilder()
-                .setType(CachesProto.ValueContainer.ValueType.TX)
-                .setValue(ByteString.copyFrom(value.json!!))
-                .setTxMeta(meta)
-                .build()
-                .toByteArray()
     }
 
-    fun fromProto(msg: ByteArray): TxContainer {
+    override fun serializeValue(value: TxContainer): ByteArray {
-        val value = CachesProto.ValueContainer.parseFrom(msg)
+        return value.json!!
-        if (value.type != CachesProto.ValueContainer.ValueType.TX) {
+    }
-            throw IllegalArgumentException("Expect TX value, receive ${value.type}")
+
-        }
+    override fun deserializeValue(value: CachesProto.ValueContainer): TxContainer {
         if (!value.hasTxMeta()) {
             throw IllegalArgumentException("Container doesn't have Tx Meta")
         }
@@ -93,55 +74,8 @@ class TxRedisCache(
         )
     }
 
-    fun evict(block: BlockContainer): Mono<Void> {
-        return Mono.just(block)
-                .map { block ->
-                    block.transactions.map {
-                        key(it)
-                    }.toTypedArray()
-                }.flatMap { keys ->
-                    redis.del(*keys)
-                }.then()
-    }
-
-    fun evict(id: TxId): Mono<Void> {
-        return Mono.just(id)
-                .flatMap {
-                    redis.del(key(it))
-                }
-                .then()
-    }
-
     fun add(tx: TxContainer, block: BlockContainer): Mono<Void> {
-        if (tx.blockId == null || block.hash == null || tx.blockId != block.hash || block.timestamp == null) {
+        return super.add(tx.hash, tx, block, tx.height)
-            return Mono.empty()
-        }
-        return Mono.just(Tuples.of(tx, block))
-                .flatMap {
-                    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))
-                    //store
-                    redis.setex(key, ttl, value)
-                }
-                .doOnError {
-                    log.warn("Failed to save TX to Redis: ${it.message}", it)
-                }
-                //if failed to cache, just continue without it
-                .onErrorResume {
-                    Mono.empty()
-                }
-                .then()
     }
 
-    /**
-     * Key in Redis
-     */
-    fun key(hash: TxId): String {
-        return "tx:${chain.id}:${hash.toHex()}"
-    }
 }

src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt

@@ -0,0 +1,31 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.data
+
+import org.slf4j.LoggerFactory
+
+class DefaultContainer<T>(
+        val txId: TxId?,
+        val blockId: BlockId?,
+        val height: Long?,
+        json: ByteArray,
+        parsed: T
+) : SourceContainer(json, parsed) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(DefaultContainer::class.java)
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt

@@ -16,7 +16,6 @@
  */
 package io.emeraldpay.dshackle.data
 
-import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.infinitape.etherjar.rpc.json.TransactionJson
 

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -144,6 +144,10 @@ open class NativeCall(
                 .map {
                     ctx.withPayload(it.value)
                 }
+                .doOnNext {
+                    ctx.upstream.postprocessor
+                            .onReceive(ctx.payload.method, ctx.payload.params, it.payload)
+                }
                 .onErrorMap {
                     log.error("Failed to make a call", it)
                     if (it is CallFailure) it

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -44,7 +44,8 @@ import kotlin.concurrent.withLock
 abstract class Multistream(
         val chain: Chain,
         private val upstreams: MutableList<Upstream>,
-        val caches: Caches
+        val caches: Caches,
+        val postprocessor: RequestPostprocessor
 ) : Upstream, Lifecycle {
 
     companion object {
@@ -189,6 +190,7 @@ abstract class Multistream(
                 caches.cache(Caches.Tag.LATEST, it)
             }
         }
+        caches.setHead(head)
     }
 
     abstract fun updateHead(): Head

src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt

@@ -0,0 +1,10 @@
+package io.emeraldpay.dshackle.upstream
+
+interface RequestPostprocessor {
+
+    fun onReceive(method: String, params: List<Any>, json: ByteArray)
+
+    class Empty : RequestPostprocessor {
+        override fun onReceive(method: String, params: List<Any>, json: ByteArray) {}
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt

@@ -32,7 +32,7 @@ open class BitcoinMultistream(
         chain: Chain,
         val upstreams: MutableList<BitcoinUpstream>,
         caches: Caches
-) : Multistream(chain, upstreams as MutableList<Upstream>, caches), Lifecycle {
+) : Multistream(chain, upstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty()), Lifecycle {
 
     companion object {
         private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt

@@ -0,0 +1,60 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.DefaultContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.upstream.RequestPostprocessor
+import io.infinitape.etherjar.rpc.json.TransactionReceiptJson
+import org.slf4j.LoggerFactory
+
+class CacheRequested(
+        private val caches: Caches
+) : RequestPostprocessor {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(CacheRequested::class.java)
+    }
+
+    override fun onReceive(method: String, params: List<Any>, json: ByteArray) {
+        try {
+            if (method == "eth_getTransactionReceipt") {
+                cacheTxReceipt(params, json)
+            }
+        } catch (e: Throwable) {
+            log.warn("Failed to cache result", e)
+        }
+    }
+
+    fun cacheTxReceipt(params: List<Any>, json: ByteArray) {
+        if (params.size != 1) {
+            return
+        }
+        val parsed = Global.objectMapper.readValue(json, TransactionReceiptJson::class.java)
+        val value = DefaultContainer<TransactionReceiptJson>(
+                TxId.from(parsed.transactionHash),
+                BlockId.from(parsed.blockHash),
+                parsed.blockNumber,
+                json,
+                parsed
+        )
+        caches.cacheReceipt(Caches.Tag.REQUESTED, value)
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt

@@ -31,7 +31,7 @@ open class EthereumMultistream(
         chain: Chain,
         val upstreams: MutableList<EthereumUpstream>,
         caches: Caches
-) : Multistream(chain, upstreams as MutableList<Upstream>, caches) {
+) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)) {
 
     companion object {
         private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt

@@ -152,6 +152,10 @@ open class EthereumReader(
         )
     }
 
+    fun receipts(): Reader<TxId, ByteArray> {
+        return caches.getReceipts()
+    }
+
     override fun isRunning(): Boolean {
         //TODO should be always running?
         return up.isRunning

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt

@@ -16,7 +16,6 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt

@@ -29,6 +29,13 @@ import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import java.math.BigInteger
 
+/**
+ * Reader for JSON RPC requests. Verifies if the method is allowed, transforms if necessary, and calls EthereumReader for data.
+ * It provides data only if it's available through the router (cached, head, etc).
+ * If data is not available locally then it returns `empty`; at this case the caller should call the remote node for actual data.
+ *
+ * @see EthereumReader
+ */
 class NativeCallRouter(
         private val reader: EthereumReader,
         private val methods: CallMethods,
@@ -100,6 +107,18 @@ class NativeCallRouter(
             method == "eth_getBlockByNumber" -> {
                 getBlockByNumber(params)
             }
+            method == "eth_getTransactionReceipt" -> {
+                if (params.size != 1) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
+                }
+                val hash: TxId
+                try {
+                    hash = TxId.from(params[0].toString())
+                } catch (e: IllegalArgumentException) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
+                }
+                reader.receipts().read(hash)
+            }
             else -> null
         }
     }

src/main/proto/cache.proto

@@ -17,6 +17,7 @@ message ValueContainer {
         UNKNOWN = 0;
         BLOCK = 1;
         TX = 2;
+        TX_RECEIPT = 3;
     }
 
     enum Compression {

src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy

@@ -21,12 +21,10 @@ 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
 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
@@ -69,8 +67,8 @@ class BlocksRedisCacheSpec extends Specification {
         )
 
         when:
-        def enc = cache.toProto(cont)
+        def enc = cache.toProto(cont, cont)
-        def dec = cache.fromProto(enc)
+        def dec = cache.deserializeValue(enc)
 
         then:
         dec.height == 100

src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptRedisCacheSpec.groovy

@@ -0,0 +1,96 @@
+package io.emeraldpay.dshackle.cache
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.DefaultContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.test.IntegrationTestingCommons
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.TransactionReceiptJson
+import io.lettuce.core.api.StatefulRedisConnection
+import spock.lang.IgnoreIf
+import spock.lang.Specification
+
+@IgnoreIf({ IntegrationTestingCommons.isDisabled("redis") })
+class ReceiptRedisCacheSpec extends Specification {
+
+    StatefulRedisConnection<String, byte[]> redis
+    ReceiptRedisCache cache
+    ObjectMapper objectMapper = Global.objectMapper
+
+    String receiptJson = '''
+    {
+        "blockHash": "0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae",
+        "blockNumber": "0x213b8a",
+        "contractAddress": null,
+        "cumulativeGasUsed": "0x5208",
+        "gasUsed": "0x5208",
+        "logs": [],
+        "transactionHash": "0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197",
+        "transactionIndex": "0x00"
+      }
+    '''
+    TransactionReceiptJson receipt = new TransactionReceiptJson().tap {
+        transactionHash = TransactionId.from("0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197")
+        transactionIndex = 0
+        blockHash = BlockHash.from("0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae")
+        blockNumber = 0x213b8a
+        cumulativeGasUsed = 0x5208
+        gasUsed = 0x5208
+        logs = []
+    }
+
+    def setup() {
+        redis = IntegrationTestingCommons.redisConnection()
+        redis.sync().flushdb()
+        cache = new ReceiptRedisCache(
+                redis.reactive(), Chain.ETHEREUM
+        )
+    }
+
+    def "Add and read"() {
+        setup:
+
+        def container = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                receiptJson.bytes,
+                receipt
+        )
+
+        when:
+        cache.add(container).block()
+        def act = cache.read(TxId.from(receipt.transactionHash)).block()
+        then:
+        act != null
+        objectMapper.readValue(act, TransactionReceiptJson) == receipt
+    }
+
+    def "Evict value"() {
+        setup:
+
+        def container = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                receiptJson.bytes,
+                receipt
+        )
+
+        when:
+        cache.add(container).block()
+        def act = cache.read(TxId.from(receipt.transactionHash)).block()
+        then:
+        act != null
+
+        when:
+        cache.evict(TxId.from(receipt.transactionHash)).subscribe()
+        act = cache.read(TxId.from(receipt.transactionHash)).block()
+        then:
+        act == null
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy

@@ -68,7 +68,7 @@ class TxRedisCacheSpec extends Specification {
                 null
         )
         when:
-        def enc = cache.toProto(cont)
+        def enc = cache.toProto(cont.hash, cont)
         def dec = cache.fromProto(enc)
 
         then:

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy

@@ -0,0 +1,45 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.data.DefaultContainer
+import spock.lang.Specification
+
+class CacheRequestedSpec extends Specification {
+
+    def "Do nothing if unsupported method"() {
+        setup:
+        def caches = Mock(Caches)
+        CacheRequested instance = new CacheRequested(caches)
+        when:
+        instance.onReceive("eth_hashrate", [], '"0x38a"'.bytes)
+        then:
+        0 * caches._(*_)
+    }
+
+    def "Caches tx receipt"() {
+        setup:
+        def caches = Mock(Caches)
+        CacheRequested instance = new CacheRequested(caches)
+        def json = '''{
+            "blockHash": "0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae",
+            "blockNumber": "0x213b8a",
+            "contractAddress": null,
+            "cumulativeGasUsed": "0x5208",
+            "gasUsed": "0x5208",
+            "logs": [],
+            "transactionHash": "0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197",
+            "transactionIndex": "0x00"
+        }'''.bytes
+
+        when:
+        instance.onReceive("eth_getTransactionReceipt", ["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json)
+
+        then:
+        1 * caches.cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer it ->
+            it.height == 0x213b8a &&
+                    it.txId.toHex() == "5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197" &&
+                    it.json == json
+        })
+    }
+
+}