problem: produces no response for unavailable methods

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

@@ -77,16 +77,21 @@ open class NativeCall(
 
     open fun nativeCallResult(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<CallResult> {
         return requestMono.flatMapMany(this::prepareCall)
-            .map(this::parseParams)
-            .parallel()
             .flatMap {
-                this.fetch(it)
+                if (it.isValid()) {
-                    .doOnError { e -> log.warn("Error during native call: ${e.message}") }
+                    val parsed = parseParams(it.get())
+                    this.fetch(parsed)
+                        .doOnError { e -> log.warn("Error during native call: ${e.message}") }
+                } else {
+                    val error = it.getError()
+                    Mono.just(
+                        CallResult(error.id, null, error)
+                    )
+                }
             }
-            .sequential()
     }
 
-    fun parseParams(it: CallContext<RawCallDetails>): CallContext<ParsedCallDetails> {
+    fun parseParams(it: ValidCallContext<RawCallDetails>): ValidCallContext<ParsedCallDetails> {
         val params = extractParams(it.payload.params)
         return it.withPayload(ParsedCallDetails(it.payload.method, params))
     }
@@ -121,7 +126,7 @@ open class NativeCall(
             .toMono()
     }
 
-    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux<CallContext<RawCallDetails>> {
+    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux<CallContext> {
         val chain = Chain.byId(request.chain.number)
         if (chain == Chain.UNSPECIFIED) {
             return Flux.error(CallFailure(0, SilentException.UnsupportedBlockchain(request.chain.number)))
@@ -140,43 +145,63 @@ open class NativeCall(
     fun prepareCall(
         request: BlockchainOuterClass.NativeCallRequest,
         upstream: Multistream
-    ): Flux<CallContext<RawCallDetails>> {
+    ): Flux<CallContext> {
         val chain = Chain.byId(request.chainValue)
-        return Flux.fromIterable(request.itemsList).flatMap {
+        return Flux.fromIterable(request.itemsList)
-            val method = it.method
+            .flatMap {
-            val params = it.payload.toStringUtf8()
+                prepareIndividualCall(chain, request, it, upstream)
-
-            // for ethereum the actual block needed for the call may be specified in the call parameters
-            val callSpecificMatcher: Mono<Selector.Matcher> =
-                if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) {
-                    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))
-
-                val callQuorum = upstream.getMethods().getQuorumFor(method) // can be null in tests
-                callQuorum.init(upstream.getHead())
-
-                // for NotLaggingQuorum it makes sense to select compatible upstreams before the call
-                if (callQuorum is NotLaggingQuorum) {
-                    val lag = callQuorum.maxLag
-                    val minHeight = ((upstream.getHead().getCurrentHeight() ?: 0) - lag).coerceAtLeast(0)
-                    val heightMatcher = Selector.HeightMatcher(minHeight)
-                    matcher.withMatcher(heightMatcher)
-                }
-
-                CallContext(it.id, upstream, matcher.build(), callQuorum, RawCallDetails(method, params))
             }
+    }
+
+    fun prepareIndividualCall(
+        chain: Chain,
+        request: BlockchainOuterClass.NativeCallRequest,
+        requestItem: BlockchainOuterClass.NativeCallItem,
+        upstream: Multistream
+    ): Mono<CallContext> {
+        val method = requestItem.method
+        val params = requestItem.payload.toStringUtf8()
+        val availableMethods = upstream.getMethods()
+
+        if (!availableMethods.isAllowed(method)) {
+            val errorMessage = "The method $method does not exist/is not available"
+            return Mono.just(
+                InvalidCallContext(
+                    CallError(requestItem.id, errorMessage, JsonRpcError(RpcResponseError.CODE_METHOD_NOT_EXIST, errorMessage))
+                )
+            )
+        }
+
+        // for ethereum the actual block needed for the call may be specified in the call parameters
+        val callSpecificMatcher: Mono<Selector.Matcher> =
+            if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) {
+                ethereumCallSelectors[chain]?.getMatcher(method, params, upstream.getHead())
+            } else {
+                null
+            } ?: Mono.empty()
+
+        return callSpecificMatcher.defaultIfEmpty(Selector.empty).map { csm ->
+            val matcher = Selector.Builder()
+                .withMatcher(csm)
+                .forMethod(method)
+                .forLabels(Selector.convertToMatcher(request.selector))
+
+            val callQuorum = availableMethods.getQuorumFor(method) // can be null in tests
+            callQuorum.init(upstream.getHead())
+
+            // for NotLaggingQuorum it makes sense to select compatible upstreams before the call
+            if (callQuorum is NotLaggingQuorum) {
+                val lag = callQuorum.maxLag
+                val minHeight = ((upstream.getHead().getCurrentHeight() ?: 0) - lag).coerceAtLeast(0)
+                val heightMatcher = Selector.HeightMatcher(minHeight)
+                matcher.withMatcher(heightMatcher)
+            }
+
+            ValidCallContext(requestItem.id, upstream, matcher.build(), callQuorum, RawCallDetails(method, params))
         }
     }
 
-    fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallResult> {
+    fun fetch(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
         return ctx.upstream.getRoutedApi(ctx.matcher)
             .flatMap { api ->
                 api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
@@ -196,7 +221,7 @@ open class NativeCall(
             }
     }
 
-    fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallResult> {
+    fun executeOnRemote(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
         if (!ctx.upstream.getMethods().isAllowed(ctx.payload.method)) {
             return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
         }
@@ -228,15 +253,33 @@ open class NativeCall(
         return req as List<Any>
     }
 
-    open class CallContext<T>(
+    interface CallContext {
+        fun isValid(): Boolean
+        fun <T> get(): ValidCallContext<T>
+        fun getError(): CallError
+    }
+
+    open class ValidCallContext<T>(
         val id: Int,
         val upstream: Multistream,
         val matcher: Selector.Matcher,
         val callQuorum: CallQuorum,
         val payload: T
-    ) {
+    ) : CallContext {
-        fun <X> withPayload(payload: X): CallContext<X> {
+        override fun isValid(): Boolean {
-            return CallContext(id, upstream, matcher, callQuorum, payload)
+            return true
+        }
+
+        override fun <X> get(): ValidCallContext<X> {
+            return this as ValidCallContext<X>
+        }
+
+        override fun getError(): CallError {
+            throw IllegalStateException("Invalid context $id")
+        }
+
+        fun <X> withPayload(payload: X): ValidCallContext<X> {
+            return ValidCallContext(id, upstream, matcher, callQuorum, payload)
         }
 
         fun getApis(): ApiSource {
@@ -244,6 +287,25 @@ open class NativeCall(
         }
     }
 
+    /**
+     * Call context when it's known in advance that the call is invalid and should return an error
+     */
+    open class InvalidCallContext(
+        private val error: CallError
+    ) : CallContext {
+        override fun isValid(): Boolean {
+            return false
+        }
+
+        override fun <T> get(): ValidCallContext<T> {
+            throw IllegalStateException("Invalid context ${error.id}")
+        }
+
+        override fun getError(): CallError {
+            return error
+        }
+    }
+
     open class CallFailure(val id: Int, val reason: Throwable) : Exception("Failed to call $id: ${reason.message}")
 
     open class CallError(val id: Int, val message: String, val upstreamError: JsonRpcError?) {

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

@@ -61,7 +61,7 @@ class NativeCallSpec extends Specification {
         def upstreams = Stub(MultistreamHolder)
 
         def nativeCall = new NativeCall(upstreams)
-        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(
+        def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(
                 1, upstream, Selector.empty, new AlwaysQuorum(),
                 new NativeCall.ParsedCallDetails("eth_test", [])
         )
@@ -82,7 +82,7 @@ class NativeCallSpec extends Specification {
         def upstreams = Stub(MultistreamHolder)
 
         def nativeCall = new NativeCall(upstreams)
-        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(
+        def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(
                 15, upstream, Selector.empty, new AlwaysQuorum(),
                 new NativeCall.ParsedCallDetails("eth_test", [])
         )
@@ -108,7 +108,7 @@ class NativeCallSpec extends Specification {
                 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, 1))
             }
         }
-        def call = new NativeCall.CallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
+        def call = new NativeCall.ValidCallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
                 new NativeCall.ParsedCallDetails("eth_test", []))
 
         when:
@@ -128,7 +128,7 @@ class NativeCallSpec extends Specification {
                 1 * read(_) >> Mono.empty()
             }
         }
-        def call = new NativeCall.CallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
+        def call = new NativeCall.ValidCallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
                 new NativeCall.ParsedCallDetails("eth_test", []))
 
         when:
@@ -295,6 +295,33 @@ class NativeCallSpec extends Specification {
         }
     }
 
+    def "Prepare call with unsupported method"() {
+        setup:
+        def upstreams = Mock(MultistreamHolder) {
+            _ * it.observeChains() >> Flux.empty()
+        }
+        def nativeCall = new NativeCall(upstreams)
+
+        def item = BlockchainOuterClass.NativeCallItem.newBuilder()
+                .setId(1)
+                .setMethod("eth_testInvalid")
+                .build()
+        def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
+                .setChain(Common.ChainRef.CHAIN_ETHEREUM)
+                .addItems(item)
+                .build()
+        when:
+        def act = nativeCall.prepareIndividualCall(Chain.ETHEREUM, req, item, TestingCommons.emptyMultistream())
+                .block(Duration.ofSeconds(1))
+        then:
+        act instanceof NativeCall.InvalidCallContext
+        with(((NativeCall.InvalidCallContext) act).error) {
+            it.upstreamError != null
+            it.upstreamError.code == -32601
+            it.upstreamError.message.contains("eth_testInvalid")
+        }
+    }
+
     def "Prepare call adds height selector for not-lagging quorum"() {
         setup:
         def methods = new ManagedCallMethods(
@@ -339,7 +366,7 @@ class NativeCallSpec extends Specification {
     def "Parse empty params"() {
         setup:
         def nativeCall = new NativeCall(Stub(MultistreamHolder))
-        def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
+        def ctx = new NativeCall.ValidCallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
                 new NativeCall.RawCallDetails("eth_test", "[]"))
         when:
         def act = nativeCall.parseParams(ctx)
@@ -352,7 +379,7 @@ class NativeCallSpec extends Specification {
     def "Parse none params"() {
         setup:
         def nativeCall = new NativeCall(Stub(MultistreamHolder))
-        def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
+        def ctx = new NativeCall.ValidCallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
                 new NativeCall.RawCallDetails("eth_test", ""))
         when:
         def act = nativeCall.parseParams(ctx)
@@ -365,7 +392,7 @@ class NativeCallSpec extends Specification {
     def "Parse single param"() {
         setup:
         def nativeCall = new NativeCall(Stub(MultistreamHolder))
-        def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
+        def ctx = new NativeCall.ValidCallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
                 new NativeCall.RawCallDetails("eth_test", "[false]"))
         when:
         def act = nativeCall.parseParams(ctx)
@@ -378,7 +405,7 @@ class NativeCallSpec extends Specification {
     def "Parse multi param"() {
         setup:
         def nativeCall = new NativeCall(Stub(MultistreamHolder))
-        def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
+        def ctx = new NativeCall.ValidCallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
                 new NativeCall.RawCallDetails("eth_test", "[false, 123]"))
         when:
         def act = nativeCall.parseParams(ctx)
@@ -397,7 +424,7 @@ class NativeCallSpec extends Specification {
         def api = TestingCommons.api()
         def upstream = TestingCommons.multistream(api)
 
-        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10,
+        def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(10,
                 upstream,
                 Selector.empty, new AlwaysQuorum(),
                 new NativeCall.ParsedCallDetails("eth_test", []))
@@ -415,7 +442,7 @@ class NativeCallSpec extends Specification {
         def nativeCall = new NativeCall(upstreams)
         def upstream = TestingCommons.multistream(TestingCommons.api())
 
-        def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10,
+        def ctx = new NativeCall.ValidCallContext<NativeCall.ParsedCallDetails>(10,
                 upstream,
                 Selector.empty, new AlwaysQuorum(),
                 new NativeCall.ParsedCallDetails("eth_test", []))