From 114f109722c5ae462096c21f949b8883384759e0 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 15 May 2020 21:06:32 -0400 Subject: [PATCH] solution: individual class for quorum based requests --- .../dshackle/quorum/AlwaysQuorum.kt | 4 + .../dshackle/quorum/BroadcastQuorum.kt | 4 + .../emeraldpay/dshackle/quorum/CallQuorum.kt | 2 + .../dshackle/quorum/NonEmptyQuorum.kt | 8 +- .../emeraldpay/dshackle/quorum/NonceQuorum.kt | 6 +- .../dshackle/quorum/NotLaggingQuorum.kt | 4 + .../dshackle/quorum/QuorumReaderFactory.kt | 38 ++++ .../dshackle/quorum/QuorumRpcReader.kt | 105 +++++++++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 61 +----- .../emeraldpay/dshackle/upstream/ApiSource.kt | 2 +- .../dshackle/upstream/FilteredApis.kt | 4 +- .../dshackle/upstream/Multistream.kt | 1 + .../upstream/rpcclient/JsonRpcResponse.kt | 15 ++ .../dshackle/quorum/NonEmptyQuorumSpec.groovy | 133 +++++++++++ .../dshackle/quorum/NonceQuorumSpec.groovy | 32 +++ .../quorum/QuorumRpcReaderSpec.groovy | 206 ++++++++++++++++++ .../quorum/ValueAwareQuorumSpec.groovy | 5 + .../dshackle/rpc/NativeCallSpec.groovy | 178 ++------------- .../dshackle/upstream/FilteredApisSpec.groovy | 38 ++-- 19 files changed, 611 insertions(+), 235 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt index 14757cd2..3887db3b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt @@ -32,6 +32,10 @@ open class AlwaysQuorum: CallQuorum { return resolved } + override fun isFailed(): Boolean { + return false + } + override fun record(response: ByteArray, upstream: Upstream): Boolean { result = response resolved = true diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt index 418f94d2..f87a0da7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt @@ -37,6 +37,10 @@ open class BroadcastQuorum( return calls >= quorum } + override fun isFailed(): Boolean { + return false + } + override fun getResult(): ByteArray? { return result } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt index d35d40b7..e6d3ae72 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt @@ -31,6 +31,8 @@ interface CallQuorum { fun init(head: Head) fun isResolved(): Boolean + fun isFailed(): Boolean + fun record(response: ByteArray, upstream: Upstream): Boolean fun record(error: RpcException, upstream: Upstream) fun getResult(): ByteArray? diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt index a705b13e..e2804696 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt @@ -34,7 +34,11 @@ open class NonEmptyQuorum( } override fun isResolved(): Boolean { - return result != null || tries >= maxTries + return result != null + } + + override fun isFailed(): Boolean { + return tries >= maxTries } override fun recordValue(response: ByteArray, responseValue: Any?, upstream: Upstream) { @@ -49,9 +53,11 @@ open class NonEmptyQuorum( } override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) { + tries++ } override fun record(error: RpcException, upstream: Upstream) { + tries++ } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt index 46302283..d40f4beb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt @@ -41,10 +41,14 @@ open class NonceQuorum( override fun isResolved(): Boolean { lock.withLock { - return receivedTimes >= tries || errors >= tries + return receivedTimes >= tries && !isFailed() } } + override fun isFailed(): Boolean { + return errors >= tries + } + override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream) { val value = responseValue?.let { str -> HexQuantity.from(str).value.toLong() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index 1a799346..bab164b5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -32,6 +32,10 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { return result.get() != null } + override fun isFailed(): Boolean { + return false + } + override fun record(response: ByteArray, upstream: Upstream): Boolean { val lagging = upstream.getLag() > maxLag if (!lagging) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt new file mode 100644 index 00000000..8e0727bb --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt @@ -0,0 +1,38 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.quorum + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.ApiSource +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest + +// creates instance of a Quorum based reader +interface QuorumReaderFactory { + + companion object { + fun default(): QuorumReaderFactory { + return Default() + } + } + + fun create(apis: ApiSource, quorum: CallQuorum): Reader + + class Default : QuorumReaderFactory { + override fun create(apis: ApiSource, quorum: CallQuorum): Reader { + return QuorumRpcReader(apis, quorum) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt new file mode 100644 index 00000000..449a763b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -0,0 +1,105 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.quorum + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.ApiSource +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.infinitape.etherjar.rpc.RpcException +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.util.function.Tuples + +/** + * Makes request with applying Quorum + */ +class QuorumRpcReader( + private val apis: ApiSource, + private val quorum: CallQuorum +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(QuorumRpcReader::class.java) + } + + override fun read(key: JsonRpcRequest): Mono { + apis.request(1) + + // uses a mix of retry strategy and managed Publisher for calls. + // retry is used when an error happened + // but if no error received, we check quorum and if not enough data received we request more + // eventually source of upstreams is Completed (or something Errored) and if finalizes the result + + val retrySpec = reactor.util.retry.Retry.from { signal -> + signal.takeUntil { + it.totalRetries() >= 3 || quorum.isResolved() || quorum.isFailed() + }.doOnNext { + // need one more API source if retried + apis.request(1) + } + } + + return Flux.from(apis) + .flatMap { api -> + api.getApi().read(key) + .flatMap(JsonRpcResponse::requireResult) + // on error notify quorum, it may use error message or other details + .doOnError { err -> + if (err is RpcException) { + quorum.record(err, api) + } + } + .map { Tuples.of(it, api) } + } + .retryWhen(retrySpec) + // record all correct responses until quorum reached + .reduce(quorum, { res, a -> + if (res.record(a.t1, a.t2)) { + apis.resolve() + } else { + apis.request(1) + } + res + }) + // if last call resulted in error it's still possible that request was resolved correctly. i.e. for BroadcastQuorum + .onErrorResume { err -> + if (quorum.isResolved()) { + Mono.just(quorum) + } else { + Mono.error(err) + } + } + .doOnNext { + if (!it.isResolved()) { + log.debug("No quorum for ${key.method} as ${quorum}") + } + } + // return nothing if not resolved + .filter { it.isResolved() } + .map { + // TODO find actual quorum number + QuorumRpcReader.Result(it.getResult()!!, 1) + } + } + + + class Result( + val value: ByteArray, + val quorum: Int + ) +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 0926e07b..078f8a66 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -24,6 +24,8 @@ 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.quorum.QuorumReaderFactory +import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -46,6 +48,8 @@ open class NativeCall( private val log = LoggerFactory.getLogger(NativeCall::class.java) + var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() + open fun nativeCall(requestMono: Mono): Flux { return requestMono.flatMapMany(this::prepareCall) .map(this::setupCallParams) @@ -138,61 +142,10 @@ open class NativeCall( 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) - var failures = 0 - return Flux.from(apis) - .flatMap { api -> - val upstream = ctx.upstream - api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params)) - .flatMap(JsonRpcResponse::requireResult) - // on error notify quorum, it may use error message or other details - .doOnError { err -> - if (err is RpcException) { - ctx.callQuorum.record(err, upstream) - } - } - .map { Tuples.of(it, upstream) } - } - .retry { - failures++ - if (ctx.callQuorum.isResolved()) { - false - } else if (failures < 3) { - apis.request(1) - true - } else { - false - } - } - // record all correct responses until quorum reached - .reduce(ctx.callQuorum, {res, a -> - if (res.record(a.t1, a.t2)) { - apis.resolve() - } else { - apis.request(1) - } - res - }) - // if last call resulted in error it's still possible that request was resolved correctly. i.e. for BroadcastQuorum - .onErrorResume { err -> - if (ctx.callQuorum.isResolved()) { - Mono.just(ctx.callQuorum) - } else { - Mono.error(err) - } - } - .doOnNext { - if (!it.isResolved()) { - log.debug("No quorum for ${ctx.payload.method} as ${ctx.callQuorum}") - } - } - .filter { it.isResolved() } + val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum) + return reader.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params)) .map { - val result = it.getResult() - ?: throw CallFailure(ctx.id, Exception("No response from upstream for ${ctx.payload.method}")) - ctx.withPayload(result) + ctx.withPayload(it.value) } .onErrorMap { log.error("Failed to make a call", it) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt index dc036c3e..2e62fec1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt @@ -21,7 +21,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.reactivestreams.Publisher -interface ApiSource : Publisher> { +interface ApiSource : Publisher { fun resolve() fun request(tries: Int) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 28b5bbe3..01e1e36a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -78,7 +78,7 @@ class FilteredApis( return Duration.ofMillis(time) } - override fun subscribe(subscriber: Subscriber>) { + override fun subscribe(subscriber: Subscriber) { val first = Flux.fromIterable(upstreams) val retries = (1 until repeatLimit).map { r -> Flux.fromIterable(upstreams).delaySubscription(waitDuration(r)) @@ -87,7 +87,6 @@ class FilteredApis( Flux.concat(first, retries) .filter(Upstream::isAvailable) .filter(matcher::matches) - .map { it.getApi() } .zipWith(control) .map { it.t1 } .subscribe(subscriber) @@ -98,6 +97,7 @@ class FilteredApis( } override fun request(tries: Int) { + println("requested ${tries}") //TODO check the buffer size before submitting repeat(tries) { control.onNext(true) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index ed4bf2f1..62c34e44 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -101,6 +101,7 @@ abstract class Multistream( val apis = getApiSource(matcher) apis.request(1) return Mono.from(apis) + .map(Upstream::getApi) .switchIfEmpty(Mono.error(Exception("No API available for $chain"))) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt index 27ffe616..218e3304 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt @@ -28,6 +28,21 @@ class JsonRpcResponse( companion object { private val NULL_VALUE = "null".toByteArray() + + @JvmStatic + fun ok(value: ByteArray): JsonRpcResponse { + return JsonRpcResponse(value, null) + } + + @JvmStatic + fun ok(value: String): JsonRpcResponse { + return JsonRpcResponse(value.toByteArray(), null) + } + + @JvmStatic + fun error(code: Int, msg: String): JsonRpcResponse { + return JsonRpcResponse(null, ResponseError(code, msg)) + } } fun hasResult(): Boolean { diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy new file mode 100644 index 00000000..3f38278e --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy @@ -0,0 +1,133 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.quorum + +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Upstream +import io.infinitape.etherjar.rpc.RpcException +import spock.lang.Specification + +class NonEmptyQuorumSpec extends Specification { + + def "Fail if too many errors"() { + setup: + def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record(new RpcException(1, "Internal"), upstream1) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record(new RpcException(1, "Internal"), upstream2) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record(new RpcException(1, "Internal"), upstream3) + then: + q.isFailed() + !q.isResolved() + } + + def "Fail first if not error"() { + setup: + def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record('"0x11"'.bytes, upstream1) + then: + q.isResolved() + !q.isFailed() + } + + def "Fail second if first is error"() { + setup: + def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record(new RpcException(1, "Internal"), upstream1) + then: + !q.isFailed() + !q.isResolved() + + + when: + q.record('"0x11"'.bytes, upstream2) + then: + q.isResolved() + !q.isFailed() + } + + def "Fail second if first is null"() { + setup: + def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record('null'.bytes, upstream2) + then: + !q.isFailed() + !q.isResolved() + + + when: + q.record('"0x11"'.bytes, upstream2) + then: + q.isResolved() + !q.isFailed() + } + + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy index a4c20409..83cededc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy @@ -96,4 +96,36 @@ class NonceQuorumSpec extends Specification { q.isResolved() objectMapper.readValue(q.result, Object) == "0x11" } + + def "Fail if too many errors"() { + setup: + def q = Spy(new NonceQuorum(objectMapper, 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record(new RpcException(1, "Internal"), upstream1) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record(new RpcException(1, "Internal"), upstream2) + then: + !q.isResolved() + !q.isFailed() + + when: + q.record(new RpcException(1, "Internal"), upstream3) + then: + q.isFailed() + !q.isResolved() + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy new file mode 100644 index 00000000..0af237c2 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -0,0 +1,206 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.quorum + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.FilteredApis +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import reactor.core.publisher.Mono +import reactor.test.StepVerifier +import spock.lang.Specification + +import java.time.Duration + +class QuorumRpcReaderSpec extends Specification { + + def "always-quorum - get the result if ok"() { + setup: + def up = Mock(Upstream) { + _ * isAvailable() >> true + 1 * getApi() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(JsonRpcResponse.ok("1")) + } + } + def apis = new FilteredApis( + [up], Selector.empty + ) + def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) + + when: + def act = reader.read(new JsonRpcRequest("eth_test", [])) + .map { + new String(it.value) + } + + then: + StepVerifier.create(act) + .expectNext("1") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "always-quorum - retry upstream error"() { + setup: + def up = Mock(Upstream) { + _ * isAvailable() >> true + _ * getApi() >> Mock(Reader) { + 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ + Mono.just(JsonRpcResponse.error(1, "test")), + Mono.just(JsonRpcResponse.ok("1")) + ] + } + } + def apis = new FilteredApis( + [up], Selector.empty + ) + def reader = new QuorumRpcReader(apis, new AlwaysQuorum()) + + when: + def act = reader.read(new JsonRpcRequest("eth_test", [])) + .map { + new String(it.value) + } + + then: + StepVerifier.create(act) + .expectNext("1") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "non-empty-quorum - get the second result if first is null"() { + setup: + def up = Mock(Upstream) { + _ * isAvailable() >> true + _ * getApi() >> Mock(Reader) { + 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ + Mono.just(JsonRpcResponse.ok("null")), + Mono.just(JsonRpcResponse.ok("1")) + ] + } + } + def apis = new FilteredApis( + [up], Selector.empty + ) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + + when: + def act = reader.read(new JsonRpcRequest("eth_test", [])) + .map { + new String(it.value) + } + + then: + StepVerifier.create(act) + .expectNext("1") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + + def "non-empty-quorum - get the second result if first is error"() { + setup: + def up = Mock(Upstream) { + _ * isAvailable() >> true + _ * getApi() >> Mock(Reader) { + 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ + Mono.just(JsonRpcResponse.error(1, "test")), + Mono.just(JsonRpcResponse.ok("1")) + ] + } + } + def apis = new FilteredApis( + [up], Selector.empty + ) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + + when: + def act = reader.read(new JsonRpcRequest("eth_test", [])) + .map { + new String(it.value) + } + + then: + StepVerifier.create(act) + .expectNext("1") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "non-empty-quorum - get the third result if first two are not ok"() { + setup: + def up = Mock(Upstream) { + _ * isAvailable() >> true + _ * getApi() >> Mock(Reader) { + 3 * read(new JsonRpcRequest("eth_test", [])) >>> [ + Mono.just(JsonRpcResponse.ok("null")), + Mono.just(JsonRpcResponse.error(1, "test")), + Mono.just(JsonRpcResponse.ok("1")) + ] + } + } + def apis = new FilteredApis( + [up], Selector.empty + ) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + + when: + def act = reader.read(new JsonRpcRequest("eth_test", [])) + .map { + new String(it.value) + } + + then: + StepVerifier.create(act) + .expectNext("1") + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "non-empty-quorum - no result if all failed"() { + setup: + def up = Mock(Upstream) { + _ * isAvailable() >> true + _ * getApi() >> Mock(Reader) { + 3 * read(new JsonRpcRequest("eth_test", [])) >>> [ + Mono.just(JsonRpcResponse.ok("null")), + Mono.just(JsonRpcResponse.error(1, "test")), + Mono.just(JsonRpcResponse.ok("null")) + ] + } + } + def apis = new FilteredApis( + [up], Selector.empty + ) + def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + + when: + def act = reader.read(new JsonRpcRequest("eth_test", [])) + .map { + new String(it.value) + } + + then: + StepVerifier.create(act) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy index 6c99483d..f985b0fc 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy @@ -89,5 +89,10 @@ class ValueAwareQuorumSpec extends Specification { byte[] getResult() { return new byte[0] } + + @Override + boolean isFailed() { + return false + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 13c06663..d43df755 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -19,6 +19,8 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.quorum.BroadcastQuorum +import io.emeraldpay.dshackle.quorum.QuorumReaderFactory +import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.quorum.AlwaysQuorum @@ -96,126 +98,43 @@ class NativeCallSpec extends Specification { def "Quorum is applied"() { setup: - def quorum = Spy(new AlwaysQuorum()) - def upstreams = Stub(MultistreamHolder) - def apiMock = TestingCommons.api() + def quorum = new AlwaysQuorum() - apiMock.answer("eth_test", [], "foo") - - def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), - Selector.empty, quorum, + def nativeCall = new NativeCall(Stub(MultistreamHolder), TestingCommons.objectMapper()) + nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { + 1 * create(_, _) >> Mock(Reader) { + 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, 1)) + } + } + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_test", [])) when: - def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) + def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1)) def act = objectMapper.readValue(resp.payload, Object) then: act == "foo" - 1 * quorum.record(_, _) - 1 * quorum.getResult() - } - - def "Quorum may return not first received value"() { - setup: - def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) - - def upstreams = Stub(MultistreamHolder) - def apiMock = TestingCommons.api() - - apiMock.answerOnce("eth_test", [], null) - apiMock.answerOnce("eth_test", [], "bar") - apiMock.answerOnce("eth_test", [], null) - - def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), - Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_test", [])) - - - when: - def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) - def act = objectMapper.readValue(resp.payload, Object) - then: - act == "bar" - 2 * quorum.record(_, _) - 1 * quorum.getResult() - } - - def "Have pause between repeats"() { - setup: - def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) - - def upstreams = Stub(MultistreamHolder) - def apiMock = TestingCommons.api() - - apiMock.answerOnce("eth_test", [], null) - apiMock.answerOnce("eth_test", [], "bar") - - def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), - Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_test", [])) - - - when: - def t1 = System.currentTimeMillis() - def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) - def delta = System.currentTimeMillis() - t1 - def act = objectMapper.readValue(resp.payload, Object) - then: - delta > 95 // should be 100, but sometimes gives less ??? - act == "bar" - } - - def "One call has no pause"() { - setup: - def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) - - def upstreams = Stub(MultistreamHolder) - ReactorRpcClient rpcClient = Stub(ReactorRpcClient) - def apiMock = TestingCommons.api() - - apiMock.answerOnce("eth_test", [], "bar") - - def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), - Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_test", [])) - - - when: - def t1 = System.currentTimeMillis() - nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) - def delta = System.currentTimeMillis() - t1 - then: - delta < 50 } def "Returns error if no quorum"() { setup: - def quorum = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) + def quorum = new AlwaysQuorum() - def upstreams = Stub(MultistreamHolder) - ReactorRpcClient rpcClient = Stub(ReactorRpcClient) - def apiMock = TestingCommons.api() - - apiMock.answer("eth_test", [], null, 3) - apiMock.answerOnce("eth_test", [], "foo") - - def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), Selector.empty, quorum, + def nativeCall = new NativeCall(Stub(MultistreamHolder), TestingCommons.objectMapper()) + nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { + 1 * create(_, _) >> Mock(Reader) { + 1 * read(_) >> Mono.empty() + } + } + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_test", [])) - 3 * quorum.record(_, _) - 1 * quorum.getResult() - when: def resp = nativeCall.executeOnRemote(call) then: StepVerifier.create(resp) - .expectErrorMatches({t -> t instanceof NativeCall.CallFailure && t.id == 1}) - .verify(Duration.ofSeconds(1)) + .expectErrorMatches({ t -> t instanceof NativeCall.CallFailure && t.id == 1 }) + .verify(Duration.ofSeconds(1)) } def "Packs call exception into response with id"() { @@ -351,59 +270,4 @@ class NativeCallSpec extends Specification { 1 * cacheMock.execute(10, "eth_test", []) >> Mono.just('{"result": "foo"}'.bytes) new String(act.block().payload) == '{"result": "foo"}' } - - def "Retries on error"() { - setup: - def quorum = Spy(new AlwaysQuorum()) - - def upstreams = Stub(MultistreamHolder) - def apiMock = TestingCommons.api() - - apiMock.answer("eth_test", [], null, 1, new TimeoutException("test 1")) - apiMock.answer("eth_test", [], null, 1, new TimeoutException("test 2")) - apiMock.answerOnce("eth_test", [], "bar") - - def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), - Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_test", [])) - - - when: - def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) - def act = objectMapper.readValue(resp.payload, Object) - then: - act == "bar" - 1 * quorum.record(_, _) - 1 * quorum.getResult() - } - - def "Send raw retries 3 times"() { - setup: - def quorum = Spy(new BroadcastQuorum(TestingCommons.objectMapper(), 3)) - - def upstreams = Stub(MultistreamHolder) - def apiMock = TestingCommons.api() - - apiMock.answer("eth_sendRawTransaction", ["0x1234"], - "0x4b66b555df9faed6f0711f2104d183736c8e2dc7434626dd2622e243f041d41b", 1) - apiMock.answer("eth_sendRawTransaction", ["0x1234"], null, 10, - new RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Transaction with the same hash was already imported")) -// apiMock.answer("eth_sendRawTransaction", ["0x1234"], -// new RpcResponseError(RpcResponseError.CODE_INVALID_REQUEST, "Transaction with the same hash was already imported"), 10) - - def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) - def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), - Selector.empty, quorum, - new NativeCall.ParsedCallDetails("eth_sendRawTransaction", ["0x1234"])) - - - when: - def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) - def act = objectMapper.readValue(resp.payload, Object) - then: - act == "0x4b66b555df9faed6f0711f2104d183736c8e2dc7434626dd2622e243f041d41b" - 1 * quorum.record(_ as byte[], _) - 2 * quorum.record(_ as RpcException, _) - } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index d449aae6..1eba282f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -68,9 +68,9 @@ class FilteredApisSpec extends Specification { iter.request(10) then: StepVerifier.create(iter) - .expectNext(upstreams[0].api) - .expectNext(upstreams[2].api) - .expectNext(upstreams[3].api) + .expectNext(upstreams[0]) + .expectNext(upstreams[2]) + .expectNext(upstreams[3]) .expectComplete() .verify(Duration.ofSeconds(1)) @@ -79,9 +79,9 @@ class FilteredApisSpec extends Specification { iter.request(10) then: StepVerifier.create(iter) - .expectNext(upstreams[2].api) - .expectNext(upstreams[3].api) - .expectNext(upstreams[0].api) + .expectNext(upstreams[2]) + .expectNext(upstreams[3]) + .expectNext(upstreams[0]) .expectComplete() .verify(Duration.ofSeconds(1)) @@ -90,12 +90,12 @@ class FilteredApisSpec extends Specification { iter.request(10) then: StepVerifier.create(iter) - .expectNext(upstreams[2].api) - .expectNext(upstreams[3].api) - .expectNext(upstreams[0].api) - .expectNext(upstreams[2].api) - .expectNext(upstreams[3].api) - .expectNext(upstreams[0].api) + .expectNext(upstreams[2]) + .expectNext(upstreams[3]) + .expectNext(upstreams[0]) + .expectNext(upstreams[2]) + .expectNext(upstreams[3]) + .expectNext(upstreams[0]) .expectComplete() .verify(Duration.ofSeconds(1)) } @@ -154,13 +154,13 @@ class FilteredApisSpec extends Specification { apis.request(10) return apis }) - .expectNext(api1, api2).as("Batch 1") + .expectNext(up1, up2).as("Batch 1") .expectNoEvent(Duration.ofMillis(100)).as("Wait 1") - .expectNext(api1, api2).as("Batch 2") - .expectNoEvent(Duration.ofMillis(400)).as("Wait 2") - .expectNext(api1, api2).as("Batch 3") - .expectNoEvent(Duration.ofMillis(900)).as("Wait 3") - .expectNext(api1, api2).as("Batch 4") + .expectNext(up1, up2).as("Batch 2") + .expectNoEvent(Duration.ofMillis(400)).as("Wait 2") + .expectNext(up1, up2).as("Batch 3") + .expectNoEvent(Duration.ofMillis(900)).as("Wait 3") + .expectNext(up1, up2).as("Batch 4") .expectComplete() .verify(Duration.ofSeconds(10)) } @@ -178,7 +178,7 @@ class FilteredApisSpec extends Specification { act.request(10) then: StepVerifier.create(act) - .expectNext(apis[2], apis[3], apis[4], apis[5], apis[0], apis[1]) + .expectNext(ups[2], ups[3], ups[4], ups[5], ups[0], ups[1]) .expectComplete() .verify(Duration.ofSeconds(1)) }