Add NotNullQuorum for tx methods (#208)

src/main/kotlin/io/emeraldpay/dshackle/Global.kt

@@ -39,6 +39,8 @@ class Global {
 
     companion object {
 
+        val nullValue: ByteArray = "null".toByteArray()
+
         var metricsExtended = false
 
         val chainNames = mapOf(

src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt

@@ -1,82 +0,0 @@
-/**
- * Copyright (c) 2020 EmeraldPay, Inc
- * Copyright (c) 2019 ETCDEV GmbH
- *
- * 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.upstream.Head
-import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
-
-open class NonEmptyQuorum(
-    val maxTries: Int = 3
-) : CallQuorum, ValueAwareQuorum<Any>(Any::class.java) {
-
-    private var result: ByteArray? = null
-    private var tries: Int = 0
-    private var sig: ResponseSigner.Signature? = null
-    private var providedUpstreamId: String? = null
-    override fun init(head: Head) {
-    }
-
-    override fun isResolved(): Boolean {
-        return result != null
-    }
-
-    override fun isFailed(): Boolean {
-        return tries >= maxTries
-    }
-
-    override fun getSignature(): ResponseSigner.Signature? {
-        return sig
-    }
-
-    override fun getProvidedUpstreamId(): String? {
-        return providedUpstreamId
-    }
-
-    override fun recordValue(
-        response: ByteArray,
-        responseValue: Any?,
-        signature: ResponseSigner.Signature?,
-        upstream: Upstream,
-        providedUpstreamId: String?
-    ) {
-        tries++
-        if (responseValue != null) {
-            result = response
-            sig = signature
-            this.providedUpstreamId = providedUpstreamId
-        }
-    }
-
-    override fun getResult(): ByteArray? {
-        return result
-    }
-
-    override fun recordError(
-        response: ByteArray?,
-        errorMessage: String?,
-        sig: ResponseSigner.Signature?,
-        upstream: Upstream,
-        providedUpstreamId: String?
-    ) {
-        tries++
-    }
-
-    override fun toString(): String {
-        return "Quorum: Accept Non Error Result"
-    }
-}

src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt

@@ -0,0 +1,73 @@
+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
+import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
+
+class NotNullQuorum : CallQuorum {
+    private var sig: ResponseSigner.Signature? = null
+    private var providedUpstreamId: String? = null
+    private var result: ByteArray? = null
+    private var rpcError: JsonRpcError? = null
+    private val resolvers = ArrayList<Upstream>()
+    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
+
+    override fun record(
+        response: ByteArray,
+        signature: ResponseSigner.Signature?,
+        upstream: Upstream,
+        providedUpstreamId: String?
+    ): Boolean {
+        allFailed = false
+        val receivedNull = response.isEmpty() || Global.nullValue.contentEquals(response)
+        val upId = upstream.getId()
+        if (seenUpstreams.contains(upId) || !receivedNull) {
+            sig = signature
+            result = response
+            this.providedUpstreamId = providedUpstreamId
+            resolvers.add(upstream)
+            return true
+        }
+        seenUpstreams.add(upId)
+        return false
+    }
+
+    override fun record(error: JsonRpcException, signature: ResponseSigner.Signature?, upstream: Upstream) {
+        val upId = upstream.getId()
+        if (seenUpstreams.contains(upId)) {
+            if (allFailed) {
+                rpcError = error.error
+            } else {
+                result = Global.nullValue
+                resolvers.add(upstream)
+            }
+            sig = signature
+        }
+        seenUpstreams.add(upId)
+    }
+
+    override fun getSignature(): ResponseSigner.Signature? = sig
+
+    override fun getProvidedUpstreamId(): String? = providedUpstreamId
+
+    override fun getResult(): ByteArray? = result
+
+    override fun getError(): JsonRpcError? = rpcError
+
+    override fun getResolvedBy(): Collection<Upstream> = resolvers.toList()
+
+    override fun toString(): String {
+        return "Quorum: Not null"
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt

@@ -169,7 +169,7 @@ class QuorumRpcReader(
     fun <T> withErrorResume(api: Upstream, key: JsonRpcRequest): Function<Mono<T>, Mono<T>> {
         return Function { src ->
             src.onErrorResume { err ->
-                log.error("Error during call upstream ${api.getId()} with method $${key.method}", err)
+                log.error("Error during call upstream ${api.getId()} with method ${key.method}", err)
                 // when the call failed with an error we want to notify the quorum because
                 // it may use the error message or other details
                 //

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -22,6 +22,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.BlockchainType
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.Global.Companion.nullValue
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.commons.LOCAL_READER
 import io.emeraldpay.dshackle.commons.REMOTE_QUORUM_RPC_READER
@@ -78,8 +79,6 @@ open class NativeCall(
     private val log = LoggerFactory.getLogger(NativeCall::class.java)
     private val objectMapper: ObjectMapper = Global.objectMapper
 
-    private val nullValue: ByteArray = "null".toByteArray()
-
     private val localRouterEnabled = config.cache?.requestsCacheEnabled ?: true
     private val passthrough = config.passthrough
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt

@@ -19,8 +19,8 @@ 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.NonEmptyQuorum
 import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
+import io.emeraldpay.dshackle.quorum.NotNullQuorum
 import io.emeraldpay.etherjar.rpc.RpcException
 import java.util.Collections
 
@@ -62,7 +62,7 @@ class DefaultBitcoinMethods : CallMethods {
     override fun createQuorumFor(method: String): CallQuorum {
         return when {
             Collections.binarySearch(hardcodedMethods, method) >= 0 -> AlwaysQuorum()
-            Collections.binarySearch(anyResponseMethods, method) >= 0 -> NonEmptyQuorum()
+            Collections.binarySearch(anyResponseMethods, method) >= 0 -> NotNullQuorum()
             Collections.binarySearch(freshMethods, method) >= 0 -> NotLaggingQuorum(2)
             Collections.binarySearch(headVerifiedMethods, method) >= 0 -> NotLaggingQuorum(0)
             Collections.binarySearch(broadcastMethods, method) >= 0 -> BroadcastQuorum()

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.quorum.BroadcastQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
 import io.emeraldpay.dshackle.quorum.NonceQuorum
 import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
+import io.emeraldpay.dshackle.quorum.NotNullQuorum
 import io.emeraldpay.etherjar.rpc.RpcException
 
 /**
@@ -77,15 +78,18 @@ class DefaultEthereumMethods(
         "eth_estimateGas"
     )
 
+    private val possibleNotIndexedMethods = listOf(
+        "eth_getTransactionByHash",
+        "eth_getTransactionReceipt"
+    )
+
     private val firstValueMethods = listOf(
         "eth_getBlockTransactionCountByHash",
         "eth_getUncleCountByBlockHash",
         "eth_getBlockByHash",
         "eth_getBlockByNumber",
-        "eth_getTransactionByHash",
         "eth_getTransactionByBlockHashAndIndex",
         "eth_getTransactionByBlockNumberAndIndex",
-        "eth_getTransactionReceipt",
         "eth_getStorageAt",
         "eth_getCode",
         "eth_getUncleByBlockHashAndIndex",
@@ -127,6 +131,7 @@ class DefaultEthereumMethods(
     init {
         allowedMethods = anyResponseMethods +
             firstValueMethods +
+            possibleNotIndexedMethods +
             specialMethods +
             headVerifiedMethods -
             chainUnsupportedMethods(chain) +
@@ -141,6 +146,7 @@ class DefaultEthereumMethods(
             firstValueMethods.contains(method) -> AlwaysQuorum()
             anyResponseMethods.contains(method) -> NotLaggingQuorum(4)
             headVerifiedMethods.contains(method) -> NotLaggingQuorum(1)
+            possibleNotIndexedMethods.contains(method) -> NotNullQuorum()
             specialMethods.contains(method) -> {
                 when (method) {
                     "eth_getTransactionCount" -> NonceQuorum()

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt

@@ -19,8 +19,8 @@ package io.emeraldpay.dshackle.upstream.calls
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
-import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
 import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
+import io.emeraldpay.dshackle.quorum.NotNullQuorum
 import org.apache.commons.collections4.Factory
 import org.slf4j.LoggerFactory
 import java.io.IOException
@@ -66,7 +66,7 @@ class ManagedCallMethods(
         val quorum = when (quorumId) {
             "always" -> Factory<CallQuorum> { AlwaysQuorum() }
             "no-lag", "not-lagging", "no_lag", "not_lagging" -> Factory<CallQuorum> { NotLaggingQuorum(0) }
-            "not-empty", "not_empty", "non-empty", "non_empty" -> Factory<CallQuorum> { NonEmptyQuorum() }
+            "not-empty", "not_empty", "non-empty", "non_empty" -> Factory<CallQuorum> { NotNullQuorum() }
             else -> {
                 log.warn("Unknown quorum: $quorumId for custom method $method")
                 return

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt

@@ -76,9 +76,8 @@ class EthereumDirectReader(
         txReader = object : Reader<TransactionId, TxContainer> {
             override fun read(key: TransactionId): Mono<TxContainer> {
                 val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
-                return readWithQuorum(request)
+                return readWithQuorum(request) // retries were removed because we use NotNullQuorum which handle errors too
                     .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
-                    .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200)))
                     .flatMap { txbytes ->
                         val tx = objectMapper.readValue(txbytes, TransactionJson::class.java)
                         if (tx == null) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.Global.Companion.nullValue
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.JsonRpcReader
@@ -27,6 +28,7 @@ import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
+import reactor.kotlin.core.publisher.switchIfEmpty
 import java.math.BigInteger
 
 /**
@@ -88,7 +90,10 @@ class EthereumLocalReader(
                 } catch (e: IllegalArgumentException) {
                     throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
                 }
-                reader.txByHashAsCont().read(hash).map { it.json!! }
+                reader.txByHashAsCont()
+                    .read(hash)
+                    .map { it.json!! }
+                    .switchIfEmpty { Mono.just(nullValue) }
             }
             method == "eth_getBlockByHash" -> {
                 if (params.size != 2) {
@@ -120,7 +125,9 @@ class EthereumLocalReader(
                 } catch (e: IllegalArgumentException) {
                     throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
                 }
-                reader.receipts().read(hash)
+                reader.receipts()
+                    .read(hash)
+                    .switchIfEmpty { Mono.just(nullValue) }
             }
             else -> null
         }

src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy

@@ -1,129 +0,0 @@
-/**
- * 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.upstream.Head
-import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
-import spock.lang.Specification
-
-class NonEmptyQuorumSpec extends Specification {
-
-    def "Fail if too many errors"() {
-        setup:
-        def q = Spy(new NonEmptyQuorum(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 JsonRpcException(1, "Internal"), null, upstream1)
-        then:
-        !q.isResolved()
-        !q.isFailed()
-
-        when:
-        q.record(new JsonRpcException(1, "Internal"), null, upstream2)
-        then:
-        !q.isResolved()
-        !q.isFailed()
-
-        when:
-        q.record(new JsonRpcException(1, "Internal"), null, upstream3)
-        then:
-        q.isFailed()
-        !q.isResolved()
-        q.signature == null
-    }
-
-    def "Fail first if not error"() {
-        setup:
-        def q = Spy(new NonEmptyQuorum(3))
-        def upstream1 = Stub(Upstream)
-
-        when:
-        q.init(Stub(Head))
-        then:
-        !q.isResolved()
-        !q.isFailed()
-
-        when:
-        q.record('"0x11"'.bytes, null, upstream1, null)
-        then:
-        q.isResolved()
-        !q.isFailed()
-    }
-
-    def "Fail second if first is error"() {
-        setup:
-        def q = Spy(new NonEmptyQuorum(3))
-        def upstream1 = Stub(Upstream)
-        def upstream2 = Stub(Upstream)
-
-        when:
-        q.init(Stub(Head))
-        then:
-        !q.isResolved()
-        !q.isFailed()
-
-        when:
-        q.record(new JsonRpcException(1, "Internal"), null, upstream1)
-        then:
-        !q.isFailed()
-        !q.isResolved()
-        q.signature == null
-
-        when:
-        q.record('"0x11"'.bytes, null, upstream2, null)
-        then:
-        q.isResolved()
-        !q.isFailed()
-    }
-
-    def "Fail second if first is null"() {
-        setup:
-        def q = Spy(new NonEmptyQuorum(3))
-        def upstream1 = Stub(Upstream)
-
-        when:
-        q.init(Stub(Head))
-        then:
-        !q.isResolved()
-        !q.isFailed()
-
-        when:
-        q.record('null'.bytes, null, upstream1, null)
-        then:
-        !q.isFailed()
-        !q.isResolved()
-
-
-        when:
-        q.record('"0x11"'.bytes, null, upstream1, null)
-        then:
-        q.isResolved()
-        !q.isFailed()
-    }
-
-
-}

src/test/groovy/io/emeraldpay/dshackle/quorum/NotNullQuorumSpec.groovy

@@ -0,0 +1,93 @@
+package io.emeraldpay.dshackle.quorum
+
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
+import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
+import spock.lang.Specification
+
+class NotNullQuorumSpec extends Specification {
+
+    def "Resolves if attempts are exhausted and response is null"() {
+        setup:
+        def up = Mock(Upstream) {
+            2 * getId() >> "id"
+        }
+        def up1 = Mock(Upstream) {
+            1 * getId() >> "id1"
+        }
+        def up2 = Mock(Upstream) {
+            1 * getId() >> "id2"
+        }
+        def value = "null".getBytes()
+        def quorum = new NotNullQuorum()
+
+        when:
+        def res = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up, "id")
+        def res1 = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up1, "id1")
+        def res2 = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up2, "id2")
+        def res3 = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up, "id")
+        then:
+        !res
+        !res1
+        !res2
+        res3
+        quorum.result == value
+        !quorum.isFailed()
+        quorum.isResolved()
+        quorum.signature == new ResponseSigner.Signature("sig1".bytes, "test", 100)
+        quorum.providedUpstreamId == "id"
+    }
+
+    def "Failed if all upstreams respond with error"() {
+        setup:
+        def up = Mock(Upstream) {
+            2 * getId() >> "id"
+        }
+        def up1 = Mock(Upstream) {
+            1 * getId() >> "id1"
+        }
+        def up2 = Mock(Upstream) {
+            1 * getId() >> "id2"
+        }
+        def quorum = new NotNullQuorum()
+
+        when:
+        quorum.record(new JsonRpcException(10, "error"), null, up)
+        quorum.record(new JsonRpcException(10, "error"), null, up1)
+        quorum.record(new JsonRpcException(10, "error"), null, up2)
+        quorum.record(new JsonRpcException(10, "error"), null, up)
+
+        then:
+        quorum.isFailed()
+        !quorum.isResolved()
+        quorum.error == new JsonRpcException(10, "error").error
+    }
+
+    def "Resolve if one of upstream responds with value"() {
+        setup:
+        def up = Mock(Upstream) {
+            2 * getId() >> "id"
+        }
+        def up1 = Mock(Upstream) {
+            1 * getId() >> "id1"
+        }
+        def up2 = Mock(Upstream) {
+            1 * getId() >> "id2"
+        }
+        def value = "null".getBytes()
+        def quorum = new NotNullQuorum()
+
+        when:
+        def res = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up, "id")
+        quorum.record(new JsonRpcException(10, "error"), new ResponseSigner.Signature("sig1".bytes, "test", 100), up1)
+        quorum.record(new JsonRpcException(10, "error"), new ResponseSigner.Signature("sig1".bytes, "test", 100), up2)
+        quorum.record(new JsonRpcException(10, "error"), new ResponseSigner.Signature("sig1".bytes, "test", 100), up)
+
+        then:
+        !res
+        quorum.isResolved()
+        !quorum.isFailed()
+        quorum.result == value
+        quorum.signature == new ResponseSigner.Signature("sig1".bytes, "test", 100)
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy

@@ -148,7 +148,7 @@ class QuorumRpcReaderSpec extends Specification {
                 Chain.ETHEREUM,
                 [up], Selector.empty
         )
-        def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer))
+        def reader = new QuorumRpcReader(apis, new NotNullQuorum(), Stub(Tracer))
 
         when:
         def act = reader.read(new JsonRpcRequest("eth_test", []))
@@ -180,39 +180,7 @@ class QuorumRpcReaderSpec extends Specification {
                 Chain.ETHEREUM,
                 [up], Selector.empty
         )
-        def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer))
+        def reader = new QuorumRpcReader(apis, new NotNullQuorum(), Stub(Tracer))
-
-        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
-            _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
-            _ * getIngressReader() >> 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(
-                Chain.ETHEREUM,
-                [up], Selector.empty
-        )
-        def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer))
 
         when:
         def act = reader.read(new JsonRpcRequest("eth_test", []))
@@ -230,13 +198,13 @@ class QuorumRpcReaderSpec extends Specification {
     def "non-empty-quorum - error if all failed"() {
         setup:
         def api = Mock(Reader) {
-            3 * read(new JsonRpcRequest("eth_test", [])) >>> [
+            2 * read(new JsonRpcRequest("eth_test", [])) >>> [
-                    Mono.just(JsonRpcResponse.ok("null")),
+                    Mono.just(JsonRpcResponse.error(1, "test")),
                     Mono.just(JsonRpcResponse.error(1, "test")),
-                    Mono.just(JsonRpcResponse.ok("null"))
             ]
         }
         def up = Mock(Upstream) {
+            _ * getId() >> "test"
             _ * isAvailable() >> true
             _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
             _ * getIngressReader() >> api
@@ -245,7 +213,7 @@ class QuorumRpcReaderSpec extends Specification {
                 Chain.ETHEREUM,
                 [up], Selector.empty
         )
-        def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3), Stub(Tracer))
+        def reader = new QuorumRpcReader(apis, new NotNullQuorum(), Stub(Tracer))
 
         when:
         def act = reader.read(new JsonRpcRequest("eth_test", []))

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy

@@ -16,7 +16,6 @@
  */
 package io.emeraldpay.dshackle.rpc
 
-
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.Chain
@@ -141,6 +140,7 @@ class TrackEthereumTxSpec extends Specification {
         act
                 .expectSubscription()
                 .expectNoEvent(Duration.ofSeconds(20)).as("Waited for updates")
+                .thenAwait(Duration.ofSeconds(2))
                 .expectComplete()
                 .verify(Duration.ofSeconds(3))
     }
@@ -171,18 +171,15 @@ class TrackEthereumTxSpec extends Specification {
         def apiMock = TestingCommons.api()
         def upstreamMock = TestingCommons.upstream(apiMock)
         MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock)
-        def scheduler = VirtualTimeScheduler.create(true)
+        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic())
-        TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, scheduler)
 
-        apiMock.answerOnce("eth_getTransactionByHash", [txId], null)
+        apiMock.answer("eth_getTransactionByHash", [txId], null, 2)
         apiMock.answer("eth_getTransactionByHash", [txId], txJson)
 
         when:
-        def act = StepVerifier.withVirtualTime({
+        def act = trackTx.subscribe(req).take(2)
-            return trackTx.subscribe(req).take(2)
-        }, { scheduler }, 5)
         then:
-        act
+        StepVerifier.create(act)
                 .expectSubscription()
                 .expectNext(exp1).as("Unknown tx")
                 .expectNext(exp2).as("Found in mempool")

src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy

@@ -7,7 +7,7 @@ import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.ChainsConfig
 import io.emeraldpay.dshackle.config.CompressionConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
+import io.emeraldpay.dshackle.quorum.NotNullQuorum
 import io.emeraldpay.dshackle.upstream.CallTargetsHolder
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import org.springframework.context.ApplicationEventPublisher
@@ -47,7 +47,7 @@ class ConfiguredUpstreamsSpec extends Specification {
         def act = configurer.buildMethods(upstream, Chain.ETHEREUM)
         then:
         act instanceof ManagedCallMethods
-        act.createQuorumFor("foo_bar") instanceof NonEmptyQuorum
+        act.createQuorumFor("foo_bar") instanceof NotNullQuorum
     }
 
     def "Got static response from extra methods"() {

src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy

@@ -17,11 +17,7 @@
 package io.emeraldpay.dshackle.upstream.calls
 
 import io.emeraldpay.dshackle.Chain
-import io.emeraldpay.dshackle.quorum.AlwaysQuorum
+import io.emeraldpay.dshackle.quorum.*
-import io.emeraldpay.dshackle.quorum.BroadcastQuorum
-import io.emeraldpay.dshackle.quorum.CallQuorum
-import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
-import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
 import spock.lang.Specification
 
 import java.util.concurrent.Executors
@@ -109,7 +105,7 @@ class ManagedCallMethodsSpec extends Specification {
         when:
         def act = managed.createQuorumFor("eth_test")
         then:
-        act instanceof NonEmptyQuorum
+        act instanceof NotNullQuorum
 
         when:
         act = managed.createQuorumFor("eth_foo")

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy

@@ -427,32 +427,6 @@ class EthereumDirectReaderSpec extends Specification {
                 .verify(Duration.ofSeconds(1))
     }
 
-    def "Reads tx by hash with retries - expects an error within 1 sec"() {
-        setup:
-        def up = Mock(Multistream) {
-            4 * getApiSource(_) >> Stub(ApiSource)
-        }
-        def calls = Mock(Factory) {
-            4 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM)
-        }
-        EthereumDirectReader reader = new EthereumDirectReader(
-                up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
-        )
-        reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
-            4 * create(_, _, _, _) >> Mock(Reader) {
-                4 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >>>
-                        [Mono.error(new RuntimeException()), Mono.error(new RuntimeException()),
-                         Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]
-            }
-        }
-        when:
-        def act = reader.txReader.read(TransactionId.from(hash1))
-        then:
-        StepVerifier.create(act)
-                .expectError()
-                .verify(Duration.ofSeconds(1))
-    }
-
     def "Reads balance with retries - expects an error within 1 sec"() {
         setup:
         def up = Mock(Multistream) {