diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index 88512d50..bd435163 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/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( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt deleted file mode 100644 index ffd07f2c..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt +++ /dev/null @@ -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::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" - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt new file mode 100644 index 00000000..e1007d38 --- /dev/null +++ b/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() + 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 + + 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 = resolvers.toList() + + override fun toString(): String { + return "Quorum: Not null" + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 02c02716..664a970f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -169,7 +169,7 @@ class QuorumRpcReader( fun withErrorResume(api: Upstream, key: JsonRpcRequest): Function, Mono> { 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 // diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 3d12309d..8c529088 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt index a7095d5c..359dcff6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt +++ b/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() 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 0e237f5b..9a92a528 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt +++ b/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() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt index 0aab3e16..61aa0fcf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt +++ b/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 { AlwaysQuorum() } "no-lag", "not-lagging", "no_lag", "not_lagging" -> Factory { NotLaggingQuorum(0) } - "not-empty", "not_empty", "non-empty", "non_empty" -> Factory { NonEmptyQuorum() } + "not-empty", "not_empty", "non-empty", "non_empty" -> Factory { NotNullQuorum() } else -> { log.warn("Unknown quorum: $quorumId for custom method $method") return diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index c49886f6..0bc57048 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -76,9 +76,8 @@ class EthereumDirectReader( txReader = object : Reader { override fun read(key: TransactionId): Mono { 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) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt index 217eb6c0..5ae28784 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt +++ b/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 } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy deleted file mode 100644 index 55dba4c0..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy +++ /dev/null @@ -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() - } - - -} diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NotNullQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NotNullQuorumSpec.groovy new file mode 100644 index 00000000..a8b79e7c --- /dev/null +++ b/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) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index 6f8463e8..8b75a784 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/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)) - - 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)) + def reader = new QuorumRpcReader(apis, new NotNullQuorum(), 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", [])) >>> [ - Mono.just(JsonRpcResponse.ok("null")), + 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ + 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", [])) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index b6039f23..4838df0b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/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, scheduler) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) - apiMock.answerOnce("eth_getTransactionByHash", [txId], null) + apiMock.answer("eth_getTransactionByHash", [txId], null, 2) apiMock.answer("eth_getTransactionByHash", [txId], txJson) when: - def act = StepVerifier.withVirtualTime({ - return trackTx.subscribe(req).take(2) - }, { scheduler }, 5) + def act = trackTx.subscribe(req).take(2) then: - act + StepVerifier.create(act) .expectSubscription() .expectNext(exp1).as("Unknown tx") .expectNext(exp2).as("Found in mempool") diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index 69b31536..db80f9fa 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/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"() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy index d5ce9f4a..d2416776 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy +++ b/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.BroadcastQuorum -import io.emeraldpay.dshackle.quorum.CallQuorum -import io.emeraldpay.dshackle.quorum.NonEmptyQuorum -import io.emeraldpay.dshackle.quorum.NotLaggingQuorum +import io.emeraldpay.dshackle.quorum.* 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") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index 470957da..3f047a27 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/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) {