problem: routing doesn't consider block ref in request (i.e. "latest")

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

@@ -25,10 +25,12 @@ import io.emeraldpay.dshackle.upstream.*
 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.rpcclient.JsonRpcError
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.grpc.BlockchainType
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.RpcResponseError
@@ -48,6 +50,7 @@ open class NativeCall(
     private val objectMapper: ObjectMapper = Global.objectMapper
 
     var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()
+    private val ethereumCallSelector = EthereumCallSelector()
 
     open fun nativeCall(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
         return nativeCallResult(requestMono)
@@ -118,11 +121,19 @@ open class NativeCall(
     }
 
     fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: Multistream): Flux<CallContext<RawCallDetails>> {
-        return request.itemsList.toFlux().map {
+        return Flux.fromIterable(request.itemsList).map {
             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) {
+                ethereumCallSelector.getMatcher(method, params, upstream.getHead())
+            } else {
+                null
+            }
+
             val matcher = Selector.Builder()
+                    .withMatcher(callSpecificMather)
                     .forMethod(method)
                     .forLabels(Selector.convertToMatcher(request.selector))
                     .build()

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

@@ -90,6 +90,13 @@ class Selector {
             return this
         }
 
+        fun withMatcher(matcher: Matcher?): Builder {
+            matcher?.let {
+                matchers.add(it)
+            }
+            return this
+        }
+
         fun build(): Matcher {
             return MultiMatcher(matchers)
         }
@@ -240,4 +247,23 @@ class Selector {
             return up.isGrpc()
         }
     }
+
+    class HeightMatcher(val height: Long): Matcher {
+        override fun matches(up: Upstream): Boolean {
+            return (up.getHead().getCurrentHeight() ?: 0) >= height
+        }
+
+        override fun equals(other: Any?): Boolean {
+            if (this === other) return true
+            if (other !is HeightMatcher) return false
+
+            if (height != other.height) return false
+
+            return true
+        }
+
+        override fun hashCode(): Int {
+            return height.hashCode()
+        }
+    }
 }

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

@@ -0,0 +1,88 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.calls
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Selector
+import io.infinitape.etherjar.hex.HexQuantity
+import org.slf4j.LoggerFactory
+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 {
+
+    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",
+                "eth_getCode",
+                "eth_getTransactionCount",
+                // no "eth_getStorageAt" because it's has different structure, and therefore separate logic
+                "eth_call"
+        ).sorted()
+    }
+
+    private val objectMapper = Global.objectMapper
+
+    /**
+     * @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? {
+        if (Collections.binarySearch(TAG_METHODS, method) >= 0) {
+            return blockTagSelector(params, 1, head)
+        } else if (method == "eth_getStorageAt") {
+            return blockTagSelector(params, 2, head)
+        }
+        return null
+    }
+
+    private fun blockTagSelector(params: String, pos: Int, head: Head): 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
+        }
+        // integer block number, or the string "latest", "earliest" or "pending"
+        val minHeight = when (val tag = list[pos].toString()) {
+            "latest" -> head.getCurrentHeight() ?: 0
+            // for earliest it doesn't nothing, we expect to have 0 block
+            "earliest" -> 0L
+            else -> if (tag.startsWith("0x")) {
+                    try {
+                        HexQuantity.from(tag).value.toLong()
+                    } catch (t: Throwable) {
+                        log.debug("Invalid tag: $tag. ${t.javaClass}: ${t.message}")
+                        0L
+                    }
+                } else {
+                    log.debug("Invalid tag: $tag")
+                    0L
+                }
+        }
+        return if (minHeight > 0) {
+            Selector.HeightMatcher(minHeight)
+        } else {
+            null
+        }
+    }
+
+}

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

@@ -292,4 +292,61 @@ class SelectorSpec extends Specification {
                 [test: "foo"],
         ]
     }
+
+    def "Match by height when equal"() {
+        setup:
+        def matcher = new Selector.HeightMatcher(1000)
+        def up = Mock(Upstream) {
+            1 * getHead() >> Mock(Head) {
+                1 * getCurrentHeight() >> 1000
+            }
+        }
+        when:
+        def act = matcher.matches(up)
+        then:
+        act
+    }
+
+    def "Match by height when higher"() {
+        setup:
+        def matcher = new Selector.HeightMatcher(1000)
+        def up = Mock(Upstream) {
+            1 * getHead() >> Mock(Head) {
+                1 * getCurrentHeight() >> 1001
+            }
+        }
+        when:
+        def act = matcher.matches(up)
+        then:
+        act
+    }
+
+    def "No match by height when less"() {
+        setup:
+        def matcher = new Selector.HeightMatcher(1000)
+        def up = Mock(Upstream) {
+            1 * getHead() >> Mock(Head) {
+                1 * getCurrentHeight() >> 999
+            }
+        }
+        when:
+        def act = matcher.matches(up)
+        then:
+        !act
+    }
+
+    def "No match by height when none"() {
+        setup:
+        def matcher = new Selector.HeightMatcher(1000)
+        def up = Mock(Upstream) {
+            1 * getHead() >> Mock(Head) {
+                1 * getCurrentHeight() >> null
+            }
+        }
+        when:
+        def act = matcher.matches(up)
+        then:
+        !act
+    }
+
 }

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

@@ -0,0 +1,81 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.calls
+
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Selector
+import spock.lang.Specification
+
+class EthereumCallSelectorSpec extends Specification {
+
+    EthereumCallSelector callSelector = new EthereumCallSelector()
+
+    def "Get height matcher for latest balance"() {
+        setup:
+        def head = Mock(Head) {
+            1 * getCurrentHeight() >> 100
+        }
+        when:
+        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "latest"]', head)
+        then:
+        act == new Selector.HeightMatcher(100)
+    }
+
+    def "Get height matcher for latest call"() {
+        setup:
+        def head = Mock(Head) {
+            1 * getCurrentHeight() >> 100
+        }
+        when:
+        def act = callSelector.getMatcher("eth_call", '["0x0000", "latest"]', head)
+        then:
+        act == new Selector.HeightMatcher(100)
+    }
+
+    def "Get height matcher for latest storageAt"() {
+        setup:
+        def head = Mock(Head) {
+            1 * getCurrentHeight() >> 100
+        }
+        when:
+        def act = callSelector.getMatcher("eth_getStorageAt", '["0x295a70b2de5e3953354a6a8344e616ed314d7251", "0x0", "latest"]', head)
+        then:
+        act == new Selector.HeightMatcher(100)
+    }
+
+    def "Get height matcher for balance on block"() {
+        setup:
+        def head = Mock(Head) {
+            _ * getCurrentHeight() >> 100
+        }
+        when:
+        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "0x40"]', head)
+        then:
+        act == new Selector.HeightMatcher(0x40)
+    }
+
+    def "No matcher for pending balance"() {
+        setup:
+        def head = Mock(Head) {
+            _ * getCurrentHeight() >> 100
+        }
+        when:
+        def act = callSelector.getMatcher("eth_getBalance", '["0x0000", "pending"]', head)
+        then:
+        act == null
+    }
+
+}