solution: cache requested transactions

2020-02-07 23:51:28 -05:00
parent d13397260f
commit d551d95a3f
25 changed files with 775 additions and 60 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt
@@ -10,7 +10,7 @@ import reactor.core.publisher.Mono
 /**
 * Connects two caches to read through them. First is cache height->hash, second is hash->block.
 */
-class BlockByHeight(
+open class BlockByHeight(
        private val heights: Reader<Long, BlockHash>,
        private val blocks: Reader<BlockHash, BlockJson<TransactionRefJson>>
 ): Reader<Long, BlockJson<TransactionRefJson>> {
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt
@@ -24,7 +24,7 @@ import reactor.core.publisher.Mono
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentLinkedQueue

-class BlocksMemCache(
+open class BlocksMemCache(
        val maxSize: Int = 64
 ): Reader<BlockHash, BlockJson<TransactionRefJson>> {

@@ -35,7 +35,11 @@ class BlocksMemCache(
        return Mono.justOrEmpty(mapping[key])
    }

-    fun add(block: BlockJson<TransactionRefJson>) {
+    open fun get(key: BlockHash): BlockJson<TransactionRefJson>? {
+        return mapping[key]
+    }
+
+    open fun add(block: BlockJson<TransactionRefJson>) {
        mapping.put(block.hash, block)
        queue.add(block.hash)

@@ -44,4 +48,5 @@ class BlocksMemCache(
            mapping.remove(old)
        }
    }
+
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt
@@ -0,0 +1,131 @@
+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
+
+open class Caches(
+        private val blocksByHash: BlocksMemCache,
+        private val blocksByHeight: HeightCache,
+        private val txsByHash: TxMemCache
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(Caches::class.java)
+
+        @JvmStatic
+        fun newBuilder(): Builder {
+            return Builder()
+        }
+
+        @JvmStatic
+        fun default(): Caches {
+            return newBuilder().build()
+        }
+    }
+
+    /**
+     * Cache data that was just requested
+     */
+    fun cacheRequested(data: Any) {
+        if (data is TransactionJson) {
+            cache(Tag.REQUESTED, data)
+        } else if (data is BlockJson<*>) {
+            cache(Tag.REQUESTED, data as BlockJson<TransactionRefJson>)
+        }
+    }
+
+    fun cache(tag: Tag, tx: TransactionJson) {
+        txsByHash.add(tx)
+    }
+
+    fun cache(tag: Tag, block: BlockJson<TransactionRefJson>) {
+        if (tag == Tag.LATEST) {
+            blocksByHash.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)
+                    evicted = true
+                }
+                if (!evicted) {
+                    txsByHash.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)
+                }
+            }
+        }
+    }
+
+    fun getBlocksByHash(): Reader<BlockHash, BlockJson<TransactionRefJson>> {
+        return blocksByHash
+    }
+
+    fun getBlockHashByHeight(): Reader<Long, BlockHash> {
+        return blocksByHeight
+    }
+
+    fun getBlocksByHeight(): Reader<Long, BlockJson<TransactionRefJson>> {
+        return BlockByHeight(blocksByHeight, blocksByHash)
+    }
+
+    fun getTxByHash(): Reader<TransactionId, TransactionJson> {
+        return txsByHash
+    }
+
+    enum class Tag {
+        /**
+         * Latest data produced by blockchain
+         */
+        LATEST,
+        /**
+         * Data requested by client
+         */
+        REQUESTED
+    }
+
+    class Builder() {
+        private var blocksByHash: BlocksMemCache? = null
+        private var blocksByHeight: HeightCache? = null
+        private var txsByHash: TxMemCache? = null
+
+        fun setBlockByHash(cache: BlocksMemCache): Builder {
+            blocksByHash = cache
+            return this
+        }
+
+        fun setBlockByHeight(cache: HeightCache): Builder {
+            blocksByHeight = cache
+            return this
+        }
+
+        fun setTxByHash(cache: TxMemCache): Builder {
+            txsByHash = cache
+            return this
+        }
+
+        fun build(): Caches {
+            if (blocksByHash == null) {
+                blocksByHash = BlocksMemCache()
+            }
+            if (blocksByHeight == null) {
+                blocksByHeight = HeightCache()
+            }
+            if (txsByHash == null) {
+                txsByHash = TxMemCache()
+            }
+            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!)
+        }
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesEnabled.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesEnabled.kt
@@ -0,0 +1,10 @@
+package io.emeraldpay.dshackle.cache
+
+/**
+ * Service is using caches
+ */
+interface CachesEnabled {
+
+    fun setCaches(caches: Caches)
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt
@@ -11,7 +11,7 @@ import java.util.concurrent.ConcurrentHashMap
 /**
 * Memory cache for blocks heights, keeps mapping height->hash.
 */
-class HeightCache(
+open class HeightCache(
        val maxSize: Int = 256
 ): Reader<Long, BlockHash> {

@@ -25,7 +25,8 @@ class HeightCache(
        return Mono.justOrEmpty(heights[key])
    }

-    fun add(block: BlockJson<TransactionRefJson>) {
+    open fun add(block: BlockJson<TransactionRefJson>): BlockHash? {
+        val existing = heights[block.number]
        heights[block.number] = block.hash

        // evict old numbers if full
@@ -34,5 +35,7 @@ class HeightCache(
            heights.remove(dropHeight)
            dropHeight++
        }
+
+        return existing
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt
@@ -0,0 +1,60 @@
+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 reactor.core.publisher.Mono
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ConcurrentLinkedQueue
+
+/**
+ * Memory cache for transactions
+ */
+open class TxMemCache(
+        // usually there is 100-150 tx per block on Ethereum, we keep data for about 32 blocks by default
+        private val maxSize: Int = 125 * 32
+): Reader<TransactionId, TransactionJson> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(TxMemCache::class.java)
+    }
+
+    private val mapping = ConcurrentHashMap<TransactionId, TransactionJson>()
+    private val queue = ConcurrentLinkedQueue<TransactionId>()
+
+    override fun read(key: TransactionId): Mono<TransactionJson> {
+        return Mono.justOrEmpty(mapping[key])
+    }
+
+    open fun evict(block: BlockJson<TransactionRefJson>) {
+        block.transactions.forEach {
+            mapping.remove(it.hash)
+        }
+    }
+
+    open fun evict(block: BlockHash) {
+        val ids = mapping.filter { it.value.blockHash == block }
+        ids.forEach {
+            mapping.remove(it.key)
+        }
+    }
+
+    open fun add(tx: TransactionJson) {
+        //do not cache fresh transactions
+        if (tx.blockHash == null || tx.blockNumber == null) {
+            return
+        }
+        mapping.put(tx.hash, tx)
+        queue.add(tx.hash)
+
+        while (queue.size > maxSize) {
+            val old = queue.remove()
+            mapping.remove(old)
+        }
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt
@@ -16,9 +16,7 @@
 package io.emeraldpay.dshackle.upstream

 import com.fasterxml.jackson.databind.ObjectMapper
-import io.emeraldpay.dshackle.cache.BlockByHeight
-import io.emeraldpay.dshackle.cache.BlocksMemCache
-import io.emeraldpay.dshackle.cache.HeightCache
+import io.emeraldpay.dshackle.cache.*
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import org.springframework.context.Lifecycle
@@ -32,12 +30,11 @@ import java.util.function.Predicate
 import kotlin.concurrent.withLock

 abstract class AggregatedUpstream(
-        val objectMapper: ObjectMapper
+        private val objectMapper: ObjectMapper,
+        val caches: Caches
 ): Upstream, Lifecycle {

    private var cacheSubscription: Disposable? = null
-    private val blockReaderByHash = BlocksMemCache()
-    private val blockReaderByHeight = HeightCache()
    var cache: CachingEthereumApi = CachingEthereumApi.empty()
    private val reconfigLock = ReentrantLock()
    private var callMethods: CallMethods? = null
@@ -109,10 +106,9 @@ abstract class AggregatedUpstream(
        reconfigLock.withLock {
            cacheSubscription?.dispose()
            cacheSubscription = head.getFlux().subscribe {
-                blockReaderByHash.add(it)
-                blockReaderByHeight.add(it)
+                caches.cache(Caches.Tag.LATEST, it)
            }
-            cache = CachingEthereumApi(objectMapper, blockReaderByHash, BlockByHeight(blockReaderByHeight, blockReaderByHash), head)
+            cache = CachingEthereumApi(objectMapper, caches, head)
        }
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt
@@ -16,17 +16,14 @@
 package io.emeraldpay.dshackle.upstream

 import com.fasterxml.jackson.databind.ObjectMapper
-import io.emeraldpay.dshackle.reader.EmptyReader
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.upstream.ethereum.EmptyEthereumHead
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.hex.HexQuantity
-import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.ResponseJson
-import io.infinitape.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import java.math.BigInteger
@@ -34,8 +31,7 @@ import java.util.function.Function

 open class CachingEthereumApi(
        private val objectMapper: ObjectMapper,
-        private val cache: Reader<BlockHash, BlockJson<TransactionRefJson>>,
-        private val cacheHeight: Reader<Long, BlockJson<TransactionRefJson>>,
+        private val caches: Caches,
        private val head: EthereumHead
 ): EthereumApi(objectMapper) {

@@ -44,10 +40,14 @@ open class CachingEthereumApi(

        @JvmStatic
        fun empty(): CachingEthereumApi {
-            return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyReader(), EmptyEthereumHead())
+            return CachingEthereumApi(ObjectMapper(), Caches.default(), EmptyEthereumHead())
        }
    }

+    private val cacheBlocks = caches.getBlocksByHash()
+    private val cacheHeight = caches.getBlocksByHeight()
+    private val cacheTx = caches.getTxByHash()
+
    override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
        return when (method) {
            "eth_blockNumber" ->
@@ -58,33 +58,54 @@ open class CachingEthereumApi(
                if (params.size == 2 && (params[1] == "false" || params[1] == false))
                    Mono.just(params[0])
                        .map { BlockHash.from(it as String) }
-                        .flatMap(cache::read)
-                        .map(toJson(id))
-                        .onErrorResume { t ->
-                            log.warn("Error during read from cache", t)
-                            Mono.empty()
-                        }
+                        .flatMap(cacheBlocks::read)
+                        .transform(converter(id))
+                        .transform(finalizer())
                else Mono.empty()
            "eth_getBlockByNumber" ->
                if (params.size == 2 && (params[1] == "false" || params[1] == false))
                    Mono.just(params[0])
                            .map { HexQuantity.from(it as String) }
-                            .filter {
-                                it.value < BigInteger.valueOf(Long.MAX_VALUE)
-                            }
+                            .filter { it.value < BigInteger.valueOf(Long.MAX_VALUE) }
                            .map { it.value.toLong() }
                            .flatMap(cacheHeight::read)
-                            .map(toJson(id))
-                            .onErrorResume { t ->
-                                log.warn("Error during read from cache", t)
-                                Mono.empty()
-                            }
+                            .transform(converter(id))
+                            .transform(finalizer())
+                else Mono.empty()
+            "eth_getTransactionByHash" ->
+                if (params.size == 1)
+                    Mono.just(params[0])
+                            .map { TransactionId.from(it as String) }
+                            .flatMap(cacheTx::read)
+                            .transform(converter(id))
+                            .transform(finalizer())
                else Mono.empty()
            else ->
                Mono.empty()
        }
    }

+    /**
+     * Convert to JSON RPC response
+     */
+    fun converter(id: Int): Function<in Mono<*>, out Mono<ByteArray>> {
+        return Function { mono ->
+            mono.map(toJson(id))
+        }
+    }
+
+    /**
+     * Handle errors and other stuff
+     */
+    fun finalizer(): Function<Mono<ByteArray>, Mono<ByteArray>> {
+        return Function { mono ->
+            mono.onErrorResume { t ->
+                log.warn("Error during read from cache", t)
+                Mono.empty()
+            }
+        }
+    }
+
    fun toJson(id: Int): Function<Any, ByteArray> {
        return Function { data ->
            val resp = ResponseJson<Any, Int>()
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream

 import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
@@ -31,8 +32,9 @@ import java.time.Duration
 open class ChainUpstreams (
        val chain: Chain,
        private val upstreams: MutableList<Upstream>,
+        caches: Caches,
        objectMapper: ObjectMapper
-) : AggregatedUpstream(objectMapper), Lifecycle {
+) : AggregatedUpstream(objectMapper, caches), Lifecycle {

    private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
    private var seq = 0
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt
@@ -144,6 +144,7 @@ open class ConfiguredUpstreams(
            }
            rpcApi = DirectEthereumApi(
                    rpcClient.build(),
+                    null,
                    objectMapper,
                    methods
            ).apply {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
@@ -16,6 +16,8 @@
 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.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -50,12 +52,18 @@ class CurrentUpstreams(
                log.info("Upstream ${change.upstream.getId()} with chain $chain has been removed")
            } else {
                if (current == null) {
-                    val created = ChainUpstreams(chain, ArrayList<Upstream>(), objectMapper)
+                    val created = ChainUpstreams(chain, ArrayList<Upstream>(), Caches.default(), objectMapper)
+                    if (up is CachesEnabled) {
+                        up.setCaches(created.caches)
+                    }
                    created.addUpstream(up)
                    created.start()
                    chainMapping[chain] = created
                    chainsBus.onNext(chain)
                } else {
+                    if (up is CachesEnabled) {
+                        up.setCaches(current.caches)
+                    }
                    current.addUpstream(up)
                }
                if (!callTargets.containsKey(chain)) {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamChange.kt
@@ -15,17 +15,25 @@
 */
 package io.emeraldpay.dshackle.upstream

+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.grpc.Chain

 class UpstreamChange(
        val chain: Chain,
        val upstream: Upstream,
        val type: ChangeType
-) {
+): CachesEnabled {
    enum class ChangeType {
        ADDED,
        REVALIDATED,
        STALE,
        REMOVED,
    }
+
+    override fun setCaches(caches: Caches) {
+        if (upstream is CachesEnabled) {
+            upstream.setCaches(caches)
+        }
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
@@ -17,18 +17,22 @@ package io.emeraldpay.dshackle.upstream.ethereum

 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.upstream.CallMethods
 import io.grpc.Status
 import io.grpc.StatusRuntimeException
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.hex.HexQuantity
 import io.infinitape.etherjar.rpc.*
 import io.infinitape.etherjar.rpc.json.ResponseJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
-import reactor.core.publisher.switchIfEmpty
-import java.time.Duration
+import java.math.BigInteger

 open class DirectEthereumApi(
        val rpcClient: ReactorRpcClient,
+        var caches: Caches?,
        private val objectMapper: ObjectMapper,
        val targets: CallMethods
 ): EthereumApi(objectMapper) {
@@ -90,8 +94,78 @@ open class DirectEthereumApi(
                }
    }

+    /**
+     * Actual request to the remote endpoint
+     */
    private fun callUpstream(method: String, params: List<Any>): Mono<out Any> {
-        return rpcClient.execute(RpcCall.create(method, Any::class.java, params))
+        return rpcClient.execute(callMapping(method, params))
                .timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout")))
+                .doOnNext { value ->
+                    caches?.cacheRequested(value)
+                }
+    }
+
+    /**
+     * Prepare RpcCall with data types specific for that particular requests. In general it may return a call that just
+     * parses JSON into Map. But the purpose of further processing and caching for some of the requests we want
+     * to have actual data types.
+     */
+    fun callMapping(method: String, params: List<Any>): RpcCall<out Any, out Any> {
+        return when {
+            method == "eth_getTransactionByHash" -> {
+                if (params.size != 1) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
+                }
+                val hash: TransactionId
+                try {
+                    hash = TransactionId.from(params[0].toString())
+                } catch (e: IllegalArgumentException) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
+                }
+                Commands.eth().getTransaction(hash)
+            }
+            method == "eth_getBlockByHash" -> {
+                if (params.size != 2) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
+                }
+                val hash: BlockHash
+                try {
+                    hash = BlockHash.from(params[0].toString())
+                } catch (e: IllegalArgumentException) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block hash")
+                }
+                val withTx = params[1].toString().toBoolean()
+                if (withTx) {
+                    Commands.eth().getBlockWithTransactions(hash)
+                } else {
+                    Commands.eth().getBlock(hash)
+                }
+            }
+            method == "eth_getBlockByNumber" -> {
+                if (params.size != 2) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
+                }
+                val number: Long
+                try {
+                    val quantity = HexQuantity.from(params[0].toString()) ?: throw IllegalArgumentException()
+                    number = quantity.value.let {
+                        if (it < BigInteger.valueOf(Long.MAX_VALUE) && it >= BigInteger.ZERO) {
+                            it.toLong()
+                        } else {
+                            throw IllegalArgumentException()
+                        }
+                    }
+                } catch (e: IllegalArgumentException) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number")
+                }
+                val withTx = params[1].toString().toBoolean()
+                if (withTx) {
+                    Commands.eth().getBlockWithTransactions(number)
+                } else {
+                    Commands.eth().getBlock(number)
+                }
+            }
+            else -> RpcCall.create(method, Any::class.java, params)
+        }
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
@@ -15,6 +15,8 @@
 */
 package io.emeraldpay.dshackle.upstream.ethereum

+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.grpc.Chain
@@ -32,7 +34,7 @@ open class EthereumUpstream(
        private val options: UpstreamsConfig.Options,
        val node: NodeDetailsList.NodeDetails,
        private val targets: CallMethods
-): DefaultUpstream(), Lifecycle {
+): DefaultUpstream(), CachesEnabled, Lifecycle {

    constructor(id: String, chain: Chain, api: DirectEthereumApi): this(id, chain, api, null,
            UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()),
@@ -48,6 +50,10 @@ open class EthereumUpstream(
        api.upstream = this
    }

+    override fun setCaches(caches: Caches) {
+        api.caches = caches;
+    }
+
    override fun getId(): String {
        return id
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt
@@ -21,6 +21,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead
@@ -53,10 +55,11 @@ open class GrpcUpstream(
        private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub,
        private val objectMapper: ObjectMapper,
        private val rpcClient: ReactorEmeraldClient
-): DefaultUpstream(), Lifecycle {
+): DefaultUpstream(), CachesEnabled, Lifecycle {

    private var allLabels: Collection<UpstreamsConfig.Labels> = ArrayList<UpstreamsConfig.Labels>()
    private val log = LoggerFactory.getLogger(GrpcUpstream::class.java)
+    private var caches: Caches? = null

    private val options = UpstreamsConfig.Options.getDefaults()
    private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
@@ -71,7 +74,7 @@ open class GrpcUpstream(
        val client = Selector.extractLabels(matcher)?.let { selector ->
            rpcClient.copyWithSelector(selector.asProto())
        } ?: rpcClient
-        return DirectEthereumApi(client, objectMapper, targets).let {
+        return DirectEthereumApi(client, caches, objectMapper, targets).let {
            it.upstream = this
            it
        }
@@ -205,4 +208,8 @@ open class GrpcUpstream(
        return options
    }

+    override fun setCaches(caches: Caches) {
+        this.caches = caches
+    }
+
 }