StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: individual class for quorum based requests

Browse Source
This commit is contained in:
Igor Artamonov
2020-05-15 21:06:32 -04:00
parent 73f55d8d81
commit 114f109722
19 changed files with 611 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
View File

@@ -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

4
src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt
View File

@@ -37,6 +37,10 @@ open class BroadcastQuorum(
return calls >= quorum
}
override fun isFailed(): Boolean {
return false
}
override fun getResult(): ByteArray? {
return result
}

2
src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
View File

@@ -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?

8
src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt
View File

@@ -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++
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt
View File

@@ -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()

4
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
View File

@@ -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) {

38
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt Normal file
View File

@@ -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<JsonRpcRequest, QuorumRpcReader.Result>
class Default : QuorumReaderFactory {
override fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result> {
return QuorumRpcReader(apis, quorum)
}
}
}

105
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt Normal file
View File

@@ -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<JsonRpcRequest, QuorumRpcReader.Result> {
companion object {
private val log = LoggerFactory.getLogger(QuorumRpcReader::class.java)
}
override fun read(key: JsonRpcRequest): Mono<QuorumRpcReader.Result> {
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
)
}

61
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -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<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
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)

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt
View File

@@ -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<Reader<JsonRpcRequest, JsonRpcResponse>> {
interface ApiSource : Publisher<Upstream> {
fun resolve()
fun request(tries: Int)

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt
View File

@@ -78,7 +78,7 @@ class FilteredApis(
return Duration.ofMillis(time)
}
override fun subscribe(subscriber: Subscriber<in Reader<JsonRpcRequest, JsonRpcResponse>>) {
override fun subscribe(subscriber: Subscriber<in Upstream>) {
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)

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -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")))
}

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt
View File

@@ -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 {

133
src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy Normal file
View File

@@ -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()
}
}

32
src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy
View File

@@ -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()
}
}

206
src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy Normal file
View File

@@ -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))
}
}

5
src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy
View File

@@ -89,5 +89,10 @@ class ValueAwareQuorumSpec extends Specification {
byte[] getResult() {
return new byte[0]
}
@Override
boolean isFailed() {
return false
}
}
}

178
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -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, _)
}
}

38
src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
View File

@@ -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))
}