eth_filter fix review comments and tests

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -69,6 +69,7 @@ open class UpstreamsConfig {
 
     class Upstream<T : UpstreamConnection> {
         var id: String? = null
+        var nodeId: Int? = null
         var chain: String? = null
         var options: Options? = null
         var isEnabled = true

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -31,6 +31,7 @@ class UpstreamsConfigReader(
 
     private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java)
     private val authConfigReader = AuthConfigReader()
+    private val knownNodeIds: MutableSet<Int> = HashSet()
 
     fun read(input: InputStream): UpstreamsConfig? {
         val configNode = readNode(input)
@@ -236,11 +237,22 @@ class UpstreamsConfigReader(
             log.warn("Invalid id: $id")
             return false
         }
-        return true
+        return upstream.nodeId?.let {
+            if (it !in 1..255) {
+                log.warn("Invalid node-id: $it. Must be in range [1, 255].")
+                false
+            } else if (!knownNodeIds.add(it)) {
+                log.warn("Duplicated node-id: $it. Must be in unique.")
+                false
+            } else {
+                true
+            }
+        } ?: true
     }
 
     internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
         upstream.id = getValueAsString(upNode, "id")
+        upstream.nodeId = getValueAsInt(upNode, "node-id")
         upstream.options = tryReadOptions(upNode)
         upstream.methods = tryReadMethods(upNode)
         getValueAsBool(upNode, "enabled")?.let {

src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt

@@ -119,7 +119,7 @@ class QuorumRpcReader(
                 .filter { it.isResolved() } // return nothing if not resolved
                 .map { quorum ->
                     // TODO find actual quorum number
-                    Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy().map { it.hash() })
+                    Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy().map { it.nodeId() })
                 }
                 .switchIfEmpty(defaultResult)
         }

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

@@ -16,6 +16,7 @@
  */
 package io.emeraldpay.dshackle.startup
 
+import com.google.common.annotations.VisibleForTesting
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.UpstreamsConfig
@@ -81,7 +82,7 @@ open class ConfiguredUpstreams(
             log.debug("Start upstream ${up.id}")
             if (up.connection is UpstreamsConfig.GrpcConnection) {
                 val options = up.options ?: UpstreamsConfig.Options()
-                buildGrpcUpstream(up.cast(UpstreamsConfig.GrpcConnection::class.java), options)
+                buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options)
             } else {
                 val chain = Global.chainById(up.chain)
                 if (chain == Chain.UNSPECIFIED) {
@@ -92,7 +93,7 @@ open class ConfiguredUpstreams(
                     .merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults())
                 val upstream = when (BlockchainType.from(chain)) {
                     BlockchainType.ETHEREUM -> {
-                        buildEthereumUpstream(up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
+                        buildEthereumUpstream(up.nodeId, up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
                     }
 
                     BlockchainType.BITCOIN -> {
@@ -101,6 +102,7 @@ open class ConfiguredUpstreams(
 
                     BlockchainType.ETHEREUM_POS -> {
                         buildEthereumPosUpstream(
+                            up.nodeId,
                             up.cast(UpstreamsConfig.EthereumPosConnection::class.java),
                             chain,
                             options
@@ -167,6 +169,7 @@ open class ConfiguredUpstreams(
     }
 
     private fun buildEthereumPosUpstream(
+        nodeId: Int?,
         config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumPosConnection>,
         chain: Chain,
         options: UpstreamsConfig.Options
@@ -194,7 +197,7 @@ open class ConfiguredUpstreams(
         val hashUrl = conn.execution!!.let {
             if (it.preferHttp) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
         }
-        val hash = getHash(hashUrl!!)
+        val hash = getHash(nodeId, hashUrl!!)
         val upstream = EthereumPosRpcUpstream(
             config.id!!,
             hash,
@@ -251,6 +254,7 @@ open class ConfiguredUpstreams(
     }
 
     private fun buildEthereumUpstream(
+        nodeId: Int?,
         config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
         chain: Chain,
         options: UpstreamsConfig.Options
@@ -275,7 +279,7 @@ open class ConfiguredUpstreams(
         val hashUrl = if (conn.preferHttp) conn.rpc?.url ?: conn.ws?.url else conn.ws?.url ?: conn.rpc?.url
         val upstream = EthereumRpcUpstream(
             config.id!!,
-            getHash(hashUrl!!),
+            getHash(nodeId, hashUrl!!),
             chain,
             options, config.role,
             methods,
@@ -287,11 +291,12 @@ open class ConfiguredUpstreams(
     }
 
     private fun buildGrpcUpstream(
+        nodeId: Int?,
         config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
         options: UpstreamsConfig.Options
     ) {
         val endpoint = config.connection!!
-        val hash = getHash("${endpoint.host}:${endpoint.port}")
+        val hash = getHash(nodeId, "${endpoint.host}:${endpoint.port}")
         val ds = GrpcUpstreams(
             config.id!!,
             hash,
@@ -366,23 +371,25 @@ open class ConfiguredUpstreams(
         return connectorFactory
     }
 
-    private fun getHash(obj: Any): Byte {
+    @VisibleForTesting
-        val hashCode = (obj.hashCode() % 255)
+    private fun getHash(nodeId: Int?, obj: Any): Byte =
-        val modifiers: List<Function<Int, Number>> = listOf(
+        nodeId?.toByte() ?: (obj.hashCode() % 255).let {
-            Function { i -> i },
+            if (it == 0) 1 else it
-            Function { i -> (-i) },
+        }.let { nonZeroHash ->
-            Function { i -> 127 - abs(i) },
+            listOf<Function<Int, Int>>(
-            Function { i -> abs(i) - 128 },
+                Function { i -> i },
-        )
+                Function { i -> (-i) },
-        return modifiers.map {
+                Function { i -> 127 - abs(i) },
-            it.apply(hashCode).toByte()
+                Function { i -> abs(i) - 128 },
-        }.firstOrNull {
+            ).map {
-            hashes[it] != true
+                it.apply(nonZeroHash).toByte()
-        }?.let {
+            }.firstOrNull {
-            hashes[it] = true
+                hashes[it] != true
-            it
+            }?.let {
-        } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
+                hashes[it] = true
-            hashes[it.toByte()] != true
+                it
-        }.toByte()
+            } ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
-    }
+                it != 0 && hashes[it.toByte()] != true
+            }.toByte()
+        }
 }

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

@@ -143,7 +143,7 @@ abstract class DefaultUpstream(
         return targets ?: throw IllegalStateException("Methods are not set")
     }
 
-    override fun hash(): Byte = hash
+    override fun nodeId(): Byte = hash
 
     private val quorumByLabel = node?.let { QuorumForLabels(it) }
         ?: QuorumForLabels(QuorumForLabels.QuorumItem.empty())

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

@@ -282,7 +282,7 @@ abstract class Multistream(
         return false
     }
 
-    override fun hash(): Byte = 0
+    override fun nodeId(): Byte = 0
 
     fun printStatus() {
         var height: Long? = null

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

@@ -397,15 +397,22 @@ class Selector {
         }
     }
 
-    class SameUpstreamMatcher(private val upstreamHash: Byte) : Matcher {
+    class SameNodeMatcher(private val upstreamHash: Byte) : Matcher {
         override fun matches(up: Upstream): Boolean =
-            up.hash() == upstreamHash
+            up.nodeId() == upstreamHash
 
         override fun describeInternal(): String =
-            "upstream hash=$upstreamHash"
+            "upstream node-id=${upstreamHash.toUByte()}"
 
         override fun toString(): String {
             return "Matcher: ${describeInternal()}"
         }
+
+        override fun equals(other: Any?): Boolean {
+            if (other === this) return true
+            if (other !is SameNodeMatcher) return false
+            return other.upstreamHash == upstreamHash
+        }
+
     }
 }

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

@@ -41,5 +41,5 @@ interface Upstream {
 
     fun <T : Upstream> cast(selfType: Class<T>): T
 
-    fun hash(): Byte
+    fun nodeId(): Byte
 }

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

@@ -69,14 +69,12 @@ class EthereumCallSelector(
             return Mono.empty()
         }
         val filterId = list[0].toString()
-        val hashHex = filterId.substring(filterId.length - 2)
+        if(filterId.length < 4) {
-        val hash = hashHex.toInt(16)
+            return Mono.just(Selector.SameNodeMatcher(0.toByte()))
-
-        if (hash < 0 || hash > 255) {
-            return Mono.empty()
         }
-
+        val hashHex = filterId.substring(filterId.length - 2)
-        return Mono.just(Selector.SameUpstreamMatcher(hash.toByte()))
+        val nodeId = hashHex.toInt(16)
+        return Mono.just(Selector.SameNodeMatcher(nodeId.toByte()))
     }
 
     private fun blockTagSelector(params: String, pos: Int, head: Head): Mono<Selector.Matcher> {

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -401,4 +401,18 @@ class UpstreamsConfigReaderSpec extends Specification {
         act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.PRIMARY
         act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.PRIMARY
     }
+
+    def "Parse node id"() {
+        setup:
+        def config = this.class.getClassLoader().getResourceAsStream("upstreams-node-id.yaml")
+        when:
+        def act = reader.read(config)
+        then:
+        act != null
+        act.upstreams.size() == 2
+        act.upstreams[0].nodeId == 1
+        act.upstreams[0].id == "has_node_id"
+        act.upstreams[1].nodeId == null
+        act.upstreams[1].id == "has_no_node_id"
+    }
 }

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

@@ -395,6 +395,68 @@ class NativeCallSpec extends Specification {
         }
     }
 
+    def "Prepare call adds decorator for eth_newFilter"() {
+        setup:
+        def methods = new ManagedCallMethods(
+                new DefaultEthereumMethods(Chain.ETHEREUM),
+                ["eth_newFilter"] as Set, [] as Set
+        )
+        methods.setQuorum("eth_newFilter", "always")
+        def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
+        multistream.customMethods = methods
+        multistream.customHead = Mock(Head)
+        def multistreamHolder = Mock(MultistreamHolder) {
+            _ * it.observeChains() >> Flux.empty()
+        }
+        def nativeCall = nativeCall(multistreamHolder)
+
+        def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
+                .setChain(Common.ChainRef.CHAIN_ETHEREUM)
+                .addItems(
+                        BlockchainOuterClass.NativeCallItem.newBuilder()
+                                .setId(1)
+                                .setMethod("eth_newFilter")
+                )
+                .build()
+        when:
+        def act = nativeCall.prepareCall(req, multistream)
+                .collectList().block(Duration.ofSeconds(1)).first()
+        then:
+        act instanceof NativeCall.ValidCallContext
+        act.resultDecorator instanceof NativeCall.CreateFilterDecorator
+    }
+
+    def "Prepare call adds decorator for eth_getFilterChanges"() {
+        setup:
+        def methods = new ManagedCallMethods(
+                new DefaultEthereumMethods(Chain.ETHEREUM),
+                ["eth_getFilterChanges"] as Set, [] as Set
+        )
+        methods.setQuorum("eth_getFilterChanges", "always")
+        def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
+        multistream.customMethods = methods
+        multistream.customHead = Mock(Head)
+        def multistreamHolder = Mock(MultistreamHolder) {
+            _ * it.observeChains() >> Flux.empty()
+        }
+        def nativeCall = nativeCall(multistreamHolder)
+
+        def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
+                .setChain(Common.ChainRef.CHAIN_ETHEREUM)
+                .addItems(
+                        BlockchainOuterClass.NativeCallItem.newBuilder()
+                                .setId(1)
+                                .setMethod("eth_getFilterChanges")
+                )
+                .build()
+        when:
+        def act = nativeCall.prepareCall(req, multistream)
+                .collectList().block(Duration.ofSeconds(1)).first()
+        then:
+        act instanceof NativeCall.ValidCallContext
+        act.requestDecorator instanceof NativeCall.GetFilterUpdatesDecorator
+    }
+
     def "Parse empty params"() {
         setup:
         def nativeCall = nativeCall()
@@ -447,6 +509,64 @@ class NativeCallSpec extends Specification {
         act.payload.method == "eth_test"
     }
 
+    def "Decorate eth_getFilterUpdates params"() {
+        setup:
+        def nativeCall = nativeCall()
+        def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
+                new NativeCall.RawCallDetails("eth_getFilterUpdates", '["0xabcd"]'),
+                new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.NoneResultDecorator())
+        when:
+        def act = nativeCall.parseParams(ctx)
+        then:
+        act.id == 1
+        act.payload.params == ["0xab"]
+        act.payload.method == "eth_getFilterUpdates"
+    }
+
+    def "Decorate eth_newFilter result"() {
+        setup:
+        def quorum = new AlwaysQuorum()
+
+        def nativeCall = nativeCall()
+        nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            1 * create(_, _, _) >> Mock(Reader) {
+                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)255)))
+            }
+        }
+        def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
+                new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.CreateFilterDecorator())
+
+        when:
+        def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
+        def act = objectMapper.readValue(resp.result, Object)
+        then:
+        act == "0xabff"
+        resp.nonce == 10
+    }
+
+    def "Decorate eth_newFilter result with short nodeId"() {
+        setup:
+        def quorum = new AlwaysQuorum()
+
+        def nativeCall = nativeCall()
+        nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
+            1 * create(_, _, _) >> Mock(Reader) {
+                1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)1)))
+            }
+        }
+        def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
+                new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.CreateFilterDecorator())
+
+        when:
+        def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
+        def act = objectMapper.readValue(resp.result, Object)
+        then:
+        act == "0xab01"
+        resp.nonce == 10
+    }
+
     @Ignore
     //TODO
     def "Calls cache before remote"() {

src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy

@@ -58,4 +58,37 @@ class ConfiguredUpstreamsSpec extends Specification {
         act instanceof ManagedCallMethods
         new String(act.executeHardcoded("foo_bar")) == "\"static_response\""
     }
+
+    def "Calculate node-id"() {
+        setup:
+        def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig)
+        )
+        expect:
+        configurer.getHash(node, src) == expected
+
+        where:
+        node | src | expected
+        1 | "" | 1
+        9 | "hohoho" | 9
+        null | "hohoho" | 120
+    }
+
+    def "Calculate node-id conflicting results"() {
+        setup:
+        def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig)
+        )
+        when:
+        def h1 = configurer.getHash(null, "hohoho")
+        def h2 = configurer.getHash(null, "hohoho")
+        def h3 = configurer.getHash(null, "hohoho")
+        def h4 = configurer.getHash(null, "hohoho")
+        def h5 = configurer.getHash(null, "hohoho")
+
+        then:
+        h1 == (byte)120
+        h2 == (byte)-120
+        h3 == (byte)-9
+        h4 == (byte)8
+        h5 == (byte)-128
+    }
 }

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

