solution: use caching for WS head

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

@@ -89,7 +89,7 @@ open class ChainUpstreams (
             upstream.setLag(0)
             upstream.getHead()
         } else {
-            val newHead = EthereumHeadMerge(upstreams.map { it.getHead().getFlux() }).apply {
+            val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }).apply {
                 this.start()
             }
             val lagObserver = HeadLagObserver(newHead, upstreams).apply {

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

@@ -35,6 +35,10 @@ abstract class EthereumApi(
 
     abstract fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray>
 
+    fun <JS, RS> execute(rpcCall: RpcCall<JS, RS>): Mono<ByteArray> {
+        return execute(0, rpcCall.method, rpcCall.params as List<Any>)
+    }
+
     fun <JS, RS> executeAndConvert(rpcCall: RpcCall<JS, RS>): Mono<RS> {
         val convertToJS = java.util.function.Function<ByteArray, Mono<JS>> { resp ->
             val inputStream: InputStream = resp.inputStream()
@@ -42,7 +46,7 @@ abstract class EthereumApi(
             if (jsonValue == null) Mono.empty<JS>()
             else Mono.just(jsonValue)
         }
-        return execute(0, rpcCall.method, rpcCall.params as List<Any>)
+        return execute(rpcCall)
                 .flatMap(convertToJS)
                 .map(rpcCall.converter::apply)
                 .doOnError { err -> log.debug("Failed to read from upstream", err) }

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

@@ -15,20 +15,15 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.infinitape.etherjar.domain.TransactionId
+import io.emeraldpay.dshackle.cache.Caches
-import io.infinitape.etherjar.rpc.json.BlockJson
+import io.emeraldpay.dshackle.cache.CachesEnabled
-import io.infinitape.etherjar.rpc.json.TransactionRefJson
-import org.reactivestreams.Publisher
-import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import java.util.concurrent.atomic.AtomicReference
 
 class EthereumHeadMerge(
-        private val fluxes: Iterable<Publisher<BlockJson<TransactionRefJson>>>
+        private val sources: Iterable<EthereumHead>
-): DefaultEthereumHead(), Lifecycle {
+): DefaultEthereumHead(), Lifecycle, CachesEnabled {
 
     private var subscription: Disposable? = null
 
@@ -37,11 +32,19 @@ class EthereumHeadMerge(
     }
 
     override fun start() {
-        subscription = super.follow(Flux.merge(fluxes))
+        subscription = super.follow(Flux.merge(sources.map { it.getFlux() }))
     }
 
     override fun stop() {
         subscription?.dispose()
     }
 
+    override fun setCaches(caches: Caches) {
+        sources.forEach {
+            if (it is CachesEnabled) {
+                it.setCaches(caches)
+            }
+        }
+    }
+
 }

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

@@ -16,9 +16,17 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
+import io.emeraldpay.dshackle.reader.EmptyReader
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.CachingEthereumApi
+import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.rpc.Batch
 import io.infinitape.etherjar.rpc.Commands
 import io.infinitape.etherjar.rpc.ReactorBatch
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory
@@ -52,6 +60,8 @@ class EthereumRpcHead(
                             .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
                 }
                 .flatMap {
+                    //fetching by Block Height here, critical to use same upstream,
+                    //different upstreams may have different blocks on the same height
                     api.rpcClient
                             .execute(Commands.eth().getBlock(it))
                             .subscribeOn(scheduler)

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

@@ -52,6 +52,9 @@ open class EthereumUpstream(
 
     override fun setCaches(caches: Caches) {
         api.caches = caches;
+        if (head is CachesEnabled) {
+            head.setCaches(caches)
+        }
     }
 
     override fun getId(): String {
@@ -92,7 +95,7 @@ open class EthereumUpstream(
             val rpc = EthereumRpcHead(api, Duration.ofSeconds(30)).apply {
                 this.start()
             }
-            EthereumHeadMerge(listOf(rpc.getFlux(), ws.getFlux())).apply {
+            EthereumHeadMerge(listOf(rpc, ws)).apply {
                 this.start()
             }
         } else {

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

@@ -16,8 +16,12 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 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.infinitape.etherjar.domain.TransactionId
+import io.emeraldpay.dshackle.reader.EmptyReader
+import io.emeraldpay.dshackle.reader.Reader
+import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.rpc.Commands
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
@@ -34,7 +38,7 @@ class EthereumWs(
         private val uri: URI,
         private val origin: URI,
         private val api: EthereumApi
-) {
+): CachesEnabled {
 
     private val log = LoggerFactory.getLogger(EthereumWs::class.java)
     private val topic = TopicProcessor
@@ -43,6 +47,8 @@ class EthereumWs(
             .build()
     var basicAuth: UpstreamsConfig.BasicAuth? = null
 
+    private var blockCache: Reader<BlockHash, BlockJson<TransactionRefJson>> = EmptyReader()
+
     fun connect() {
         log.info("Connecting to WebSocket: $uri")
         val clientBuilder = WebsocketClient.newBuilder()
@@ -54,24 +60,27 @@ class EthereumWs(
         val client = clientBuilder.build()
         try {
             client.connect()
+            client.onNewBlock(this::onNewBlock)
         } catch (e: Exception) {
             log.error("Failed to connect to websocket at $uri. Error: ${e.message}")
-            return
         }
-        client.onNewBlock {
+    }
-            if (it.totalDifficulty == null || it.transactions == null) {
+
-                Mono.just(it.hash).flatMap { hash ->
+    fun onNewBlock(block: BlockJson<TransactionRefJson>) {
-                    api.executeAndConvert(Commands.eth().getBlock(hash))
+        if (block.totalDifficulty == null || block.transactions == null) {
-                }.repeatWhenEmpty { n ->
+            Mono.just(block.hash).flatMap { hash ->
-                    Repeat.times<Any>(10)
+                        // first check in cache, if empty then check api
-                            .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250))
+                        blockCache.read(hash)
-                            .apply(n)
+                                .switchIfEmpty(api.executeAndConvert(Commands.eth().getBlock(hash)))
-                }
+                    }.repeatWhenEmpty { n ->
-                .timeout(Defaults.timeout, Mono.empty())
+                        Repeat.times<Any>(10)
-                .subscribe(topic::onNext)
+                                .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250))
-            } else {
+                                .apply(n)
-                topic.onNext(it)
+                    }
-            }
+                    .timeout(Defaults.timeout, Mono.empty())
+                    .subscribe(topic::onNext)
+        } else {
+            topic.onNext(block)
         }
     }
 
@@ -79,4 +88,8 @@ class EthereumWs(
         return Flux.from(this.topic)
                 .onBackpressureLatest()
     }
+
+    override fun setCaches(caches: Caches) {
+        blockCache = caches.getBlocksByHash()
+    }
 }

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

@@ -15,13 +15,15 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 
 class EthereumWsHead(
         private val ws: EthereumWs
-): DefaultEthereumHead(), Lifecycle {
+): DefaultEthereumHead(), Lifecycle, CachesEnabled {
 
     private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)
 
@@ -40,4 +42,8 @@ class EthereumWsHead(
         subscription = null
     }
 
+    override fun setCaches(caches: Caches) {
+        ws.setCaches(caches)
+    }
+
 }

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

@@ -32,6 +32,8 @@ import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 
+import java.util.concurrent.Callable
+
 class EthereumApiMock extends DirectEthereumApi {
 
     private static final Logger log = LoggerFactory.getLogger(this)
@@ -55,26 +57,30 @@ class EthereumApiMock extends DirectEthereumApi {
 
     @Override
     Mono<byte[]> execute(int id, @NotNull String method, @NotNull List<?> params) {
-        def predefined = predefined.find { it.isSame(id, method, params) }
+        Callable<byte[]> call = {
-        ResponseJson json = new ResponseJson<Object, Integer>(id: id)
+            def predefined = predefined.find { it.isSame(id, method, params) }
-        if (predefined != null) {
+            ResponseJson json = new ResponseJson<Object, Integer>(id: id)
-            if (predefined.exception != null) {
+            if (predefined != null) {
+                if (predefined.exception != null) {
+                    predefined.onCalled()
+                    predefined.print()
+                    throw predefined.exception
+                }
+                if (predefined.result instanceof RpcResponseError) {
+                    json.error = predefined.result
+                } else {
+                    json.result = predefined.result
+                }
                 predefined.onCalled()
                 predefined.print()
-                return Mono.error(predefined.exception)
-            }
-            if (predefined.result instanceof RpcResponseError) {
-                json.error = predefined.result
             } else {
-                json.result = predefined.result
+                log.error("Method ${method} with ${params} is not mocked")
+                json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked")
             }
-        } else {
+            byte[] result = objectMapper.writeValueAsBytes(json)
-            log.error("Method ${method} with ${params} is not mocked")
+            return result
-            json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked")
+        } as Callable<byte[]>
-        }
+        return Mono.fromCallable(call)
-        predefined.onCalled()
-        predefined.print()
-        return Mono.just(objectMapper.writeValueAsBytes(json))
     }
 
     def nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {

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

@@ -0,0 +1,73 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.cache.BlocksMemCache
+import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.cache.HeightCache
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.rpc.ReactorRpcClient
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import io.infinitape.etherjar.rpc.ws.WebsocketClient
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+import java.time.Instant
+import java.time.temporal.ChronoUnit
+
+class EthereumWsSpec extends Specification {
+
+    def "Uses cache to fetch block"() {
+        setup:
+        ReactorRpcClient rpcClient = Stub(ReactorRpcClient)
+        def apiMock = TestingCommons.api(rpcClient)
+        def ws = new EthereumWs(new URI("http://localhost"), new URI("http://localhost"), apiMock)
+        def blocksCache = Mock(BlocksMemCache)
+        def caches = Caches.newBuilder().setBlockByHash(blocksCache).build()
+        ws.setCaches(caches)
+
+        def block = new BlockJson<TransactionRefJson>()
+        block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
+        block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
+
+        when:
+        ws.onNewBlock(block)
+
+        then:
+        1 * blocksCache.read(BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")) >> Mono.just(block)
+        StepVerifier.create(ws.flux.take(1))
+            .expectNext(block)
+            .expectComplete()
+            .verify(Duration.ofSeconds(1))
+    }
+
+    def "Fetch block if cache is empty"() {
+        setup:
+        ReactorRpcClient rpcClient = Stub(ReactorRpcClient)
+        def apiMock = TestingCommons.api(rpcClient)
+        def ws = new EthereumWs(new URI("http://localhost"), new URI("http://localhost"), apiMock)
+        def blocksCache = Mock(BlocksMemCache)
+        def caches = Caches.newBuilder().setBlockByHash(blocksCache).build()
+        ws.setCaches(caches)
+
+        def block = new BlockJson<TransactionRefJson>()
+        block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
+        block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
+        block.transactions = []
+        block.uncles = []
+
+        apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block)
+
+        when:
+        ws.onNewBlock(block)
+
+        then:
+        1 * blocksCache.read(_) >> Mono.empty()
+        StepVerifier.create(ws.flux.take(1))
+                .expectNext(block)
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+}