From 0002d2c7894329762539431df5133d8fa554ea35 Mon Sep 17 00:00:00 2001 From: Vyacheslav Date: Mon, 7 Aug 2023 15:09:40 +0300 Subject: [PATCH] Maximum value quorum for BroadcastReader (#272) Fixes eth_getTransactionCount when tx did not propagated to all nodes --- .../dshackle/quorum/AlwaysQuorum.kt | 4 - .../dshackle/quorum/BroadcastQuorum.kt | 22 +---- .../emeraldpay/dshackle/quorum/CallQuorum.kt | 4 - .../dshackle/quorum/MaximumValueQuorum.kt | 60 ++++++++++++ .../dshackle/quorum/NotLaggingQuorum.kt | 4 - .../dshackle/quorum/NotNullQuorum.kt | 4 - .../dshackle/reader/BroadcastReader.kt | 96 +++++++------------ .../dshackle/reader/RpcReaderFactory.kt | 4 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 2 - .../upstream/calls/DefaultEthereumMethods.kt | 5 +- .../quorum/BroadcastQuorumSpec.groovy | 31 +----- .../quorum/MaximumValueQuorumSpec.groovy | 77 +++++++++++++++ .../quorum/ValueAwareQuorumSpec.groovy | 5 - .../reader/BroadcastReaderSpec.groovy | 16 ++-- 14 files changed, 191 insertions(+), 143 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt index 5211765d..dddfe18a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle.quorum -import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException @@ -30,9 +29,6 @@ open class AlwaysQuorum : CallQuorum { private var sig: ResponseSigner.Signature? = null private val resolvers = ArrayList() - override fun init(head: Head) { - } - override fun isResolved(): Boolean { return resolved } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt index ec3faf3f..be126e61 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt @@ -16,28 +16,21 @@ */ package io.emeraldpay.dshackle.quorum -import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.signature.ResponseSigner -open class BroadcastQuorum( - val quorum: Int = 3 -) : CallQuorum, ValueAwareQuorum(String::class.java) { +open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum(String::class.java) { private var result: ByteArray? = null private var txid: String? = null - private var calls = 0 private var sig: ResponseSigner.Signature? = null - override fun init(head: Head) { - } - override fun isResolved(): Boolean { - return calls >= quorum && txid != null + return result != null } override fun isFailed(): Boolean { - return calls >= quorum && getError() != null + return result == null } override fun getResult(): ByteArray? { @@ -54,7 +47,6 @@ open class BroadcastQuorum( signature: ResponseSigner.Signature?, upstream: Upstream ) { - calls++ if (txid == null && responseValue != null) { txid = responseValue sig = signature @@ -68,16 +60,10 @@ open class BroadcastQuorum( signature: ResponseSigner.Signature?, upstream: Upstream ) { - // can be "message: known transaction: TXID", "Transaction with the same hash was already imported" or "message: Nonce too low" - calls++ - if (result == null) { - result = response - sig = signature - } resolvers.add(upstream) } override fun toString(): String { - return "Quorum: Broadcast to $quorum upstreams" + return "Quorum: Broadcast to upstreams" } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt index 5fc34f87..735ab232 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt @@ -16,16 +16,12 @@ */ package io.emeraldpay.dshackle.quorum -import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.signature.ResponseSigner interface CallQuorum { - - fun init(head: Head) - fun isResolved(): Boolean fun isFailed(): Boolean diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt new file mode 100644 index 00000000..4ad0d6b4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt @@ -0,0 +1,60 @@ +package io.emeraldpay.dshackle.quorum + +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.signature.ResponseSigner +import io.emeraldpay.etherjar.hex.HexQuantity + +class MaximumValueQuorum : CallQuorum, ValueAwareQuorum(String::class.java) { + private var max: Long? = null + private var result: ByteArray? = null + private var sig: ResponseSigner.Signature? = null + + override fun isResolved(): Boolean { + return result != null + } + + override fun isFailed(): Boolean { + return result == null + } + + override fun getResult(): ByteArray? { + return result + } + + override fun getSignature(): ResponseSigner.Signature? { + return sig + } + override fun recordValue( + response: ByteArray, + responseValue: String?, + signature: ResponseSigner.Signature?, + upstream: Upstream + ) { + val value = responseValue?.let { str -> + HexQuantity.from(str).value.toLong() + } + if (value != null) { + max = max.let { + if (it == null || it < value) { + sig = signature + resolvers.clear() + result = response + value + } else { + it + } + } + } + } + + override fun recordError( + response: ByteArray?, + errorMessage: String?, + signature: ResponseSigner.Signature?, + upstream: Upstream + ) { + if (max == null) { + resolvers.add(upstream) + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index 1e71b774..04fb1cc0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle.quorum -import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException @@ -36,9 +35,6 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum { private var sig: ResponseSigner.Signature? = null private val resolvers = ArrayList() - override fun init(head: Head) { - } - override fun isResolved(): Boolean { return !isFailed() && result.get() != null } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt index 6f06c91d..e07df613 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt @@ -1,7 +1,6 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException @@ -15,9 +14,6 @@ class NotNullQuorum : CallQuorum { private var allFailed = true private val seenUpstreams = HashSet() // just to prevent calling retry upstreams in FilteredApis - override fun init(head: Head) { - } - override fun isResolved(): Boolean = result != null override fun isFailed(): Boolean = rpcError != null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt index d7138291..ff7935dc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt @@ -2,14 +2,16 @@ package io.emeraldpay.dshackle.reader import io.emeraldpay.dshackle.commons.BROADCAST_READER import io.emeraldpay.dshackle.commons.SPAN_REQUEST_UPSTREAM_ID +import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import org.slf4j.LoggerFactory import org.springframework.cloud.sleuth.Tracer +import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.util.concurrent.atomic.AtomicInteger @@ -17,6 +19,7 @@ class BroadcastReader( private val upstreams: List, matcher: Selector.Matcher, signer: ResponseSigner?, + private val quorum: CallQuorum, private val tracer: Tracer ) : RpcReader(signer) { private val internalMatcher = Selector.MultiMatcher( @@ -32,61 +35,38 @@ class BroadcastReader( } override fun read(key: JsonRpcRequest): Mono { - return Mono.just(upstreams) - .map { ups -> - ups.filter { internalMatcher.matches(it) }.map { execute(key, it) } - }.flatMap { - Mono.zip(it) { responses -> - analyzeResponses( - key, - getJsonRpcResponses(key.method, responses) + return Flux.fromIterable(upstreams) + .filter { internalMatcher.matches(it) } + .flatMap { up -> + execute(key, up) + }.map { + if (it.jsonRpcResponse.hasResult()) { + val sig = getSignature(key, it.jsonRpcResponse, it.upstream.getId()) + quorum.record(it.jsonRpcResponse.getResult(), sig, it.upstream) + } else { + val err = JsonRpcException(JsonRpcResponse.NumberId(key.id), it.jsonRpcResponse.error!!, it.upstream.getId()) + quorum.record(err, null, it.upstream) + } + quorum + }.onErrorResume { err -> + log.error("Broadcast error: ${err.message}") + Mono.error(handleError(null, 0, null)) + }.collectList() + .flatMap { + if (quorum.isResolved()) { + val res = Result( + quorum.getResult()!!, + quorum.getSignature(), + upstreams.size, + quorum.getResolvedBy().first() ) - }.onErrorResume { err -> - log.error("Broadcast error: ${err.message}") - Mono.error(handleError(null, 0, null)) - }.flatMap { broadcastResult -> - if (broadcastResult.result != null) { - Mono.just( - Result(broadcastResult.result, broadcastResult.signature, 0, null) - ) - } else { - val err = handleError(broadcastResult.error, key.id, null) - Mono.error(err) - } + Mono.just(res) + } else { + Mono.error(handleError(quorum.getError(), key.id, null)) } } } - private fun analyzeResponses(key: JsonRpcRequest, jsonRpcResponses: List): BroadcastResult { - val errors = mutableListOf() - jsonRpcResponses.forEach { - val response = it.jsonRpcResponse - if (response.hasResult()) { - val signature = getSignature(key, response, it.upstreamId) - return BroadcastResult(response.getResult(), null, signature) - } else if (response.hasError()) { - errors.add(response.error!!) - } - } - - val error = errors.takeIf { it.isNotEmpty() }?.get(0) - - return BroadcastResult(error) - } - - private fun getJsonRpcResponses(method: String, responses: Array) = - responses - .map { response -> - (response as BroadcastResponse) - .also { r -> - if (r.jsonRpcResponse.hasResult()) { - log.info( - "Response for $method from upstream ${r.upstreamId}: ${String(r.jsonRpcResponse.getResult())}" - ) - } - } - } - private fun execute( key: JsonRpcRequest, upstream: Upstream @@ -95,24 +75,16 @@ class BroadcastReader( upstream.getIngressReader(), tracer, BROADCAST_READER, mapOf(SPAN_REQUEST_UPSTREAM_ID to upstream.getId()) ) .read(key) - .map { BroadcastResponse(it, upstream.getId()) } + .map { BroadcastResponse(it, upstream) } .onErrorResume { log.warn("Error during execution ${key.method} from upstream ${upstream.getId()} with message - ${it.message}") Mono.just( - BroadcastResponse(JsonRpcResponse(null, getError(key, it).error), upstream.getId()) + BroadcastResponse(JsonRpcResponse(null, getError(key, it).error), upstream) ) } private class BroadcastResponse( val jsonRpcResponse: JsonRpcResponse, - val upstreamId: String + val upstream: Upstream ) - - private class BroadcastResult( - val result: ByteArray?, - val error: JsonRpcError?, - val signature: ResponseSigner.Signature? - ) { - constructor(error: JsonRpcError?) : this(null, error, null) - } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt index b109fb13..dfd585da 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt @@ -62,8 +62,8 @@ interface RpcReaderFactory { class Default : RpcReaderFactory { override fun create(data: RpcReaderData): RpcReader { - if (data.method == "eth_sendRawTransaction") { - return BroadcastReader(data.multistream.getAll(), data.matcher, data.signer, data.tracer) + if (data.method == "eth_sendRawTransaction" || data.method == "eth_getTransactionCount") { + return BroadcastReader(data.multistream.getAll(), data.matcher, data.signer, data.quorum, data.tracer) } val apis = data.multistream.getApiSource(data.matcher) return QuorumRpcReader(apis, data.quorum, data.signer, data.tracer) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 1d01de93..03b7b5ff 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -324,8 +324,6 @@ open class NativeCall( .forLabels(Selector.convertToMatcher(request.selector)) val callQuorum = availableMethods.createQuorumFor(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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt index 00a140dc..8fbf2544 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt @@ -19,7 +19,9 @@ package io.emeraldpay.dshackle.upstream.calls import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.quorum.AlwaysQuorum +import io.emeraldpay.dshackle.quorum.BroadcastQuorum import io.emeraldpay.dshackle.quorum.CallQuorum +import io.emeraldpay.dshackle.quorum.MaximumValueQuorum import io.emeraldpay.dshackle.quorum.NotLaggingQuorum import io.emeraldpay.dshackle.quorum.NotNullQuorum import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods.HardcodedData.Companion.createHardcodedData @@ -197,9 +199,10 @@ class DefaultEthereumMethods( possibleNotIndexedMethods.contains(method) -> NotNullQuorum() specialMethods.contains(method) -> { when (method) { - "eth_getTransactionCount" -> AlwaysQuorum() + "eth_getTransactionCount" -> MaximumValueQuorum() "eth_getBalance" -> AlwaysQuorum() "eth_blockNumber" -> NotLaggingQuorum(0) + "eth_sendRawTransaction" -> BroadcastQuorum() else -> AlwaysQuorum() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy index ee770db4..022d9130 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy @@ -29,58 +29,43 @@ class BroadcastQuorumSpec extends Specification { def "Resolved with first after 3 tries"() { setup: - def q = Spy(new BroadcastQuorum(3)) + def q = Spy(new BroadcastQuorum()) def upstream1 = Stub(Upstream) def upstream2 = Stub(Upstream) def upstream3 = Stub(Upstream) - when: - q.init(Stub(Head)) - then: - !q.isResolved() - when: q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream1) then: - !q.isResolved() 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) when: q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2) then: - !q.isResolved() 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) when: q.record(new JsonRpcException(1, "Nonce too low"), null, upstream3) then: 1 * q.recordError(_, _, _, _) - q.isResolved() objectMapper.readValue(q.result, Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c" } def "Remembers first response"() { setup: - def q = Spy(new BroadcastQuorum(3)) + def q = Spy(new BroadcastQuorum()) def upstream1 = Stub(Upstream) def upstream2 = Stub(Upstream) def upstream3 = Stub(Upstream) - when: - q.init(Stub(Head)) - then: - !q.isResolved() - when: q.record(new JsonRpcException(1, "Internal error"), null, upstream1) then: - !q.isResolved() 1 * q.recordError(_, _, _, _) when: q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2) then: - !q.isResolved() 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) when: @@ -93,21 +78,11 @@ class BroadcastQuorumSpec extends Specification { def "Failed if error received 3+ times"() { setup: - def quorum = new BroadcastQuorum(3) + def quorum = new BroadcastQuorum() def up = Stub(Upstream) when: quorum.record(new JsonRpcException(1, "test 1"), null, up) - then: - !quorum.isFailed() - !quorum.isResolved() - - when: quorum.record(new JsonRpcException(1, "test 2"), null, up) - then: - !quorum.isFailed() - !quorum.isResolved() - - when: quorum.record(new JsonRpcException(1, "test 3"), null, up) then: quorum.isFailed() diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy new file mode 100644 index 00000000..1abde219 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy @@ -0,0 +1,77 @@ +package io.emeraldpay.dshackle.quorum + +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException +import spock.lang.Specification + +class MaximumValueQuorumSpec extends Specification { + def "selects maximum from 3 values"() { + setup: + def up = Mock(Upstream) { + getId() >> "id" + } + def up1 = Mock(Upstream) { + getId() >> "id1" + } + def up2 = Mock(Upstream) { + getId() >> "id2" + } + when: + def quorum = new MaximumValueQuorum() + quorum.record('"0x137"'.bytes, null, up) + quorum.record('"0x138"'.bytes, null, up1) + quorum.record('"0x139"'.bytes, null, up2) + then: + quorum.result == '"0x139"'.bytes + quorum.resolvedBy.size() == 1 + quorum.isResolved() + quorum.resolvedBy.contains(up2) + } + + def "selects maximum from 2 values and an error"() { + setup: + def up = Mock(Upstream) { + getId() >> "id" + } + def up1 = Mock(Upstream) { + getId() >> "id1" + } + def up2 = Mock(Upstream) { + getId() >> "id2" + } + when: + def quorum = new MaximumValueQuorum() + quorum.record('"0x137"'.bytes, null, up) + quorum.record('"0x138"'.bytes, null, up1) + quorum.record(new JsonRpcException(10, "error"), null, up2) + then: + quorum.result == '"0x138"'.bytes + quorum.isResolved() + quorum.resolvedBy.size() == 1 + quorum.resolvedBy.contains(up1) + } + + def "returns error is all error"() { + setup: + def up = Mock(Upstream) { + getId() >> "id" + } + def up1 = Mock(Upstream) { + getId() >> "id1" + } + def up2 = Mock(Upstream) { + getId() >> "id2" + } + when: + def quorum = new MaximumValueQuorum() + quorum.record(new JsonRpcException(10, "error1"), null, up) + quorum.record(new JsonRpcException(10, "error2"), null, up1) + quorum.record(new JsonRpcException(10, "error3"), null, up2) + then: + !quorum.isResolved() + quorum.isFailed() + quorum.error == new JsonRpcException(10, "error3").error + quorum.resolvedBy.size() == 3 + } + +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy index d008040a..ceec57bf 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy @@ -77,11 +77,6 @@ class ValueAwareQuorumSpec extends Specification { } - @Override - void init(@NotNull Head head) { - - } - @Override boolean isResolved() { return false diff --git a/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy index 29a2ccd7..e06a2b81 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.reader +import io.emeraldpay.dshackle.quorum.BroadcastQuorum import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException @@ -41,7 +42,7 @@ class BroadcastReaderSpec extends Specification { Mono.just(new JsonRpcResponse(result, null)) } } - def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, Stub(Tracer)) + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) then: @@ -79,7 +80,7 @@ class BroadcastReaderSpec extends Specification { 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> Mono.error(new JsonRpcException(1, "too low")) } } - def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, Stub(Tracer)) + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) then: @@ -112,7 +113,7 @@ class BroadcastReaderSpec extends Specification { 0 * getId() >> "id" 0 * getIngressReader() >> Mock(Reader) } - def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, Stub(Tracer)) + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) then: @@ -150,7 +151,7 @@ class BroadcastReaderSpec extends Specification { Mono.error(new JsonRpcException(1, "too low")) } } - def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, Stub(Tracer)) + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer)) when: def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) then: @@ -176,17 +177,14 @@ class BroadcastReaderSpec extends Specification { 0 * getId() >> "id" 0 * getIngressReader() >> Mock(Reader) } - def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, Stub(Tracer)) + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer)) when: def act = reader .read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) .switchIfEmpty(Mono.just(new RpcReader.Result(new byte[0], null, 0, null))) then: StepVerifier.create(act) - .expectNextMatches { - it.value == new byte[0] - } - .expectComplete() + .expectErrorMessage("Unhandled Upstream error") .verify(Duration.ofSeconds(3)) } }