@@ -349,4 +349,27 @@ class SelectorSpec extends Specification {
         !act
     }
 
+    def "Matches same nodeId"() {
+        setup:
+        def up = Mock(Upstream) {
+            nodeId() >> (byte)5
+        }
+        def matcher = new Selector.SameNodeMatcher((byte)5)
+        when:
+        def act = matcher.matches(up)
+        then:
+        act
+    }
+
+    def "Not matches nodeId"() {
+        setup:
+        def up = Mock(Upstream) {
+            nodeId() >> (byte)5
+        }
+        def matcher = new Selector.SameNodeMatcher((byte)1)
+        when:
+        def act = matcher.matches(up)
+        then:
+        !act
+    }
 }

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

@@ -179,4 +179,22 @@ class EthereumCallSelectorSpec extends Specification {
         then:
         act == new Selector.HeightMatcher(100)
     }
+
+    def "Get same matcher for getFilterChanges method"() {
+        setup:
+        def callSelector = new EthereumCallSelector(Mock(Reader))
+        def head = Mock(Head)
+
+        expect:
+        callSelector.getMatcher("eth_getFilterChanges", param, head).block()
+                == new Selector.SameNodeMatcher((byte)hash)
+
+        where:
+        param | hash
+        '["0xff09"]' | 9
+        '["0xff"]' | 255
+        '[]' | 0
+        '[""]' | 0
+        '["0x0"]' | 0
+    }
 }

src/test/resources/upstreams-node-id.yaml

@@ -0,0 +1,34 @@
+version: v1
+upstreams:
+  - id: has_node_id
+    node-id: 1
+    chain: ethereum
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+  - id: has_no_node_id
+    chain: ethereum
+    connection:
+      ethereum:
+        rpc:
+          url: "http://localhost:8545"
+        ws:
+          url: "ws://localhost:8546"
+  - id: conflicted_node_id
+    node-id: 1
+    chain: ethereum
+    connection:
+      prefer-http: true
+      ethereum:
+        rpc:
+          url: "http://localhost:9545"
+        ws:
+          url: "ws://localhost:9546"
+  - id: invalid_node_id
+    node-id: 256
+    chain: ethereum
+    connection:
+      grpc:
+        host: "localhost"
+        port: 2449