Merge pull request #47 from p2p-org/forward-selectors

Added forwarding of selectors to grpc upstreams

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

@@ -242,6 +242,8 @@ open class NativeCall(
             val requestDecorator = getRequestDecorator(requestItem.method)
             val resultDecorator = getResultDecorator(requestItem.method)
 
+            val selector = request.takeIf { it.hasSelector() }?.let { Selectors.keepForwarded(it.selector) }
+
             ValidCallContext(
                 requestItem.id,
                 nonce,
@@ -250,7 +252,8 @@ open class NativeCall(
                 callQuorum,
                 RawCallDetails(method, params),
                 requestDecorator,
-                resultDecorator
+                resultDecorator,
+                selector
             )
         }
     }
@@ -267,7 +270,7 @@ open class NativeCall(
     fun fetch(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
         return ctx.upstream.getRoutedApi(ctx.matcher)
             .flatMap { api ->
-                api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce))
+                api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
                     .flatMap(JsonRpcResponse::requireResult)
                     .map {
                         if (ctx.nonce != null) {
@@ -300,7 +303,7 @@ open class NativeCall(
             AtomicInteger(-1)
         }
         return reader
-            .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce))
+            .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
             .map {
                 val bytes = ctx.resultDecorator.processResult(it)
                 CallResult(ctx.id, ctx.nonce, bytes, null, it.signature)
@@ -418,7 +421,8 @@ open class NativeCall(
         val callQuorum: CallQuorum,
         val payload: T,
         val requestDecorator: RequestDecorator,
-        val resultDecorator: ResultDecorator
+        val resultDecorator: ResultDecorator,
+        val forwardedSelector: BlockchainOuterClass.Selector?
     ) : CallContext {
 
         constructor(
@@ -428,7 +432,7 @@ open class NativeCall(
             matcher: Selector.Matcher,
             callQuorum: CallQuorum,
             payload: T
-        ) : this(id, nonce, upstream, matcher, callQuorum, payload, NoneRequestDecorator(), NoneResultDecorator())
+        ) : this(id, nonce, upstream, matcher, callQuorum, payload, NoneRequestDecorator(), NoneResultDecorator(), null)
 
         override fun isValid(): Boolean {
             return true
@@ -443,7 +447,7 @@ open class NativeCall(
         }
 
         fun <X> withPayload(payload: X): ValidCallContext<X> {
-            return ValidCallContext(id, nonce, upstream, matcher, callQuorum, payload, requestDecorator, resultDecorator)
+            return ValidCallContext(id, nonce, upstream, matcher, callQuorum, payload, requestDecorator, resultDecorator, forwardedSelector)
         }
 
         fun getApis(): ApiSource {

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

@@ -0,0 +1,43 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.api.proto.BlockchainOuterClass.*
+
+object Selectors {
+
+    private fun processMultiple(original: List<Selector>, creator: (List<Selector>) -> Selector.Builder): Selector? {
+        val newSelectors = original.mapNotNull { keepForwarded(it) }
+        return if (newSelectors.size > 1) {
+            return creator(newSelectors).setShouldBeForwarded(true).build()
+        } else if (newSelectors.size == 1) {
+            newSelectors.first()
+        } else {
+            null
+        }
+    }
+
+    fun keepForwarded(selector: Selector): Selector? {
+        if (!selector.shouldBeForwarded) return null
+
+        if (selector.hasOrSelector()) {
+            return processMultiple(selector.orSelector.selectorsList) {
+                Selector.newBuilder()
+                    .setOrSelector(OrSelector.newBuilder().addAllSelectors(it))
+            }
+        } else if (selector.hasAndSelector()) {
+            return processMultiple(selector.andSelector.selectorsList) {
+                Selector.newBuilder()
+                    .setAndSelector(AndSelector.newBuilder().addAllSelectors(it))
+            }
+        } else if (selector.hasNotSelector()) {
+            val inner = keepForwarded(selector.notSelector.selector)
+            return inner?.let {
+                Selector.newBuilder()
+                    .setNotSelector(NotSelector.newBuilder().setSelector(it).build())
+                    .setShouldBeForwarded(true)
+                    .build()
+            }
+        } else {
+            return selector
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt

@@ -26,7 +26,6 @@ import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
@@ -67,7 +66,7 @@ class BitcoinGrpcUpstream(
     }
 
     private val extractBlock = ExtractBlock()
-    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)
+    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
     private val blockConverter: Function<BlockchainOuterClass.ChainHead, BlockContainer> = Function { value ->
         val block = BlockContainer(
             value.height,

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt

@@ -28,7 +28,6 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
@@ -106,7 +105,7 @@ open class EthereumGrpcUpstream(
     private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
     private var capabilities: Set<Capability> = emptySet()
 
-    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)
+    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
     var timeout = Defaults.timeout
 
     override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt

@@ -28,7 +28,6 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
@@ -106,7 +105,7 @@ open class EthereumPosGrpcUpstream(
     private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, NoChoiceWithPriorityForkChoice(nodeRating))
     private var capabilities: Set<Capability> = emptySet()
 
-    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)
+    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
     var timeout = Defaults.timeout
 
     override fun start() {

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt

@@ -22,7 +22,6 @@ import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.reader.Reader
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
@@ -40,14 +39,13 @@ class JsonRpcGrpcClient(
         private val log = LoggerFactory.getLogger(JsonRpcGrpcClient::class.java)
     }
 
-    fun forSelector(matcher: Selector.Matcher): Reader<JsonRpcRequest, JsonRpcResponse> {
+    fun getReader(): Reader<JsonRpcRequest, JsonRpcResponse> {
-        return Executor(stub, chain, matcher, metrics)
+        return Executor(stub, chain, metrics)
     }
 
     class Executor(
         private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
         private val chain: Chain,
-        private val matcher: Selector.Matcher,
         private val metrics: RpcMetrics
     ) : Reader<JsonRpcRequest, JsonRpcResponse> {
 
@@ -56,11 +54,7 @@ class JsonRpcGrpcClient(
             val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
                 .setChainValue(chain.id)
 
-            if (matcher != Selector.empty) {
+            key.selector?.let { req.selector = it }
-                Selector.extractLabels(matcher)?.asProto().let {
-                    req.setSelector(it)
-                }
-            }
 
             val reqItem = BlockchainOuterClass.NativeCallItem.newBuilder()
                 .setId(1)

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt

@@ -19,16 +19,23 @@ import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.databind.DeserializationContext
 import com.fasterxml.jackson.databind.JsonDeserializer
 import com.fasterxml.jackson.databind.JsonNode
+import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.Global
 
 data class JsonRpcRequest(
     val method: String,
     val params: List<Any?>,
     val id: Int,
-    val nonce: Long?
+    val nonce: Long?,
+    val selector: BlockchainOuterClass.Selector?
 ) {
 
-    @JvmOverloads constructor(method: String, params: List<Any?>, nonce: Long? = null) : this(method, params, 1, nonce)
+    @JvmOverloads constructor(
+        method: String,
+        params: List<Any?>,
+        nonce: Long? = null,
+        selectors: BlockchainOuterClass.Selector? = null
+    ) : this(method, params, 1, nonce, selectors)
 
     fun toJson(): ByteArray {
         val json = mapOf(
@@ -63,7 +70,7 @@ data class JsonRpcRequest(
                     throw IllegalStateException("Unsupported param type: ${it.asToken()}")
                 }
             }
-            return JsonRpcRequest(method, params, id, null)
+            return JsonRpcRequest(method, params, id, null, null)
         }
     }
 }

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

@@ -543,7 +543,7 @@ class NativeCallSpec extends Specification {
         def nativeCall = nativeCall()
         def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
                 new NativeCall.RawCallDetails("eth_getFilterUpdates", '["0xabcd"]'),
-                new NativeCall.WithFilterIdDecorator(), new NativeCall.NoneResultDecorator())
+                new NativeCall.WithFilterIdDecorator(), new NativeCall.NoneResultDecorator(), null)
         when:
         def act = nativeCall.parseParams(ctx)
         then:
@@ -564,7 +564,7 @@ class NativeCallSpec extends Specification {
         }
         def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
                 new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
-                new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator())
+                new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null)
 
         when:
         def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
@@ -586,7 +586,7 @@ class NativeCallSpec extends Specification {
         }
         def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
                 new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
-                new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator())
+                new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null)
 
         when:
         def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))

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

@@ -84,7 +84,7 @@ class WsConnectionSpec extends Specification {
 
         when:
         Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
-        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null))
+        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
 
         then:
         StepVerifier.create(act)
@@ -106,7 +106,7 @@ class WsConnectionSpec extends Specification {
 
         when:
         Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
-        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null))
+        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
 
         then:
         StepVerifier.create(act)
@@ -130,7 +130,7 @@ class WsConnectionSpec extends Specification {
 
         when:
         Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe()
-        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null))
+        def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null))
 
         then:
         StepVerifier.create(act)

