Maximum value quorum for BroadcastReader (#272)

Fixes eth_getTransactionCount when tx did not propagated to all nodes

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<Upstream>()
 
-    override fun init(head: Head) {
-    }
-
     override fun isResolved(): Boolean {
         return resolved
     }

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(
+open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum<String>(String::class.java) {
-    val quorum: Int = 3
-) : CallQuorum, ValueAwareQuorum<String>(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"
     }
 }

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
 

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

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<Upstream>()
 
-    override fun init(head: Head) {
-    }
-
     override fun isResolved(): Boolean {
         return !isFailed() && result.get() != null
     }

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<String>() // 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

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<Upstream>,
     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<Result> {
-        return Mono.just(upstreams)
+        return Flux.fromIterable(upstreams)
-            .map { ups ->
+            .filter { internalMatcher.matches(it) }
-                ups.filter { internalMatcher.matches(it) }.map { execute(key, it) }
+            .flatMap { up ->
-            }.flatMap {
+                execute(key, up)
-                Mono.zip(it) { responses ->
+            }.map {
-                    analyzeResponses(
+                if (it.jsonRpcResponse.hasResult()) {
-                        key,
+                    val sig = getSignature(key, it.jsonRpcResponse, it.upstream.getId())
-                        getJsonRpcResponses(key.method, responses)
+                    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 ->
+                    Mono.just(res)
-                    log.error("Broadcast error: ${err.message}")
+                } else {
-                    Mono.error(handleError(null, 0, null))
+                    Mono.error(handleError(quorum.getError(), key.id, 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)
-                    }
                 }
             }
     }
 
-    private fun analyzeResponses(key: JsonRpcRequest, jsonRpcResponses: List<BroadcastResponse>): BroadcastResult {
-        val errors = mutableListOf<JsonRpcError>()
-        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<Any>) =
-        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)
-    }
 }

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") {
+            if (data.method == "eth_sendRawTransaction" || data.method == "eth_getTransactionCount") {
-                return BroadcastReader(data.multistream.getAll(), data.matcher, data.signer, data.tracer)
+                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)

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

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

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

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

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

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 {
+                .expectErrorMessage("Unhandled Upstream error")
-                    it.value == new byte[0]
-                }
-                .expectComplete()
                 .verify(Duration.ofSeconds(3))
     }
 }