problem: doesn't select upstream by available block hash, see EIP-1898

rel: #81

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

@@ -35,7 +35,8 @@ open class Caches(
         private val memTxsByHash: TxMemCache,
         private val redisBlocksByHash: BlocksRedisCache?,
         private val redisTxsByHash: TxRedisCache?,
-        private val redisReceipts: ReceiptRedisCache?
+        private val redisReceipts: ReceiptRedisCache?,
+        private val redisHeightByHashCache: HeightByHashRedisCache?
 ) {
 
     companion object {
@@ -52,6 +53,8 @@ open class Caches(
         }
     }
 
+    private val memHeightByHash: HeightByHashMemCache = HeightByHashMemCache()
+
     private val blocksByHash: Reader<BlockId, BlockContainer>
     private val txsByHash: Reader<TxId, TxContainer>
     private val receiptByHash: Reader<TxId, ByteArray>
@@ -105,6 +108,9 @@ open class Caches(
 
     fun cache(tag: Tag, block: BlockContainer) {
         val job = ArrayList<Mono<Void>>()
+
+        redisHeightByHashCache?.add(block)?.let(job::add)
+
         if (tag == Tag.LATEST) {
             //for LATEST data cache it in memory, it may be short living so better to avoid Redis
             memoizeBlock(block)
@@ -147,6 +153,7 @@ open class Caches(
      */
     fun memoizeBlock(block: BlockContainer) {
         memBlocksByHash.add(block)
+        memHeightByHash.add(block)
         val replaced = blocksByHeight.add(block)
         //evict cached transactions if an existing block was updated
         replaced?.let { evict(it) }
@@ -193,6 +200,14 @@ open class Caches(
         return receiptByHash
     }
 
+    fun getLastHeightByHash(): Reader<BlockId, Long> {
+        return memHeightByHash
+    }
+
+    fun getRedisHeightByHash(): HeightByHashCache? {
+        return redisHeightByHashCache
+    }
+
     enum class Tag {
         /**
          * Latest data produced by blockchain
@@ -212,6 +227,7 @@ open class Caches(
         private var redisBlocksByHash: BlocksRedisCache? = null
         private var redisTxsByHash: TxRedisCache? = null
         private var redisReceiptCache: ReceiptRedisCache? = null
+        private var redisHeightByHashCache: HeightByHashRedisCache? = null
 
         fun setBlockByHash(cache: BlocksMemCache): Builder {
             blocksByHash = cache
@@ -243,6 +259,11 @@ open class Caches(
             return this
         }
 
+        fun setHeightByHash(cache: HeightByHashRedisCache): Builder {
+            redisHeightByHashCache = cache
+            return this
+        }
+
         fun build(): Caches {
             if (blocksByHash == null) {
                 blocksByHash = BlocksMemCache()
@@ -253,7 +274,8 @@ open class Caches(
             if (txsByHash == null) {
                 txsByHash = TxMemCache()
             }
-            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash, redisReceiptCache)
+            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!,
+                    redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache)
         }
     }
 }

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

@@ -96,6 +96,7 @@ class CachesFactory(
             caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain))
             caches.setTxByHash(TxRedisCache(redis.reactive(), chain))
             caches.setReceipts(ReceiptRedisCache(redis.reactive(), chain))
+            caches.setHeightByHash(HeightByHashRedisCache(redis.reactive(), chain))
         }
         return caches.build()
     }

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

@@ -0,0 +1,80 @@
+/**
+ * Copyright (c) 2021 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.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.reader.Reader
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+/**
+ * Height By Hash reader, that adds to Redis cache missing values.
+ *
+ * Tries to read value from current memory cache, then from redis if it's available, and if both empty read from blockchain.
+ * With the latter case it adds read value back to the redis cache (again, if available).
+ *
+ */
+class HeightByHashAdding(
+        private val mem: Reader<BlockId, Long>,
+        private val redis: HeightByHashCache?,
+        private val upstreamReader: Reader<BlockId, BlockContainer>
+) : Reader<BlockId, Long> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(HeightByHashAdding::class.java)
+    }
+
+    constructor(caches: Caches, upstreamReader: Reader<BlockId, BlockContainer>) :
+            this(caches.getLastHeightByHash(), caches.getRedisHeightByHash(), upstreamReader)
+
+    private val delegate: Reader<BlockId, Long>
+
+    init {
+        if (redis != null) {
+            delegate = object : Reader<BlockId, Long> {
+                override fun read(key: BlockId): Mono<Long> {
+                    return mem.read(key)
+                            .switchIfEmpty(
+                                    Mono.just(key)
+                                            .flatMap { redis.read(it) }
+                            )
+                            .switchIfEmpty(
+                                    Mono.just(key)
+                                            .flatMap { upstreamReader.read(it) }
+                                            .flatMap { redis.add(it).then(Mono.just(it.height)) }
+                            )
+                }
+            }
+        } else {
+            delegate = object : Reader<BlockId, Long> {
+                override fun read(key: BlockId): Mono<Long> {
+                    return mem.read(key)
+                            .switchIfEmpty(
+                                    Mono.just(key)
+                                            .flatMap { upstreamReader.read(it) }
+                                            .map { it.height }
+                            )
+                }
+            }
+        }
+    }
+
+    override fun read(key: BlockId): Mono<Long> {
+        return delegate.read(key)
+    }
+
+}

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

@@ -0,0 +1,26 @@
+/**
+ * Copyright (c) 2021 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.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.reader.Reader
+import reactor.core.publisher.Mono
+
+interface HeightByHashCache : Reader<BlockId, Long> {
+
+    fun add(block: BlockContainer): Mono<Void>
+}

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

@@ -0,0 +1,44 @@
+/**
+ * Copyright (c) 2021 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.github.benmanes.caffeine.cache.Caffeine
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.reader.Reader
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+class HeightByHashMemCache(
+        maxSize: Int = 256
+) : Reader<BlockId, Long> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(HeightByHashMemCache::class.java)
+    }
+
+    private val heights = Caffeine.newBuilder()
+            .maximumSize(maxSize.toLong())
+            .build<BlockId, Long>()
+
+    override fun read(key: BlockId): Mono<Long> {
+        return Mono.justOrEmpty(heights.getIfPresent(key))
+    }
+
+    fun add(block: BlockContainer) {
+        heights.put(block.hash, block.height)
+    }
+}

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

@@ -0,0 +1,102 @@
+/**
+ * Copyright (c) 2021 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.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+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.nio.ByteBuffer
+import java.nio.ByteOrder
+import java.util.concurrent.TimeUnit
+
+/**
+ * Cache for height by hash.
+ * Different from blocks cache because for height we don't really care about eviction and replaced blocks (also a fall-back block
+ * reader would use full block's cache to find out height).
+ */
+class HeightByHashRedisCache(
+        private val redis: RedisReactiveCommands<String, ByteArray>,
+        private val chain: Chain
+) : Reader<BlockId, Long>, HeightByHashCache {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(HeightByHashRedisCache::class.java)
+
+        private const val MAX_CACHE_TIME_MINUTES = 60L * 4
+    }
+
+    override fun read(key: BlockId): Mono<Long> {
+        return redis.get(key(key))
+                .flatMap { data ->
+                    Mono.justOrEmpty(fromBytes(data)) as Mono<Long>
+                }.onErrorResume {
+                    log.warn("Failed to read Block Height. ${it.javaClass}:${it.message}")
+                    Mono.empty()
+                }
+    }
+
+    override fun add(block: BlockContainer): Mono<Void> {
+        return Mono.just(block)
+                .flatMap { block ->
+                    // even if block replaced, the mapping hash-long is still valid, so can be cached for long time
+                    // even for fresh blocks
+                    val ttl = TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES)
+
+                    val key = key(block.hash)
+                    val value = asBytes(block.height)
+                    redis.setex(key, ttl, value)
+                }
+                .doOnError {
+                    log.warn("Failed to save Block Height. ${it.javaClass}:${it.message}")
+                }
+                //if failed to cache, just continue without it
+                .onErrorResume {
+                    Mono.empty()
+                }
+                .then()
+    }
+
+    fun asBytes(value: Long): ByteArray {
+        val result = ByteArray(8)
+        val bb = ByteBuffer.allocate(8)
+                .order(ByteOrder.BIG_ENDIAN)
+        bb.asLongBuffer()
+                .put(value)
+        bb.get(result)
+        return result
+    }
+
+    fun fromBytes(value: ByteArray): Long? {
+        if (value.size != 8) {
+            return null
+        }
+        return ByteBuffer.wrap(value)
+                .order(ByteOrder.BIG_ENDIAN)
+                .asLongBuffer()
+                .get()
+    }
+
+    /**
+     * Key in Redis
+     */
+    fun key(hash: BlockId): String {
+        return "height:${chain.id}:${hash.toHex()}"
+    }
+}

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

@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
 import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
 import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -40,6 +41,7 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import reactor.core.publisher.*
 import java.lang.Exception
+import java.util.*
 
 @Service
 open class NativeCall(
@@ -50,7 +52,18 @@ open class NativeCall(
     private val objectMapper: ObjectMapper = Global.objectMapper
 
     var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()
-    private val ethereumCallSelector = EthereumCallSelector()
+    private val ethereumCallSelectors = EnumMap<Chain, EthereumCallSelector>(Chain::class.java)
+
+    init {
+        multistreamHolder.observeChains().subscribe { chain ->
+            if (!ethereumCallSelectors.containsKey(chain)) {
+                multistreamHolder.getUpstream(chain)?.let { up ->
+                    val reader = up.cast(EthereumMultistream::class.java).getReader()
+                    ethereumCallSelectors[chain] = EthereumCallSelector(reader.heightByHash())
+                }
+            }
+        }
+    }
 
     open fun nativeCall(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
         return nativeCallResult(requestMono)
@@ -121,27 +134,30 @@ open class NativeCall(
     }
 
     fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: Multistream): Flux<CallContext<RawCallDetails>> {
-        return Flux.fromIterable(request.itemsList).map {
+        val chain = Chain.byId(request.chainValue)
+        return Flux.fromIterable(request.itemsList).flatMap {
             val method = it.method
             val params = it.payload.toStringUtf8()
 
             // for ethereum the actual block needed for the call may be specified in the call parameters
-            val callSpecificMather = if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) {
+            val callSpecificMatcher: Mono<Selector.Matcher> = if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) {
-                ethereumCallSelector.getMatcher(method, params, upstream.getHead())
+                ethereumCallSelectors[chain]?.getMatcher(method, params, upstream.getHead())
             } else {
                 null
+            } ?: Mono.empty()
+
+            callSpecificMatcher.defaultIfEmpty(Selector.empty).map { csm ->
+                val matcher = Selector.Builder()
+                        .withMatcher(csm)
+                        .forMethod(method)
+                        .forLabels(Selector.convertToMatcher(request.selector))
+                        .build()
+
+                val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum()
+                callQuorum.init(upstream.getHead())
+
+                CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params))
             }
-
-            val matcher = Selector.Builder()
-                    .withMatcher(callSpecificMather)
-                    .forMethod(method)
-                    .forLabels(Selector.convertToMatcher(request.selector))
-                    .build()
-
-            val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum()
-            callQuorum.init(upstream.getHead())
-
-            CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params))
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt

@@ -16,20 +16,26 @@
 package io.emeraldpay.dshackle.upstream.calls
 
 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Selector
 import io.infinitape.etherjar.hex.HexQuantity
 import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
 import java.util.*
 
 /**
  * Get a matcher based on a criteria provided with a RPC request. I.e. when the client requests data for "latest", or "0x19f816" block.
  * The implementation is specific for Ethereum.
  */
-class EthereumCallSelector {
+class EthereumCallSelector(
+        private val heightReader: Reader<BlockId, Long>
+) {
 
     companion object {
         private val log = LoggerFactory.getLogger(EthereumCallSelector::class.java)
+
         // ref https://eth.wiki/json-rpc/API#the-default-block-parameter
         private val TAG_METHODS = listOf(
                 "eth_getBalance",
@@ -46,20 +52,20 @@ class EthereumCallSelector {
      * @param method JSON RPC name
      * @param params JSON-encoded list of parameters for the method
      */
-    fun getMatcher(method: String, params: String, head: Head): Selector.Matcher? {
+    fun getMatcher(method: String, params: String, head: Head): Mono<Selector.Matcher> {
         if (Collections.binarySearch(TAG_METHODS, method) >= 0) {
             return blockTagSelector(params, 1, head)
         } else if (method == "eth_getStorageAt") {
             return blockTagSelector(params, 2, head)
         }
-        return null
+        return Mono.empty()
     }
 
-    private fun blockTagSelector(params: String, pos: Int, head: Head): Selector.Matcher? {
+    private fun blockTagSelector(params: String, pos: Int, head: Head): Mono<Selector.Matcher> {
         val list = objectMapper.readerFor(Any::class.java).readValues<Any>(params).readAll()
         if (list.size < pos + 1) {
             log.debug("Tag is not specified. Ignoring")
-            return null
+            return Mono.empty()
         }
         // integer block number, a string "latest", "earliest" or "pending", or an object with block reference
         val minHeight: Long? = when (val tag = list[pos].toString()) {
@@ -76,15 +82,27 @@ class EthereumCallSelector {
             } else if (tag.startsWith("{") && list[pos] is Map<*, *>) {
                 // see https://eips.ethereum.org/EIPS/eip-1898
                 val obj = list[pos] as Map<*, *>
-                if (obj.containsKey("blockNumber")) {
+                when {
-                    try {
+                    obj.containsKey("blockNumber") -> {
-                        HexQuantity.from(obj["blockNumber"].toString()).value.toLong()
+                        try {
-                    } catch (t: Throwable) {
+                            HexQuantity.from(obj["blockNumber"].toString()).value.toLong()
-                        log.warn("Invalid blockNumber: $tag")
+                        } catch (t: Throwable) {
-                        null
+                            log.warn("Invalid blockNumber: $tag")
+                            null
+                        }
                     }
-                } else {
+                    obj.containsKey("blockHash") -> {
-                    null
+                        try {
+                            val blockId = BlockId.from(obj["blockHash"].toString())
+                            return heightReader.read(blockId)
+                                    .switchIfEmpty(Mono.justOrEmpty(head.getCurrentHeight()))
+                                    .map { Selector.HeightMatcher(it) }
+                        } catch (t: Throwable) {
+                            log.warn("Invalid blockHash: $tag")
+                            null
+                        }
+                    }
+                    else -> null
                 }
             } else {
                 log.debug("Invalid tag: $tag")
@@ -92,9 +110,9 @@ class EthereumCallSelector {
             }
         }
         return if (minHeight != null && minHeight >= 0) {
-            Selector.HeightMatcher(minHeight)
+            Mono.just(Selector.HeightMatcher(minHeight))
         } else {
-            null
+            Mono.empty()
         }
     }
 

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

@@ -19,6 +19,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CurrentBlockCache
+import io.emeraldpay.dshackle.cache.HeightByHashAdding
 import io.emeraldpay.dshackle.data.*
 import io.emeraldpay.dshackle.reader.*
 import io.emeraldpay.dshackle.upstream.Multistream
@@ -77,6 +78,13 @@ open class EthereumReader(
     private val txHashToId = Function<TransactionId, TxId> { hash -> TxId.from(hash) }
     private val idToTxHash = Function<TxId, TransactionId> { id -> TransactionId.from(id.value) }
 
+    private val blocksByIdAsCont = CompoundReader(
+            caches.getBlocksByHash(),
+            RekeyingReader(idToBlockHash, directReader.blockReader)
+    )
+
+    private val heightByHash = HeightByHashAdding(caches, blocksByIdAsCont)
+
     fun blocksByHashAsCont(): Reader<BlockHash, BlockContainer> {
         return CompoundReader(
                 RekeyingReader(blockHashToId, caches.getBlocksByHash()),
@@ -99,10 +107,7 @@ open class EthereumReader(
     }
 
     open fun blocksByIdAsCont(): Reader<BlockId, BlockContainer> {
-        return CompoundReader(
+        return blocksByIdAsCont
-                caches.getBlocksByHash(),
-                RekeyingReader(idToBlockHash, directReader.blockReader)
-        )
     }
 
     open fun blocksByHeightAsCont(): Reader<Long, BlockContainer> {
@@ -140,6 +145,10 @@ open class EthereumReader(
         return caches.getReceipts()
     }
 
+    fun heightByHash(): Reader<BlockId, Long> {
+        return heightByHash
+    }
+
     override fun isRunning(): Boolean {
         //TODO should be always running?
         return up.isRunning

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

@@ -0,0 +1,130 @@
+/**
+ * Copyright (c) 2021 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.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import reactor.core.publisher.Mono
+import spock.lang.Specification
+import io.emeraldpay.dshackle.reader.Reader
+
+import java.time.Instant
+
+class HeightByHashAddingSpec extends Specification {
+
+    def block = new BlockContainer(
+            12079192L, BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"),
+            BigInteger.ONE, Instant.now(), false, "".bytes, null, []
+    )
+
+    def "use memory if available"() {
+        setup:
+        def mem = new HeightByHashMemCache()
+        def upstream = Mock(Reader)
+        def reader = new HeightByHashAdding(
+                mem, null, upstream
+        )
+
+        when:
+        mem.add(block)
+        def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block()
+        then:
+        act == 12079192L
+        0 * upstream.read(_)
+    }
+
+    def "call remote if not in memory and no redis"() {
+        setup:
+        def mem = new HeightByHashMemCache()
+        def upstream = Mock(Reader)
+        def reader = new HeightByHashAdding(
+                mem, null, upstream
+        )
+
+        when:
+        def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block()
+        then:
+        act == 12079192L
+        1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(block)
+    }
+
+    def "get from redis if not in memory"() {
+        setup:
+        def mem = new HeightByHashMemCache()
+        def upstream = Mock(Reader)
+        def redis = Mock(HeightByHashCache)
+        def reader = new HeightByHashAdding(
+                mem, redis, upstream
+        )
+
+        when:
+        def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block()
+        then:
+        act == 12079192L
+        0 * upstream.read(_)
+        1 * redis.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(12079192L)
+    }
+
+    def "call remote if not in memory and not in redis, add to redis"() {
+        setup:
+        def mem = new HeightByHashMemCache()
+        def upstream = Mock(Reader)
+        def redis = Mock(HeightByHashCache)
+        def reader = new HeightByHashAdding(
+                mem, redis, upstream
+        )
+
+        when:
+        def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block()
+        then:
+        act == 12079192L
+        1 * redis.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty()
+        1 * redis.add(block) >> Mono.just(true).then()
+        1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(block)
+    }
+
+    def "empty is nowhere found, without redis"() {
+        setup:
+        def mem = new HeightByHashMemCache()
+        def upstream = Mock(Reader)
+        def reader = new HeightByHashAdding(
+                mem, null, upstream
+        )
+
+        when:
+        def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block()
+        then:
+        act == null
+        1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty()
+    }
+
+    def "empty is nowhere found, with redis"() {
+        setup:
+        def mem = new HeightByHashMemCache()
+        def upstream = Mock(Reader)
+        def redis = Mock(HeightByHashCache)
+        def reader = new HeightByHashAdding(
+                mem, redis, upstream
+        )
+
+        when:
+        def act = reader.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")).block()
+        then:
+        act == null
+        1 * redis.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty()
+        1 * upstream.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty()
+    }
+}

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

@@ -0,0 +1,72 @@
+/**
+ * Copyright (c) 2021 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.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.test.IntegrationTestingCommons
+import io.emeraldpay.grpc.Chain
+import io.lettuce.core.api.StatefulRedisConnection
+import spock.lang.IgnoreIf
+import spock.lang.Specification
+
+import java.time.Instant
+
+@IgnoreIf({ IntegrationTestingCommons.isDisabled("redis") })
+class HeightByHashRedisCacheSpec extends Specification {
+
+    def block1 = new BlockContainer(
+            12079192L, BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"),
+            BigInteger.ONE, Instant.now(), false, "".bytes, null, []
+    )
+    def block2 = new BlockContainer(
+            12079193L, BlockId.from("0xd27944b460632699768fbfec3e5d454db590cae43d470b5f42fc4d091e372c25"),
+            BigInteger.ONE, Instant.now(), false, "".bytes, null, []
+    )
+
+    StatefulRedisConnection<String, byte[]> redis
+    HeightByHashRedisCache cache
+
+    def setup() {
+        redis = IntegrationTestingCommons.redisConnection()
+        redis.sync().flushdb()
+        cache = new HeightByHashRedisCache(
+                redis.reactive(), Chain.ETHEREUM
+        )
+    }
+
+    def "Add and read"() {
+        when:
+        cache.add(block1).subscribe()
+        def act = cache.read(block1.hash).block()
+        then:
+        act == 12079192L
+    }
+
+    def "Add and read multiple"() {
+        when:
+        cache.add(block1).subscribe()
+        cache.add(block2).subscribe()
+
+        def act = cache.read(block1.hash).block()
+        then:
+        act == 12079192L
+
+        def act2 = cache.read(block2.hash).block()
+        then:
+        act2 == 12079193L
+    }
+}

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

@@ -34,6 +34,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.RpcResponseError
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
 import spock.lang.Ignore
@@ -210,7 +211,9 @@ class NativeCallSpec extends Specification {
 
     def "Returns error for unsupported chain"() {
         setup:
-        def upstreams = Mock(MultistreamHolder)
+        def upstreams = Mock(MultistreamHolder) {
+            _ * it.observeChains() >> Flux.empty()
+        }
         def nativeCall = new NativeCall(upstreams)
 
         def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
@@ -234,7 +237,9 @@ class NativeCallSpec extends Specification {
 
     def "Prepare call"() {
         setup:
-        def upstreams = Mock(MultistreamHolder)
+        def upstreams = Mock(MultistreamHolder) {
+            _ * it.observeChains() >> Flux.empty()
+        }
         def nativeCall = new NativeCall(upstreams)
 
         def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
@@ -260,7 +265,9 @@ class NativeCallSpec extends Specification {
 
     def "Prepare call without payload"() {
         setup:
-        def upstreams = Mock(MultistreamHolder)
+        def upstreams = Mock(MultistreamHolder) {
+            _ * it.observeChains() >> Flux.empty()
+        }
         def nativeCall = new NativeCall(upstreams)
 
         def req = BlockchainOuterClass.NativeCallRequest.newBuilder()

src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy

@@ -15,75 +15,115 @@
  */
 package io.emeraldpay.dshackle.upstream.calls
 
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Selector
+import reactor.core.publisher.Mono
 import spock.lang.Specification
 
+
 class EthereumCallSelectorSpec extends Specification {
 
-    EthereumCallSelector callSelector = new EthereumCallSelector()
-
     def "Get height matcher for latest balance"() {
         setup:
+        EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader))
         def head = Mock(Head) {
             1 * getCurrentHeight() >> 100
         }
         when:
-        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "latest"]', head)
+        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "latest"]', head).block()
         then:
         act == new Selector.HeightMatcher(100)
     }
 
     def "Get height matcher for latest call"() {
         setup:
+        EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader))
         def head = Mock(Head) {
             1 * getCurrentHeight() >> 100
         }
         when:
-        def act = callSelector.getMatcher("eth_call", '["0x0000", "latest"]', head)
+        def act = callSelector.getMatcher("eth_call", '["0x0000", "latest"]', head).block()
         then:
         act == new Selector.HeightMatcher(100)
     }
 
     def "Get height matcher for latest storageAt"() {
         setup:
+        EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader))
         def head = Mock(Head) {
             1 * getCurrentHeight() >> 100
         }
         when:
-        def act = callSelector.getMatcher("eth_getStorageAt", '["0x295a70b2de5e3953354a6a8344e616ed314d7251", "0x0", "latest"]', head)
+        def act = callSelector.getMatcher("eth_getStorageAt", '["0x295a70b2de5e3953354a6a8344e616ed314d7251", "0x0", "latest"]', head).block()
         then:
         act == new Selector.HeightMatcher(100)
     }
 
     def "Get height matcher for balance on block"() {
         setup:
+        EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader))
         def head = Mock(Head) {
             _ * getCurrentHeight() >> 100
         }
         when:
-        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "0x40"]', head)
+        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "0x40"]', head).block()
         then:
         act == new Selector.HeightMatcher(0x40)
     }
 
     def "No matcher for pending balance"() {
         setup:
+        EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader))
         def head = Mock(Head) {
             _ * getCurrentHeight() >> 100
         }
         when:
-        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "pending"]', head)
+        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "pending"]', head).block()
         then:
         act == null
     }
 
     def "Get height matcher with EIP-1898"() {
         setup:
+        EthereumCallSelector callSelector = new EthereumCallSelector(Stub(Reader))
         def head = Stub(Head)
         when:
-        def act = callSelector.getMatcher("eth_call", '["0x0000", {"blockNumber": "0x100"}]', head)
+        def act = callSelector.getMatcher("eth_call", '["0x0000", {"blockNumber": "0x100"}]', head).block()
         then:
         act == new Selector.HeightMatcher(0x100)
     }
+
+    def "Get hash matcher with EIP-1898"() {
+        setup:
+        def heights = Mock(Reader) {
+            1 * it.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.just(12079192L)
+        }
+        EthereumCallSelector callSelector = new EthereumCallSelector(heights)
+        def head = Stub(Head)
+        when:
+        def act = callSelector.getMatcher("eth_call",
+                '["0x0000", {"blockHash": "0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"}]', head)
+                .block()
+        then:
+        act == new Selector.HeightMatcher(12079192)
+    }
+
+    def "Match head if hash matcher for unknown hash"() {
+        setup:
+        def heights = Mock(Reader) {
+            1 * it.read(BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32")) >> Mono.empty()
+        }
+        EthereumCallSelector callSelector = new EthereumCallSelector(heights)
+        def head = Mock(Head) {
+            1 * it.getCurrentHeight() >> 100
+        }
+        when:
+        def act = callSelector.getMatcher("eth_call",
+                '["0x0000", {"blockHash": "0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"}]', head)
+                .block()
+        then:
+        act == new Selector.HeightMatcher(100)
+    }
 }