src/test/kotlin/io/emeraldpay/dshackle/rpc/SelectorsTest.kt

@@ -0,0 +1,117 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.api.proto.BlockchainOuterClass.*
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.Arguments
+import org.junit.jupiter.params.provider.MethodSource
+import java.util.stream.Stream
+
+internal class SelectorsTest {
+
+    companion object {
+        @JvmStatic
+        fun data(): Stream<Arguments> {
+            val leafLabelSelector = LabelSelector.newBuilder().build()
+
+            return Stream.of(
+                Arguments.of(
+                    Selector.newBuilder().setLabelSelector(leafLabelSelector).build(),
+                    null
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setLabelSelector(leafLabelSelector).build(),
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setLabelSelector(leafLabelSelector).build()
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setOrSelector(OrSelector.newBuilder()).build(),
+                    null
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setOrSelector(OrSelector.newBuilder()).build(),
+                    null
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setOrSelector(
+                            OrSelector.newBuilder().addSelectors(
+                                Selector.newBuilder().setLabelSelector(leafLabelSelector)
+                            )
+                        ).build(),
+                    null
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setOrSelector(
+                            OrSelector.newBuilder().addSelectors(
+                                Selector.newBuilder().setShouldBeForwarded(true)
+                                    .setLabelSelector(leafLabelSelector)
+                            )
+                        ).build(),
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setLabelSelector(leafLabelSelector).build()
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setOrSelector(
+                            OrSelector.newBuilder()
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                                .addSelectors(Selector.newBuilder().setLabelSelector(leafLabelSelector))
+                        ).build(),
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setOrSelector(
+                            OrSelector.newBuilder()
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                        ).build()
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setAndSelector(
+                            AndSelector.newBuilder()
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                                .addSelectors(Selector.newBuilder().setLabelSelector(leafLabelSelector))
+                        ).build(),
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setAndSelector(
+                            AndSelector.newBuilder()
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                                .addSelectors(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                        ).build()
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setNotSelector(
+                            NotSelector.newBuilder()
+                                .setSelector(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                        ).build(),
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setNotSelector(
+                            NotSelector.newBuilder()
+                                .setSelector(Selector.newBuilder().setShouldBeForwarded(true).setLabelSelector(leafLabelSelector))
+                        ).build()
+                ),
+                Arguments.of(
+                    Selector.newBuilder().setShouldBeForwarded(true)
+                        .setNotSelector(
+                            NotSelector.newBuilder()
+                                .setSelector(Selector.newBuilder().setLabelSelector(leafLabelSelector))
+                        ).build(),
+                    null
+                )
+
+            )
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("data")
+    fun testKeepForwarded(input: Selector, expected: Selector?) {
+        assertEquals(expected, Selectors.keepForwarded(input))
+    }
+}