From 2a3d7ecb244c0f1111cb77c2c43e4d7eb74863f4 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 22 Dec 2021 21:47:31 -0500 Subject: [PATCH] problem: produces no response for unavailable methods --- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 150 +++++++++++++----- .../dshackle/rpc/NativeCallSpec.groovy | 47 ++++-- 2 files changed, 143 insertions(+), 54 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index b638b079..6b75b14f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -77,16 +77,21 @@ open class NativeCall( open fun nativeCallResult(requestMono: Mono): Flux { return requestMono.flatMapMany(this::prepareCall) - .map(this::parseParams) - .parallel() .flatMap { - this.fetch(it) - .doOnError { e -> log.warn("Error during native call: ${e.message}") } + if (it.isValid()) { + 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): CallContext { + fun parseParams(it: ValidCallContext): ValidCallContext { 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> { + fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux { 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> { + ): Flux { 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 callSpecificMatcher: Mono = - 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)) + return Flux.fromIterable(request.itemsList) + .flatMap { + prepareIndividualCall(chain, request, it, upstream) } + } + + fun prepareIndividualCall( + chain: Chain, + request: BlockchainOuterClass.NativeCallRequest, + requestItem: BlockchainOuterClass.NativeCallItem, + upstream: Multistream + ): Mono { + 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 = + 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): Mono { + fun fetch(ctx: ValidCallContext): Mono { 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): Mono { + fun executeOnRemote(ctx: ValidCallContext): Mono { 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 } - open class CallContext( + interface CallContext { + fun isValid(): Boolean + fun get(): ValidCallContext + fun getError(): CallError + } + + open class ValidCallContext( val id: Int, val upstream: Multistream, val matcher: Selector.Matcher, val callQuorum: CallQuorum, val payload: T - ) { - fun withPayload(payload: X): CallContext { - return CallContext(id, upstream, matcher, callQuorum, payload) + ) : CallContext { + override fun isValid(): Boolean { + return true + } + + override fun get(): ValidCallContext { + return this as ValidCallContext + } + + override fun getError(): CallError { + throw IllegalStateException("Invalid context $id") + } + + fun withPayload(payload: X): ValidCallContext { + 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 get(): ValidCallContext { + 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?) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 32dfbca8..16e9c630 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/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( + def ctx = new NativeCall.ValidCallContext( 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( + def ctx = new NativeCall.ValidCallContext( 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(10, + def ctx = new NativeCall.ValidCallContext(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(10, + def ctx = new NativeCall.ValidCallContext(10, upstream, Selector.empty, new AlwaysQuorum(), new NativeCall.ParsedCallDetails("eth_test", []))