diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 123bdfb5..fbe72bf4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -130,8 +130,9 @@ class QuorumRpcReader( .flatMap { response -> response.requireResult() .transform(withSignature(api, key, response)) - .transform(withErrorResume(api, key)) } + // must catch not only the processing of a response but also errors thrown from the .read() call + .transform(withErrorResume(api, key)) .map { Tuples.of(it.t1, it.t2, api) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index 1a65b65f..c69450ca 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.grpc.Chain import io.emeraldpay.etherjar.rpc.RpcException import reactor.core.publisher.Mono @@ -62,7 +63,7 @@ class QuorumRpcReaderSpec extends Specification { .verify(Duration.ofSeconds(1)) } - def "always-quorum - return upstream error"() { + def "always-quorum - return upstream error returned"() { setup: def api = Mock(Reader) { 1 * read(new JsonRpcRequest("eth_test", [])) >>> [ @@ -94,6 +95,43 @@ class QuorumRpcReaderSpec extends Specification { .verify(Duration.ofSeconds(1)) } + def "always-quorum - return upstream error thrown"() { + setup: + def api = Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_test", [])) >>> [ + Mono.error( + new RpcException( + RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, + "test-123" + ) + ) + ] + } + def up = Mock(Upstream) { + _ * isAvailable() >> true + _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY + _ * getApi() >> api + } + def apis = new FilteredApis( + Chain.ETHEREUM, + [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) + .expectErrorMatches { + it instanceof JsonRpcException && ((JsonRpcException) it).error.message == "test-123" + } + .verify(Duration.ofSeconds(1)) + } + def "non-empty-quorum - get the second result if first is null"() { setup: def up = Mock(Upstream) {