From 73f55d8d815212545e1e5b0c3fa488cfb408c5cf Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 14 May 2020 23:47:59 -0400 Subject: [PATCH] problem: routed api is not executed first --- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 27 +++++++--- .../dshackle/upstream/Multistream.kt | 2 +- .../upstream/bitcoin/BitcoinMultistream.kt | 3 +- .../upstream/ethereum/EthereumMultistream.kt | 4 +- .../upstream/ethereum/NativeCallRouter.kt | 1 - .../dshackle/rpc/NativeCallSpec.groovy | 54 +++++++++++++++++++ .../dshackle/test/EthereumUpstreamMock.groovy | 17 ++++-- .../ethereum/NativeCallRouterSpec.groovy | 1 - 8 files changed, 91 insertions(+), 18 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index b8ca8bbe..0926e07b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -19,14 +19,17 @@ package io.emeraldpay.dshackle.rpc import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.RpcResponseError import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -116,17 +119,25 @@ open class NativeCall( } fun fetch(ctx: CallContext): Mono> { -// ctx.upstream.getRoutedApi(ctx.matcher) -// .flatMap { api -> -// api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params)) -// }.switchIfEmpty( -// Mono.just(ctx).flatMap(this::executeOnRemote) -// ) - //TODO use routed api - return executeOnRemote(ctx) + return ctx.upstream.getRoutedApi(ctx.matcher) + .flatMap { api -> + api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params)) + .flatMap(JsonRpcResponse::requireResult) + .map { + ctx.withPayload(it) + } + }.switchIfEmpty( + Mono.just(ctx).flatMap(this::executeOnRemote) + ) + .onErrorMap { + CallFailure(ctx.id, it) + } } fun executeOnRemote(ctx: CallContext): Mono> { + if (!ctx.upstream.getMethods().isAllowed(ctx.payload.method)) { + return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method")) + } //TODO move to routed api val apis = ctx.getApis() apis.request(1) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 5c9feb39..ed4bf2f1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -101,7 +101,7 @@ abstract class Multistream( val apis = getApiSource(matcher) apis.request(1) return Mono.from(apis) - .switchIfEmpty(Mono.error(Exception("No API available"))) + .switchIfEmpty(Mono.error(Exception("No API available for $chain"))) } /** diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index 131c3433..6a75e7d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -74,7 +75,7 @@ open class BitcoinMultistream( override fun getRoutedApi(matcher: Selector.Matcher): Mono> { //TODO - return getDirectApi(matcher) + return Mono.just(EmptyReader()) } open fun getReader(): BitcoinReader { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index de6d9f7b..c607b503 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -119,9 +119,7 @@ open class EthereumMultistream( } override fun getRoutedApi(matcher: Selector.Matcher): Mono> { - return getDirectApi(matcher).map { api -> - NativeCallRouter(objectMapper, reader, api, getMethods()) - } + return Mono.just(NativeCallRouter(objectMapper, reader, getMethods())) } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt index 22afa79d..52d3f2f7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt @@ -32,7 +32,6 @@ import java.math.BigInteger class NativeCallRouter( private val objectMapper: ObjectMapper, private val reader: EthereumReader, - private val directApi: Reader, private val methods: CallMethods ) : Reader { diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 6da8e1df..13c06663 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -19,11 +19,16 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.quorum.BroadcastQuorum +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.NonEmptyQuorum +import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.MultistreamHolder +import io.emeraldpay.dshackle.upstream.ethereum.NativeCallRouter +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.rpc.ReactorRpcClient import io.infinitape.etherjar.rpc.RpcException @@ -40,6 +45,55 @@ class NativeCallSpec extends Specification { def objectMapper = TestingCommons.objectMapper() + def "Tries router first"() { + def routedApi = Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(new JsonRpcResponse("1".bytes, null)) + } + def upstream = Mock(Multistream) { + 1 * getRoutedApi(_) >> Mono.just(routedApi) + } + def upstreams = Stub(MultistreamHolder) + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def ctx = new NativeCall.CallContext( + 1, upstream, Selector.empty, new AlwaysQuorum(), + new NativeCall.ParsedCallDetails("eth_test", []) + ) + + when: + def act = nativeCall.fetch(ctx).block(Duration.ofSeconds(1)) + then: + act.payload == "1".bytes + } + + def "Return error if router denied the requests"() { + def routedApi = Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.error(new RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Test message")) + } + def upstream = Mock(Multistream) { + 1 * getRoutedApi(_) >> Mono.just(routedApi) + } + def upstreams = Stub(MultistreamHolder) + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def ctx = new NativeCall.CallContext( + 15, upstream, Selector.empty, new AlwaysQuorum(), + new NativeCall.ParsedCallDetails("eth_test", []) + ) + + when: + def act = nativeCall.fetch(ctx) //.block(Duration.ofSeconds(1)) + then: + StepVerifier.create(act) + .expectErrorMatches { t -> + t instanceof NativeCall.CallFailure && + t.id == 15 && + t.reason instanceof RpcException && + t.reason.rpcMessage == "Test message" + } + .verify(Duration.ofSeconds(1)) + } + def "Quorum is applied"() { setup: def quorum = Spy(new AlwaysQuorum()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy index 5d2eca77..f9ccff6f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy @@ -16,13 +16,16 @@ */ package io.emeraldpay.dshackle.test -import io.emeraldpay.dshackle.cache.Caches + import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods +import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -36,12 +39,20 @@ class EthereumUpstreamMock extends EthereumUpstream { EthereumHeadMock ethereumHeadMock = new EthereumHeadMock() + static CallMethods allMethods() { + new AggregatedCallMethods([ + new DefaultEthereumMethods(TestingCommons.objectMapper(), Chain.ETHEREUM), + new DefaultBitcoinMethods(TestingCommons.objectMapper()), + new DirectCallMethods(["eth_test"]) + ]) + } + EthereumUpstreamMock(@NotNull Chain chain, @NotNull Reader api) { - this(chain, api, new DefaultEthereumMethods(TestingCommons.objectMapper(), chain)) + this(chain, api, allMethods()) } EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api) { - this(id, chain, api, new DefaultEthereumMethods(TestingCommons.objectMapper(), chain)) + this(id, chain, api, allMethods()) } EthereumUpstreamMock(@NotNull Chain chain, @NotNull Reader api, CallMethods methods) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy index 712bb047..f35c516f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouterSpec.groovy @@ -23,7 +23,6 @@ class NativeCallRouterSpec extends Specification { Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper() ), - new EmptyReader(), methods ) when: