solution: cache latest blocks in memory

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

@@ -0,0 +1,30 @@
+package io.emeraldpay.dshackle.cache
+
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import reactor.core.publisher.Mono
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ConcurrentLinkedQueue
+
+class BlocksMemCache(
+        val maxSize: Int = 64
+) {
+
+    private val mapping = ConcurrentHashMap<BlockHash, BlockJson<TransactionId>>()
+    private val queue = ConcurrentLinkedQueue<BlockHash>()
+
+    fun get(hash: BlockHash): Mono<BlockJson<TransactionId>> {
+        return Mono.justOrEmpty(mapping[hash])
+    }
+
+    fun add(block: BlockJson<TransactionId>) {
+        mapping.put(block.hash, block)
+        queue.add(block.hash)
+
+        while (queue.size > maxSize) {
+            val old = queue.remove()
+            mapping.remove(old)
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt

@@ -0,0 +1,27 @@
+package io.emeraldpay.dshackle.reader
+
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.Commands
+import io.infinitape.etherjar.rpc.json.BlockJson
+import reactor.core.publisher.Mono
+import reactor.retry.Repeat
+import java.time.Duration
+
+class BlockApiReader(
+        val upstream: Upstream
+): Reader<BlockHash, BlockJson<TransactionId>> {
+
+    override fun read(key: BlockHash): Mono<BlockJson<TransactionId>> {
+        return Mono.just(key)
+                .flatMap {
+                    upstream.getApi(Selector.empty).executeAndConvert(Commands.eth().getBlock(it))
+                }.repeatWhenEmpty { n ->
+                    Repeat.times<Any>(3)
+                            .exponentialBackoff(Duration.ofMillis(100), Duration.ofMillis(500))
+                            .apply(n)
+                }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt

@@ -0,0 +1,16 @@
+package io.emeraldpay.dshackle.reader
+
+import io.emeraldpay.dshackle.cache.BlocksMemCache
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import reactor.core.publisher.Mono
+
+class BlockCacheReader(
+        val cache: BlocksMemCache
+): Reader<BlockHash, BlockJson<TransactionId>> {
+
+    override fun read(key: BlockHash): Mono<BlockJson<TransactionId>> {
+        return cache.get(key)
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt

@@ -0,0 +1,23 @@
+package io.emeraldpay.dshackle.reader
+
+import reactor.core.publisher.Mono
+
+class CompoundReader<K, D>(
+        private val readers: Collection<Reader<K, D>>
+): Reader<K, D> {
+
+    override fun read(key: K): Mono<D> {
+        if (readers.isEmpty()) {
+            return Mono.empty()
+        }
+        var result = readers.first().read(key)
+        if (readers.size == 1) {
+            return result
+        }
+        readers.stream().skip(1).forEach {
+            result = result.switchIfEmpty(it.read(key))
+        }
+        return result
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/reader/EmptyReader.kt

@@ -0,0 +1,10 @@
+package io.emeraldpay.dshackle.reader
+
+import reactor.core.publisher.Mono
+
+class EmptyReader<K, D>: Reader<K, D> {
+
+    override fun read(key: K): Mono<D> {
+        return Mono.empty()
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/reader/Reader.kt

@@ -0,0 +1,9 @@
+package io.emeraldpay.dshackle.reader
+
+import reactor.core.publisher.Mono
+
+interface Reader<K, D> {
+
+    fun read(key: K): Mono<D>
+
+}

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

@@ -31,16 +31,16 @@ class NativeCall(
         return requestMono.flatMapMany(this::prepareCall)
             .map(this::setupCallParams)
             .parallel()
-            .flatMap(this::executeOnRemote)
+            .flatMap(this::fetch)
             .sequential()
             .map(this::buildResponse)
             .doOnError { e -> log.warn("Error during native call", e) }
             .onErrorResume(this::processException)
     }
 
-    fun setupCallParams(it: CallContext<Tuple2<String, String>>): CallContext<Tuple2<String, List<Any>>> {
+    fun setupCallParams(it: CallContext<RawCallDetails>): CallContext<ParsedCallDetails> {
-        val params = extractParams(it.payload.t2)
+        val params = extractParams(it.payload.params)
-        return it.withPayload(Tuples.of(it.payload.t1, params))
+        return it.withPayload(ParsedCallDetails(it.payload.method, params))
     }
 
     fun buildResponse(it: CallContext<ByteArray>): BlockchainOuterClass.NativeCallReplyItem {
@@ -65,46 +65,57 @@ class NativeCall(
                 .toMono()
     }
 
-    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux<CallContext<Tuple2<String, String>>> {
+    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux<CallContext<RawCallDetails>> {
         val chain = Chain.byId(request.chain.number)
         if (chain == Chain.UNSPECIFIED) {
-            return Flux.error<CallContext<Tuple2<String, String>>>(CallFailure(0, Exception("Invalid chain id: ${request.chain.number}")))
+            return Flux.error(CallFailure(0, Exception("Invalid chain id: ${request.chain.number}")))
         }
         val upstream = upstreams.getUpstream(chain)
-                ?: return Flux.error<CallContext<Tuple2<String, String>>>(CallFailure(0, Exception("Chain ${chain.id} is unavailable")))
+                ?: return Flux.error(CallFailure(0, Exception("Chain ${chain.id} is unavailable")))
 
         return prepareCall(request, upstream)
     }
 
-    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux<CallContext<Tuple2<String, String>>> {
+    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux<CallContext<RawCallDetails>> {
         val matcher = Selector.convertToMatcher(request.selector)
-        val apis = upstream.getApis(matcher)
         return request.itemsList.toFlux().map {
             val method = it.method
             val params = it.payload.toStringUtf8()
             val callQuorum = upstream.targets?.getQuorumFor(method) ?: AlwaysQuorum()
             callQuorum.init(upstream.getHead())
 
-            CallContext(it.id, apis, callQuorum, Tuples.of(method, params))
+            CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params))
         }
     }
 
-    fun executeOnRemote(ctx: CallContext<Tuple2<String, List<Any>>>): Mono<CallContext<ByteArray>> {
+    fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
+        return fetchFromCache(ctx)
+                .switchIfEmpty(
+                        Mono.just(ctx).flatMap(this::executeOnRemote)
+                )
+    }
+
+    fun fetchFromCache(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
+        val cachingApi = ctx.upstream.cache
+        return cachingApi.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { ctx.withPayload(it) }
+    }
+
+    fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
         val p: Predicate<Any> = CallQuorum.untilResolved(ctx.callQuorum)
-        val all = ctx.apis.toFlux().share()
+        val all = ctx.getApis().toFlux().share()
         //execute on the first API immediately, and then make a delay between each call to not dos upstreams
         val immediate = Flux.from(all).take(1)
         val retries = Flux.from(all).delayElements(Duration.ofMillis(200))
         return Flux.concat(immediate, retries)
                 .takeWhile(p)
                 .flatMap { api ->
-                    api.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { Tuples.of(it, api.upstream!!) }
+                    api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) }
                 }
                 .reduce(ctx.callQuorum, CallQuorum.asReducer())
                 .filter { it.isResolved() }
                 .map {
                     val result  = it.getResult()
-                            ?: throw CallFailure(ctx.id, Exception("No response from upstream for ${ctx.payload.t1}"))
+                            ?: throw CallFailure(ctx.id, Exception("No response from upstream for ${ctx.payload.method}"))
                     ctx.withPayload(result)
                 }
                 .onErrorMap {
@@ -113,7 +124,7 @@ class NativeCall(
                     else CallFailure(ctx.id, it)
                 }
                 .switchIfEmpty(
-                        Mono.error<CallContext<ByteArray>>(CallFailure(ctx.id, Exception("No response or no available upstream for ${ctx.payload.t1}")))
+                        Mono.error<CallContext<ByteArray>>(CallFailure(ctx.id, Exception("No response or no available upstream for ${ctx.payload.method}")))
                 )
     }
 
@@ -125,11 +136,18 @@ class NativeCall(
         return req as List<Any>
     }
 
-    open class CallContext<T>(val id: Int, val apis: Iterator<EthereumApi>, val callQuorum: CallQuorum, val payload: T) {
+    open class CallContext<T>(val id: Int, val upstream: AggregatedUpstream, val matcher: Selector.Matcher, val callQuorum: CallQuorum, val payload: T) {
         fun <X> withPayload(payload: X): CallContext<X> {
-            return CallContext(id, apis, callQuorum, payload)
+            return CallContext(id, upstream, matcher, callQuorum, payload)
+        }
+
+        fun getApis(): Iterator<DirectEthereumApi> {
+            return upstream.getApis(matcher)
         }
     }
 
     open class CallFailure(val id: Int, val reason: Throwable): Exception("Failed to call $id: ${reason.message}")
+
+    class RawCallDetails(val method: String, val params: String)
+    class ParsedCallDetails(val method: String, val params: List<Any>)
 }

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

@@ -1,19 +1,40 @@
 package io.emeraldpay.dshackle.upstream
 
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.cache.BlocksMemCache
 import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.reader.BlockCacheReader
+import io.emeraldpay.dshackle.reader.CompoundReader
+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 org.springframework.context.Lifecycle
+import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.locks.ReentrantLock
 import java.util.function.Predicate
+import kotlin.concurrent.withLock
 
 abstract class AggregatedUpstream(
-        val targets: CallMethods
+        val targets: CallMethods,
-): Upstream {
+        val objectMapper: ObjectMapper
+): Upstream, Lifecycle {
+
+    private val blocksCache = BlocksMemCache()
+    private var cacheSubscription: Disposable? = null
+    private val blockReader: Reader<BlockHash, BlockJson<TransactionId>> = CompoundReader(
+            listOf(BlockCacheReader(blocksCache))
+    )
+    var cache: CachingEthereumApi = CachingEthereumApi.empty()
+    private val reconfigLock = ReentrantLock()
 
     abstract fun getAll(): List<Upstream>
     abstract fun addUpstream(upstream: Upstream)
-    abstract fun getApis(matcher: Selector.Matcher): Iterator<EthereumApi>
+    abstract fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi>
 
     override fun observeStatus(): Flux<UpstreamAvailability> {
         val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } }
@@ -61,4 +82,23 @@ abstract class AggregatedUpstream(
             return changed
         }
     }
+
+    override fun start() {
+    }
+
+    override fun stop() {
+        cacheSubscription?.dispose()
+        cacheSubscription = null
+    }
+
+    fun onHeadUpdated(head: EthereumHead) {
+        reconfigLock.withLock {
+            cacheSubscription?.dispose()
+            cacheSubscription = head.getFlux().subscribe {
+                blocksCache.add(it)
+            }
+            cache = CachingEthereumApi(objectMapper, blockReader, head)
+        }
+    }
+
 }

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

@@ -0,0 +1,48 @@
+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.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 reactor.core.publisher.Mono
+import java.util.function.Function
+
+open class CachingEthereumApi(
+        private val objectMapper: ObjectMapper,
+        private val cache: Reader<BlockHash, BlockJson<TransactionId>>,
+        private val head: EthereumHead
+): EthereumApi(objectMapper) {
+
+    companion object {
+        @JvmStatic
+        fun empty(): CachingEthereumApi {
+            return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyEthereumHead())
+        }
+    }
+
+    override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
+        return when (method) {
+            "eth_blockNumber" -> head.getFlux().next()
+                    .map { HexQuantity.from(it.number).toHex() }
+                    .map(toJson(id))
+            "eth_getBlockByHash" -> Mono.just(params[0])
+                    .map { BlockHash.from(it as String) }
+                    .flatMap(cache::read)
+                    .map(toJson(id))
+            else -> Mono.empty()
+        }
+    }
+
+    fun toJson(id: Int): Function<Any, ByteArray> {
+        return Function { data ->
+            val resp = ResponseJson<Any, Int>()
+            resp.id = id
+            resp.result = data
+            objectMapper.writer().writeValueAsBytes(resp)
+        }
+    }
+}

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

@@ -1,19 +1,19 @@
 package io.emeraldpay.dshackle.upstream
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
-import java.io.Closeable
 import java.lang.IllegalStateException
 import java.time.Duration
 
-class ChainUpstreams (
+open class ChainUpstreams (
         val chain: Chain,
         private val upstreams: MutableList<Upstream>,
-        targets: CallMethods
+        targets: CallMethods,
-) : AggregatedUpstream(targets), Lifecycle {
+        objectMapper: ObjectMapper
-
+) : AggregatedUpstream(targets, objectMapper), Lifecycle {
 
     private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
     private var seq = 0
@@ -30,12 +30,14 @@ class ChainUpstreams (
     }
 
     override fun start() {
+        super.start()
         subscription = observeStatus()
                 .distinctUntilChanged()
                 .subscribe { printStatus() }
     }
 
     override fun stop() {
+        super.stop()
         subscription?.dispose()
         subscription = null
         head?.let {
@@ -54,7 +56,7 @@ class ChainUpstreams (
         }
         lagObserver?.stop()
         lagObserver = null
-        return if (upstreams.size == 1) {
+        val head = if (upstreams.size == 1) {
             val upstream = upstreams.first()
             upstream.setLag(0)
             upstream.getHead()
@@ -68,6 +70,8 @@ class ChainUpstreams (
             this.lagObserver = lagObserver
             newHead
         }
+        onHeadUpdated(head)
+        return head
     }
 
     override fun getAll(): List<Upstream> {
@@ -79,7 +83,7 @@ class ChainUpstreams (
         head = updateHead()
     }
 
-    override fun getApis(matcher: Selector.Matcher): Iterator<EthereumApi> {
+    override fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi> {
         val i = seq++
         if (seq >= Int.MAX_VALUE / 2) {
             seq = 0
@@ -87,7 +91,7 @@ class ChainUpstreams (
         return FilteringApiIterator(upstreams, i, matcher)
     }
 
-    override fun getApi(matcher: Selector.Matcher): EthereumApi {
+    override fun getApi(matcher: Selector.Matcher): DirectEthereumApi {
         return getApis(matcher).next()
     }
 

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

@@ -104,7 +104,7 @@ open class ConfiguredUpstreams(
                                       chain: Chain,
                                       options: UpstreamsConfig.Options,
                                       labels: UpstreamsConfig.Labels) {
-        var rpcApi: EthereumApi? = null
+        var rpcApi: DirectEthereumApi? = null
         val urls = ArrayList<URI>()
         up.rpc?.let { endpoint ->
             val rpcTransport = DefaultRpcTransport(endpoint.url)
@@ -117,10 +117,9 @@ open class ConfiguredUpstreams(
                 }
             }
             val rpcClient = DefaultRpcClient(rpcTransport)
-            rpcApi = EthereumApi(
+            rpcApi = DirectEthereumApi(
                     rpcClient,
                     objectMapper,
-                    chain,
                     targetFor(chain)
             )
             urls.add(endpoint.url)
@@ -171,7 +170,7 @@ open class ConfiguredUpstreams(
     override fun addUpstream(chain: Chain, up: Upstream): ChainUpstreams {
         val current = chainMapping[chain]
         if (current == null) {
-            val created = ChainUpstreams(chain, ArrayList<Upstream>(), targetFor(chain))
+            val created = ChainUpstreams(chain, ArrayList<Upstream>(), targetFor(chain), objectMapper)
             created.addUpstream(up)
             created.start()
             chainMapping[chain] = created

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

@@ -0,0 +1,58 @@
+package io.emeraldpay.dshackle.upstream
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.infinitape.etherjar.rpc.RpcCall
+import io.infinitape.etherjar.rpc.RpcClient
+import io.infinitape.etherjar.rpc.RpcException
+import io.infinitape.etherjar.rpc.json.ResponseJson
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import java.time.Duration
+
+open class DirectEthereumApi(
+        val rpcClient: RpcClient,
+        private val objectMapper: ObjectMapper,
+        val targets: CallMethods
+): EthereumApi(objectMapper) {
+
+    private val timeout = Duration.ofSeconds(5)
+    private val log = LoggerFactory.getLogger(EthereumApi::class.java)
+
+    override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
+        val result: Mono<out Any> = when {
+            targets.isHardcoded(method) -> Mono.just(method).map { targets.hardcoded(it) }
+            targets.isAllowed(method)   -> callUpstream(method, params)
+            else -> Mono.error(RpcException(-32601, "Method not allowed or not found"))
+        }
+        return result
+                .doOnError { t ->
+                    log.warn("Upstream error: ${t.message} for ${method}")
+                }
+                .map {
+                    val resp = ResponseJson<Any, Int>()
+                    resp.id = id
+                    resp.result = it
+                    objectMapper.writer().writeValueAsBytes(resp)
+                }
+                .onErrorMap { t ->
+                    if (RpcException::class.java.isAssignableFrom(t.javaClass)) {
+                        t
+                    } else {
+                        log.warn("Convert to RPC error. Exception: ${t.message}")
+                        RpcException(-32020, "Error reading from upstream", null, t)
+                    }
+                }
+                .onErrorResume(RpcException::class.java) { t ->
+                    val resp = ResponseJson<Any, Int>()
+                    resp.id = id
+                    resp.error = t.error
+                    Mono.just(objectMapper.writer().writeValueAsBytes(resp))
+                }
+    }
+
+    private fun callUpstream(method: String, params: List<Any>): Mono<out Any> {
+        return Mono.fromCompletionStage(
+                rpcClient.execute(RpcCall.create(method, Any::class.java, params))
+        ).timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout")))
+    }
+}

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

@@ -0,0 +1,12 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import reactor.core.publisher.Flux
+
+class EmptyEthereumHead : EthereumHead {
+
+    override fun getFlux(): Flux<BlockJson<TransactionId>> {
+        return Flux.empty()
+    }
+}

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

@@ -7,28 +7,22 @@ import io.infinitape.etherjar.rpc.*
 import io.infinitape.etherjar.rpc.json.ResponseJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
+import java.io.InputStream
 import java.time.Duration
 
-open class EthereumApi(
+abstract class EthereumApi(
-        val rpcClient: RpcClient,
+        objectMapper: ObjectMapper
-        private val objectMapper: ObjectMapper,
-        private val chain: Chain,
-        val targets: CallMethods
 ) {
 
     private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
     var upstream: Upstream? = null
 
-    private val timeout = Duration.ofSeconds(5)
+    abstract fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray>
-    private val log = LoggerFactory.getLogger(EthereumApi::class.java)
-    var ws: EthereumWs? = null
-        set(value) {
-            field = value
-        }
 
-    open fun <JS, RS> executeAndConvert(rpcCall: RpcCall<JS, RS>): Mono<RS> {
+    fun <JS, RS> executeAndConvert(rpcCall: RpcCall<JS, RS>): Mono<RS> {
         val convertToJS = java.util.function.Function<ByteArray, Mono<JS>> { resp ->
-            val jsonValue: JS? = jacksonRpcConverter.fromJson(resp.inputStream(), rpcCall.jsonType, Int::class.java)
+            val inputStream: InputStream = resp.inputStream()
+            val jsonValue: JS? = jacksonRpcConverter.fromJson(inputStream, rpcCall.jsonType, Int::class.java)
             if (jsonValue == null) Mono.empty<JS>()
             else Mono.just(jsonValue)
         }
@@ -36,51 +30,4 @@ open class EthereumApi(
                 .flatMap(convertToJS)
                 .map(rpcCall.converter::apply)
     }
-
-    open fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
-        val result: Mono<out Any> = when {
-            targets.isHardcoded(method) -> Mono.just(method).map { targets.hardcoded(it) }
-            targets.isAllowed(method)   -> callUpstream(method, params)
-            else -> Mono.error(RpcException(-32601, "Method not allowed or not found"))
-        }
-        return result
-                .doOnError { t ->
-                    log.warn("Upstream error: ${t.message} for ${method} on $chain")
-                }
-                .map {
-                    val resp = ResponseJson<Any, Int>()
-                    resp.id = id
-                    resp.result = it
-                    objectMapper.writer().writeValueAsBytes(resp)
-                }
-                .onErrorMap { t ->
-                    if (RpcException::class.java.isAssignableFrom(t.javaClass)) {
-                        t
-                    } else {
-                        log.warn("Convert to RPC error. Exception: ${t.message}")
-                        RpcException(-32020, "Error reading from upstream", null, t)
-                    }
-                }
-                .onErrorResume(RpcException::class.java) { t ->
-                    val resp = ResponseJson<Any, Int>()
-                    resp.id = id
-                    resp.error = t.error
-                    Mono.just(objectMapper.writer().writeValueAsBytes(resp))
-                }
-    }
-
-    private fun callUpstream(method: String, params: List<Any>): Mono<out Any> {
-        if (method == "eth_blockNumber") {
-            val current = upstream?.getHead()?.getFlux()?.next()?.let { head ->
-                head.map { HexQuantity.from(it.number).toHex() }
-            }
-            if (current != null) {
-                return current
-            }
-        }
-        return Mono.fromCompletionStage(
-                rpcClient.execute(RpcCall.create(method, Any::class.java, params))
-        ).timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout")))
-    }
-
 }

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

@@ -14,7 +14,7 @@ import java.time.Duration
 import java.util.concurrent.atomic.AtomicReference
 
 class EthereumRpcHead(
-    private val api: EthereumApi
+    private val api: DirectEthereumApi
 ): EthereumHead, Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java)

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

@@ -8,14 +8,14 @@ import reactor.core.Disposable
 
 open class EthereumUpstream(
         val chain: Chain,
-        private val api: EthereumApi,
+        private val api: DirectEthereumApi,
         private val ethereumWs: EthereumWs? = null,
         private val options: UpstreamsConfig.Options,
         val node: NodeDetailsList.NodeDetails,
         private val targets: CallMethods
 ): DefaultUpstream(), Lifecycle {
 
-    constructor(chain: Chain, api: EthereumApi): this(chain, api, null,
+    constructor(chain: Chain, api: DirectEthereumApi): this(chain, api, null,
             UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()),
             DirectCallMethods())
 
@@ -84,11 +84,11 @@ open class EthereumUpstream(
         return head
     }
 
-    override fun getApi(matcher: Selector.Matcher): EthereumApi {
+    override fun getApi(matcher: Selector.Matcher): DirectEthereumApi {
         return api
     }
 
-    fun getApi(): EthereumApi {
+    fun getApi(): DirectEthereumApi {
         return api
     }
 

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

@@ -5,7 +5,7 @@ class FilteringApiIterator(
         private var pos: Int,
         private val matcher: Selector.Matcher,
         private val repeatLimit: Int = 3
-): Iterator<EthereumApi> {
+): Iterator<DirectEthereumApi> {
 
     private var nextUpstream: Upstream? = null
     private var consumed = 0
@@ -31,7 +31,7 @@ class FilteringApiIterator(
         return nextInternal()
     }
 
-    override fun next(): EthereumApi {
+    override fun next(): DirectEthereumApi {
         if (nextInternal()) {
             val curr = nextUpstream!!
             nextUpstream = null

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

@@ -44,9 +44,9 @@ open class GrpcUpstream(
 
     private var headSubscription: Disposable? = null
 
-    open fun createApi(matcher: Selector.Matcher): EthereumApi {
+    open fun createApi(matcher: Selector.Matcher): DirectEthereumApi {
         val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher))
-        return EthereumApi(rpcClient, objectMapper, chain, targets).let {
+        return DirectEthereumApi(rpcClient, objectMapper, targets).let {
             it.upstream = this
             it
         }
@@ -156,7 +156,7 @@ open class GrpcUpstream(
         return head
     }
 
-    override fun getApi(matcher: Selector.Matcher): EthereumApi {
+    override fun getApi(matcher: Selector.Matcher): DirectEthereumApi {
         return createApi(matcher)
     }
 

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

@@ -8,7 +8,8 @@ interface Upstream {
     fun getStatus(): UpstreamAvailability
     fun observeStatus(): Flux<UpstreamAvailability>
     fun getHead(): EthereumHead
-    fun getApi(matcher: Selector.Matcher): EthereumApi
+    fun getApi(matcher: Selector.Matcher): DirectEthereumApi
+//    fun getCache(): CachingEthereumApi
     fun getOptions(): UpstreamsConfig.Options
     fun getSupportedTargets(): Set<String>
     fun setLag(lag: Long)

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

@@ -0,0 +1,52 @@
+package io.emeraldpay.dshackle.cache
+
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import spock.lang.Specification
+
+class BlocksMemCacheSpec extends Specification {
+
+    String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
+    String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
+    String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
+    String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
+
+    def "Add and read"() {
+        setup:
+        def cache = new BlocksMemCache()
+        def block = new BlockJson<TransactionId>()
+        block.number = 100
+        block.hash = BlockHash.from(hash1)
+
+        when:
+        cache.add(block)
+        def act = cache.get(BlockHash.from(hash1)).block()
+        then:
+        act == block
+    }
+
+    def "Keeps only configured amount"() {
+        setup:
+        def cache = new BlocksMemCache(3)
+        [hash1]
+
+        when:
+        [hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i ->
+            def block = new BlockJson<TransactionId>()
+            block.number = 100 + i
+            block.hash = BlockHash.from(hash)
+            cache.add(block)
+        }
+
+        def act1 = cache.get(BlockHash.from(hash1)).block()
+        def act2 = cache.get(BlockHash.from(hash2)).block()
+        def act3 = cache.get(BlockHash.from(hash3)).block()
+        def act4 = cache.get(BlockHash.from(hash4)).block()
+        then:
+        act2.hash.toHex() == hash2
+        act3.hash.toHex() == hash3
+        act4.hash.toHex() == hash4
+        act1 == null
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy

@@ -1,17 +1,23 @@
 package io.emeraldpay.dshackle.rpc
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.test.EthereumApiMock
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.AggregatedUpstream
 import io.emeraldpay.dshackle.upstream.AlwaysQuorum
+import io.emeraldpay.dshackle.upstream.CachingEthereumApi
 import io.emeraldpay.dshackle.upstream.CallQuorum
+import io.emeraldpay.dshackle.upstream.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.EthereumApi
 import io.emeraldpay.dshackle.upstream.NonEmptyQuorum
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.rpc.RpcClient
+import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
 import reactor.util.function.Tuples
 import spock.lang.Specification
@@ -33,7 +39,9 @@ class NativeCallSpec extends Specification {
         apiMock.answer("eth_test", [], "foo")
 
         def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
-        def call = new NativeCall.CallContext(1, [apiMock].multiply(59).iterator(), quorum, Tuples.of("eth_test", []))
+        def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock),
+                Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_test", []))
 
         when:
         def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2))
@@ -59,7 +67,9 @@ class NativeCallSpec extends Specification {
         apiMock.answerOnce("eth_test", [], null)
 
         def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
-        def call = new NativeCall.CallContext(1, [apiMock].multiply(5).iterator(), quorum, Tuples.of("eth_test", []))
+        def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock),
+                Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_test", []))
 
 
         when:
@@ -85,7 +95,8 @@ class NativeCallSpec extends Specification {
         apiMock.answerOnce("eth_test", [], "foo")
 
         def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
-        def call = new NativeCall.CallContext(1, [apiMock].multiply(5).iterator(), quorum, Tuples.of("eth_test", []))
+        def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_test", []))
 
         (4..5) * quorum.isResolved()
         3 * quorum.record(_, _)
@@ -135,9 +146,10 @@ class NativeCallSpec extends Specification {
         def upstreams = Stub(Upstreams)
         def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
         def json = [jsonrpc:"2.0", id:1, result: "foo"]
+
         when:
         def resp = nativeCall.buildResponse(
-                new NativeCall.CallContext<byte[]>(1561, [].iterator(), new AlwaysQuorum(), objectMapper.writeValueAsBytes(json))
+                new NativeCall.CallContext<byte[]>(1561, TestingCommons.aggregatedUpstream(Stub(DirectEthereumApi)), Selector.empty, new AlwaysQuorum(), objectMapper.writeValueAsBytes(json))
         )
         then:
         resp.id == 1561
@@ -191,4 +203,42 @@ class NativeCallSpec extends Specification {
 //                .expectComplete()
                 .verify(Duration.ofSeconds(1))
     }
+
+    def "Calls cache before remote"() {
+        setup:
+        def upstreams = Stub(Upstreams)
+        def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
+        def api = Mock(DirectEthereumApi)
+        def upstream = TestingCommons.aggregatedUpstream(api)
+        def cacheMock = Mock(CachingEthereumApi)
+        upstream.cache = cacheMock
+
+        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10,
+                upstream,
+                Selector.empty, new AlwaysQuorum(),
+                new NativeCall.ParsedCallDetails("eth_test", []))
+        when:
+        nativeCall.fetch(ctx)
+        then:
+        1 * cacheMock.execute(10, "eth_test", []) >> Mono.empty()
+    }
+
+    def "Uses cached value"() {
+        setup:
+        def upstreams = Stub(Upstreams)
+        def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
+        def upstream = TestingCommons.aggregatedUpstream(Stub(DirectEthereumApi))
+        def cacheMock = Mock(CachingEthereumApi)
+        upstream.cache = cacheMock
+
+        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10,
+                upstream,
+                Selector.empty, new AlwaysQuorum(),
+                new NativeCall.ParsedCallDetails("eth_test", []))
+        when:
+        def act = nativeCall.fetch(ctx)
+        then:
+        1 * cacheMock.execute(10, "eth_test", []) >> Mono.just('{"result": "foo"}'.bytes)
+        new String(act.block().payload) == '{"result": "foo"}'
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy

@@ -5,6 +5,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.test.EthereumUpstreamMock
 import io.emeraldpay.dshackle.test.UpstreamsMock
+import io.emeraldpay.dshackle.upstream.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.EthereumApi
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.grpc.Chain
@@ -57,7 +58,7 @@ class StreamHeadSpec extends Specification {
                 .build()
         }
 
-        def upstream = new EthereumUpstreamMock(Chain.ETHEREUM, Mock(EthereumApi))
+        def upstream = new EthereumUpstreamMock(Chain.ETHEREUM, Stub(DirectEthereumApi.class))
         def upstreams = new UpstreamsMock(Chain.ETHEREUM, upstream)
         def streamHead = new StreamHead(upstreams)
         when:

src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.upstream.DirectCallMethods
+import io.emeraldpay.dshackle.upstream.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.EthereumApi
 import io.emeraldpay.dshackle.upstream.QuorumBasedMethods
 import io.emeraldpay.dshackle.upstream.Upstream
@@ -17,14 +18,14 @@ import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 
-class EthereumApiMock extends EthereumApi {
+class EthereumApiMock extends DirectEthereumApi {
 
     private static final Logger log = LoggerFactory.getLogger(this)
     List<PredefinedResponse> predefined = []
     private ObjectMapper objectMapper
 
     EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) {
-        super(rpcClient, objectMapper, chain, new DirectCallMethods())
+        super(rpcClient, objectMapper, new DirectCallMethods())
         this.objectMapper = objectMapper
     }
 

src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy

@@ -1,5 +1,6 @@
 package io.emeraldpay.dshackle.test
 
+import io.emeraldpay.dshackle.upstream.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.EthereumApi
 import io.emeraldpay.dshackle.upstream.EthereumHead
 import io.emeraldpay.dshackle.upstream.EthereumUpstream
@@ -13,7 +14,7 @@ class EthereumUpstreamMock extends EthereumUpstream {
 
     EthereumHeadMock ethereumHeadMock = new EthereumHeadMock()
 
-    EthereumUpstreamMock(@NotNull Chain chain, @NotNull EthereumApi api) {
+    EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api) {
         super(chain, api)
         setLag(0)
         setStatus(UpstreamAvailability.OK)

src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy

@@ -4,6 +4,10 @@ import com.fasterxml.jackson.core.Version
 import com.fasterxml.jackson.databind.DeserializationFeature
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.module.SimpleModule
+import io.emeraldpay.dshackle.upstream.AggregatedUpstream
+import io.emeraldpay.dshackle.upstream.ChainUpstreams
+import io.emeraldpay.dshackle.upstream.DirectCallMethods
+import io.emeraldpay.dshackle.upstream.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.EthereumApi
 import io.emeraldpay.dshackle.upstream.EthereumUpstream
 import io.emeraldpay.dshackle.upstream.Upstream
@@ -42,7 +46,11 @@ class TestingCommons {
         return new JacksonRpcConverter(objectMapper())
     }
 
-    static EthereumUpstreamMock upstream(EthereumApi api) {
+    static EthereumUpstreamMock upstream(DirectEthereumApi api) {
         return new EthereumUpstreamMock(Chain.ETHEREUM, api)
     }
+
+    static AggregatedUpstream aggregatedUpstream(DirectEthereumApi api) {
+        return new ChainUpstreams(Chain.ETHEREUM, [upstream(api)], new DirectCallMethods(), objectMapper())
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/test/UpstreamsMock.groovy

@@ -26,7 +26,7 @@ class UpstreamsMock implements Upstreams {
     @Override
     AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) {
         if (!upstreams.containsKey(chain)) {
-            upstreams[chain] = new ChainUpstreams(chain, [up], targetFor(chain))
+            upstreams[chain] = new ChainUpstreams(chain, [up], targetFor(chain), TestingCommons.objectMapper())
         } else {
             upstreams[chain].addUpstream(up)
         }

src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy

@@ -23,11 +23,13 @@ class EthereumGrpcTransportSpec extends Specification {
 
     def "Make simple call"() {
         setup:
+        def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)
+
         def callData = [:]
         def otherSideUpstreams = Mock(Upstreams)
-        def otherSideAggr = Mock(AggregatedUpstream)
+        def otherSideAggr = TestingCommons.aggregatedUpstream(otherSideApi)
+
         def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
-        def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)
         otherSideApi.upstream = otherSideAggr
 
         def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() {
@@ -47,9 +49,6 @@ class EthereumGrpcTransportSpec extends Specification {
 
         then:
         1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr
-        1 * otherSideAggr.getApis(_) >> [otherSideApi].iterator()
-        _ * otherSideAggr.getHead() >> Stub(EthereumHead)
-        _ * otherSideAggr.getTargets() >> ethereumTargets
         status.failed == 0
         status.succeed == 1
         status.total == 1
@@ -67,11 +66,12 @@ class EthereumGrpcTransportSpec extends Specification {
 
     def "Make few calls"() {
         setup:
+        def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)
+
         def callData = [:]
         def otherSideUpstreams = Mock(Upstreams)
-        def otherSideAggr = Mock(AggregatedUpstream)
+        def otherSideAggr = TestingCommons.aggregatedUpstream(otherSideApi)
         def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
-        def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM)
         otherSideApi.upstream = otherSideAggr
 
         def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() {
@@ -95,9 +95,6 @@ class EthereumGrpcTransportSpec extends Specification {
 
         then:
         1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr
-        1 * otherSideAggr.getApis(_) >> [otherSideApi].multiply(34).iterator()
-        _ * otherSideAggr.getHead() >> Stub(EthereumHead)
-        _ * otherSideAggr.getTargets() >> ethereumTargets
         status.failed == 0
         status.succeed == 2
         status.total == 2

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy

@@ -23,7 +23,7 @@ class FilteringApiIteratorSpec extends Specification {
         ].collect {
             new EthereumUpstream(
                     Chain.ETHEREUM,
-                    new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM, ethereumTargets),
+                    new DirectEthereumApi(rpcClient, objectMapper, ethereumTargets),
                     (EthereumWs) null,
                     new UpstreamsConfig.Options(),
                     new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap(it)),