From 674c69f4d3004dcdd1c64fe3beba11e48cb750b8 Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Tue, 1 Aug 2023 17:26:31 +0400 Subject: [PATCH] Add BroadcastReader instead of BroadcastQuorum (#263) --- .../emeraldpay/dshackle/commons/constatnts.kt | 5 +- .../dshackle/quorum/QuorumReaderFactory.kt | 45 ---- .../dshackle/quorum/QuorumRpcReader.kt | 32 +-- .../dshackle/reader/BroadcastReader.kt | 118 +++++++++++ .../dshackle/reader/RpcReaderFactory.kt | 81 ++++++++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 24 ++- .../upstream/calls/DefaultEthereumMethods.kt | 2 - .../upstream/ethereum/EthereumDirectReader.kt | 24 +-- .../quorum/QuorumRpcReaderSpec.groovy | 7 + .../reader/BroadcastReaderSpec.groovy | 192 ++++++++++++++++++ .../dshackle/rpc/NativeCallSpec.groovy | 37 ++-- .../ethereum/EthereumCachingReaderSpec.groovy | 137 +++++-------- 12 files changed, 497 insertions(+), 207 deletions(-) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt b/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt index 52ba194b..271d7be4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt @@ -7,16 +7,15 @@ const val SPAN_STATUS_MESSAGE = "status.message" const val SPAN_READER_RESULT = "reader.result" const val SPAN_REQUEST_API_TYPE = "request.api.type" const val SPAN_REQUEST_UPSTREAM_ID = "request.upstreamId" -const val SPAN_RESPONSE_UPSTREAM_ID = "response.upstreamId" const val SPAN_REQUEST_ID = "request.id" const val SPAN_NO_RESPONSE_MESSAGE = "no-response.message" const val LOCAL_READER = "localReader" -const val REMOTE_QUORUM_RPC_READER = "remoteQuorumRpcReader" +const val RPC_READER = "rpcReader" const val API_READER = "apiReader" const val CACHE_BLOCK_BY_HASH_READER = "cacheBlockByHashReader" const val DIRECT_QUORUM_RPC_READER = "directQuorumRpcReader" -const val CACHE_HEIGHT_BY_HASH_READER = "cacheHeightByHashReader" const val CACHE_BLOCK_BY_HEIGHT_READER = "cacheBlockByHeightReader" const val CACHE_TX_BY_HASH_READER = "cacheTxByHashReader" const val CACHE_RECEIPTS_READER = "cacheReceiptsReader" +const val BROADCAST_READER = "broadcastReader" diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt deleted file mode 100644 index d2821865..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt +++ /dev/null @@ -1,45 +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.reader.Reader -import io.emeraldpay.dshackle.upstream.ApiSource -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest -import io.emeraldpay.dshackle.upstream.signature.ResponseSigner -import org.springframework.cloud.sleuth.Tracer -import java.util.concurrent.atomic.AtomicInteger - -interface QuorumReader : Reader { - fun attempts(): AtomicInteger -} - -// creates instance of a Quorum based reader -interface QuorumReaderFactory { - - companion object { - fun default(): QuorumReaderFactory { - return Default() - } - } - - fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): QuorumReader - - class Default : QuorumReaderFactory { - override fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): QuorumReader { - return QuorumRpcReader(apis, quorum, signer, tracer) - } - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 78ba2798..9b295f8f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -20,10 +20,10 @@ import io.emeraldpay.dshackle.commons.API_READER import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE import io.emeraldpay.dshackle.commons.SPAN_REQUEST_API_TYPE import io.emeraldpay.dshackle.commons.SPAN_REQUEST_UPSTREAM_ID +import io.emeraldpay.dshackle.reader.RpcReader import io.emeraldpay.dshackle.reader.SpannedReader import io.emeraldpay.dshackle.upstream.ApiSource 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 @@ -47,9 +47,9 @@ import java.util.function.Function class QuorumRpcReader( private val apiControl: ApiSource, private val quorum: CallQuorum, - private val signer: ResponseSigner?, + signer: ResponseSigner?, private val tracer: Tracer -) : QuorumReader { +) : RpcReader(signer) { companion object { private val log = LoggerFactory.getLogger(QuorumRpcReader::class.java) @@ -158,12 +158,7 @@ class QuorumRpcReader( private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function, Mono>>> { return Function { src -> src.map { - val signature = response.providedSignature - ?: if (key.nonce != null) { - signer?.sign(key.nonce, response.getResult(), api.getId()) - } else { - null - } + val signature = getSignature(key, response, api.getId()) Tuples.of(it, Optional.ofNullable(signature)) } } @@ -176,14 +171,7 @@ class QuorumRpcReader( // when the call failed with an error we want to notify the quorum because // it may use the error message or other details // - val cleanErr: JsonRpcException = when (err) { - is RpcException -> JsonRpcException.from(err) - is JsonRpcException -> err - else -> JsonRpcException( - JsonRpcResponse.NumberId(key.id), - JsonRpcError(-32603, "Unhandled internal error: ${err.javaClass}: ${err.message}") - ) - } + val cleanErr: JsonRpcException = getError(key, err) quorum.record(cleanErr, null, api,) // if it's failed after that, then we don't need more calls, stop api source if (quorum.isFailed()) { @@ -202,8 +190,7 @@ class QuorumRpcReader( return Mono.just(quorum).flatMap { q -> if (q.isFailed()) { val resolvedBy = resolvedBy()?.getId() - val err = q.getError()?.asException(JsonRpcResponse.NumberId(key.id), resolvedBy) - ?: JsonRpcException(JsonRpcResponse.NumberId(key.id), JsonRpcError(-32603, "Unhandled Upstream error"), resolvedBy) + val err = handleError(q.getError(), key.id, resolvedBy) log.warn("Quorum is failed. Method ${key.method}, message ${err.message}") Mono.error(err) } else { @@ -229,11 +216,4 @@ class QuorumRpcReader( } } ?: Mono.error(RpcException(1, "Quorum [$q] is not resolved [isResolved - ${q.isResolved()}]")) } - - class Result( - val value: ByteArray, - val signature: ResponseSigner.Signature?, - val quorum: Int, - val resolvedBy: Upstream? - ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt new file mode 100644 index 00000000..d7138291 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt @@ -0,0 +1,118 @@ +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.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError +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.Mono +import java.util.concurrent.atomic.AtomicInteger + +class BroadcastReader( + private val upstreams: List, + matcher: Selector.Matcher, + signer: ResponseSigner?, + private val tracer: Tracer +) : RpcReader(signer) { + private val internalMatcher = Selector.MultiMatcher( + listOf(Selector.AvailabilityMatcher(), matcher) + ) + + companion object { + private val log = LoggerFactory.getLogger(BroadcastReader::class.java) + } + + override fun attempts(): AtomicInteger { + return AtomicInteger(1) + } + + 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) + ) + }.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) + } + } + } + } + + 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 + ): Mono = + SpannedReader( + upstream.getIngressReader(), tracer, BROADCAST_READER, mapOf(SPAN_REQUEST_UPSTREAM_ID to upstream.getId()) + ) + .read(key) + .map { BroadcastResponse(it, upstream.getId()) } + .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()) + ) + } + + private class BroadcastResponse( + val jsonRpcResponse: JsonRpcResponse, + val upstreamId: String + ) + + 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 new file mode 100644 index 00000000..b109fb13 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt @@ -0,0 +1,81 @@ +package io.emeraldpay.dshackle.reader + +import io.emeraldpay.dshackle.quorum.CallQuorum +import io.emeraldpay.dshackle.quorum.QuorumRpcReader +import io.emeraldpay.dshackle.reader.RpcReader.Result +import io.emeraldpay.dshackle.upstream.Multistream +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 io.emeraldpay.etherjar.rpc.RpcException +import org.springframework.cloud.sleuth.Tracer +import java.util.concurrent.atomic.AtomicInteger + +abstract class RpcReader( + private val signer: ResponseSigner?, +) : Reader { + abstract fun attempts(): AtomicInteger + + protected fun getError(key: JsonRpcRequest, err: Throwable) = + when (err) { + is RpcException -> JsonRpcException.from(err) + is JsonRpcException -> err + else -> JsonRpcException( + JsonRpcResponse.NumberId(key.id), + JsonRpcError(-32603, "Unhandled internal error: ${err.javaClass}: ${err.message}") + ) + } + + protected fun handleError(error: JsonRpcError?, id: Int, resolvedBy: String?) = + error?.asException(JsonRpcResponse.NumberId(id), resolvedBy) + ?: JsonRpcException(JsonRpcResponse.NumberId(id), JsonRpcError(-32603, "Unhandled Upstream error"), resolvedBy) + + protected fun getSignature(key: JsonRpcRequest, response: JsonRpcResponse, upstreamId: String) = + response.providedSignature + ?: if (key.nonce != null) { + signer?.sign(key.nonce, response.getResult(), upstreamId) + } else { + null + } + + class Result( + val value: ByteArray, + val signature: ResponseSigner.Signature?, + val quorum: Int, + val resolvedBy: Upstream? + ) +} + +interface RpcReaderFactory { + + companion object { + fun default(): RpcReaderFactory { + return Default() + } + } + + fun create(data: RpcReaderData): RpcReader + + 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) + } + val apis = data.multistream.getApiSource(data.matcher) + return QuorumRpcReader(apis, data.quorum, data.signer, data.tracer) + } + } + + data class RpcReaderData( + val multistream: Multistream, + val method: String, + val matcher: Selector.Matcher, + val quorum: CallQuorum, + val signer: ResponseSigner?, + val tracer: Tracer + ) +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 39d8ab6d..1d01de93 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -25,15 +25,16 @@ 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 +import io.emeraldpay.dshackle.commons.RPC_READER import io.emeraldpay.dshackle.commons.SPAN_ERROR import io.emeraldpay.dshackle.commons.SPAN_REQUEST_ID import io.emeraldpay.dshackle.commons.SPAN_STATUS_MESSAGE import io.emeraldpay.dshackle.config.MainConfig import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.NotLaggingQuorum -import io.emeraldpay.dshackle.quorum.QuorumReaderFactory -import io.emeraldpay.dshackle.quorum.QuorumRpcReader +import io.emeraldpay.dshackle.reader.RpcReader +import io.emeraldpay.dshackle.reader.RpcReaderFactory +import io.emeraldpay.dshackle.reader.RpcReaderFactory.RpcReaderData import io.emeraldpay.dshackle.reader.SpannedReader import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.ApiSource @@ -79,7 +80,7 @@ open class NativeCall( private val localRouterEnabled = config.cache?.requestsCacheEnabled ?: true private val passthrough = config.passthrough - var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() + var rpcReaderFactory: RpcReaderFactory = RpcReaderFactory.default() private val ethereumCallSelectors = EnumMap(Chain::class.java) companion object { @@ -391,15 +392,18 @@ open class NativeCall( if (!ctx.upstream.getMethods().isCallable(ctx.payload.method)) { return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method")) } - val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum, signer, tracer) + val reader = rpcReaderFactory.create( + RpcReaderData(ctx.upstream, ctx.payload.method, ctx.matcher, ctx.callQuorum, signer, tracer) + ) val counter = reader.attempts() - return SpannedReader(reader, tracer, REMOTE_QUORUM_RPC_READER) + return SpannedReader(reader, tracer, RPC_READER) .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector)) .map { val bytes = ctx.resultDecorator.processResult(it) validateResult(bytes, "remote", ctx) - CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, it.resolvedBy?.getId(), ctx) + val upId = it.resolvedBy?.getId() ?: ctx.upstream.getId() + CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, upId, ctx) } .onErrorResume { t -> Mono.just(CallResult.fail(ctx.id, ctx.nonce, t, ctx)) @@ -470,11 +474,11 @@ open class NativeCall( } interface ResultDecorator { - fun processResult(result: QuorumRpcReader.Result): ByteArray + fun processResult(result: RpcReader.Result): ByteArray } open class NoneResultDecorator : ResultDecorator { - override fun processResult(result: QuorumRpcReader.Result): ByteArray = result.value + override fun processResult(result: RpcReader.Result): ByteArray = result.value } open class CreateFilterDecorator : ResultDecorator { @@ -482,7 +486,7 @@ open class NativeCall( companion object { const val quoteCode = '"'.code.toByte() } - override fun processResult(result: QuorumRpcReader.Result): ByteArray { + override fun processResult(result: RpcReader.Result): ByteArray { val bytes = result.value if (bytes.last() == quoteCode && result.resolvedBy != null) { val suffix = result.resolvedBy.nodeId() 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 1c54280f..4d4a84fc 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,6 @@ 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.NotLaggingQuorum import io.emeraldpay.dshackle.quorum.NotNullQuorum @@ -198,7 +197,6 @@ class DefaultEthereumMethods( when (method) { "eth_getTransactionCount" -> AlwaysQuorum() "eth_getBalance" -> AlwaysQuorum() - "eth_sendRawTransaction" -> BroadcastQuorum() "eth_blockNumber" -> NotLaggingQuorum(0) else -> AlwaysQuorum() } 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 de01432b..279a0ce0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -10,8 +10,8 @@ import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.DefaultContainer import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxId -import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.reader.RpcReaderFactory import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.calls.CallMethods @@ -51,7 +51,7 @@ class EthereumDirectReader( } private val objectMapper: ObjectMapper = Global.objectMapper - var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() + var rpcReaderFactory: RpcReaderFactory = RpcReaderFactory.default() val blockReader: Reader> val blockByHeightReader: Reader> @@ -183,19 +183,17 @@ class EthereumDirectReader( request: JsonRpcRequest, matcher: Selector.Matcher = Selector.empty ): Mono> { - return Mono.just(quorumReaderFactory) + return Mono.just(rpcReaderFactory) .map { + val requestMatcher = Selector.Builder() + .withMatcher(matcher) + .forMethod(request.method) + .build() it.create( - up.getApiSource( - Selector.Builder() - .withMatcher(matcher) - .forMethod(request.method) - .build() - ), - callMethodsFactory.create().createQuorumFor(request.method), - // we do not use Signer for internal requests because it doesn't make much sense - null, - tracer + RpcReaderFactory.RpcReaderData( + up, request.method, requestMatcher, + callMethodsFactory.create().createQuorumFor(request.method), null, tracer + ) ) }.flatMap { it.read(request) diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy index fd292513..ce71f16e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy @@ -39,6 +39,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getId() >> "id" _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY 1 * getIngressReader() >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(JsonRpcResponse.ok("1")) @@ -72,6 +73,7 @@ class QuorumRpcReaderSpec extends Specification { } def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getId() >> "id" _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getIngressReader() >> api } @@ -109,6 +111,7 @@ class QuorumRpcReaderSpec extends Specification { } def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getId() >> "id" _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getIngressReader() >> api } @@ -136,6 +139,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getId() >> "id" _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getIngressReader() >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ @@ -168,6 +172,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getId() >> "id" _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getIngressReader() >> Mock(Reader) { 2 * read(new JsonRpcRequest("eth_test", [])) >>> [ @@ -236,6 +241,7 @@ class QuorumRpcReaderSpec extends Specification { } def up = Mock(Upstream) { _ * isAvailable() >> true + _ * getId() >> "id" _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getIngressReader() >> api } @@ -264,6 +270,7 @@ class QuorumRpcReaderSpec extends Specification { setup: def up = Mock(Upstream) { _ * getLag() >> 0 + _ * getId() >> "id" _ * isAvailable() >> true _ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY _ * getIngressReader() >> Mock(Reader) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy new file mode 100644 index 00000000..29a2ccd7 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy @@ -0,0 +1,192 @@ +package io.emeraldpay.dshackle.reader + +import io.emeraldpay.dshackle.upstream.Selector +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 org.springframework.cloud.sleuth.Tracer +import reactor.core.publisher.Mono +import reactor.test.StepVerifier +import spock.lang.Specification + +import java.time.Duration + +class BroadcastReaderSpec extends Specification { + + def "Return responses from all upstreams"() { + setup: + def result = "123".getBytes() + def up = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.just(new JsonRpcResponse(result, null)) + } + } + def up1 = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.just(new JsonRpcResponse(result, null)) + } + } + def up2 = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.just(new JsonRpcResponse(result, null)) + } + } + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, Stub(Tracer)) + when: + def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) + then: + StepVerifier.create(act) + .expectNextMatches { + it.value == result + } + .expectComplete() + .verify(Duration.ofSeconds(3)) + } + + def "Return response if at least one upstream responds"() { + setup: + def result = "123".getBytes() + def up = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.just(new JsonRpcResponse(result, null)) + } + } + def up1 = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.error(new JsonRpcException(1, "too low")) + } + } + def up2 = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 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)) + when: + def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) + then: + StepVerifier.create(act) + .expectNextMatches { + it.value == result + } + .expectComplete() + .verify(Duration.ofSeconds(3)) + } + + def "Return response from matched upstream"() { + setup: + def result = "123".getBytes() + def up = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.just(new JsonRpcResponse(result, null)) + } + } + def up1 = Mock(Upstream) { + 1 * isAvailable() >> false + 0 * getId() >> "id" + 0 * getIngressReader() >> Mock(Reader) + } + def up2 = Mock(Upstream) { + 1 * isAvailable() >> false + 0 * getId() >> "id" + 0 * getIngressReader() >> Mock(Reader) + } + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, Stub(Tracer)) + when: + def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) + then: + StepVerifier.create(act) + .expectNextMatches { + it.value == result + } + .expectComplete() + .verify(Duration.ofSeconds(3)) + } + + def "Return error if all upstreams return error"() { + setup: + def up = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.error(new JsonRpcException(1, "too low")) + } + } + def up1 = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 1 * read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) >> + Mono.error(new JsonRpcException(1, "too low")) + } + } + def up2 = Mock(Upstream) { + 1 * isAvailable() >> true + _ * getId() >> "id" + 1 * getIngressReader() >> Mock(Reader) { + 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)) + when: + def act = reader.read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) + then: + StepVerifier.create(act) + .expectError(JsonRpcException.class) + .verify(Duration.ofSeconds(3)) + } + + def "No response if no available upstreams"() { + setup: + def up = Mock(Upstream) { + 1 * isAvailable() >> false + 0 * getId() >> "id" + 0 * getIngressReader() >> Mock(Reader) + } + def up1 = Mock(Upstream) { + 1 * isAvailable() >> false + 0 * getId() >> "id" + 0 * getIngressReader() >> Mock(Reader) + } + def up2 = Mock(Upstream) { + 1 * isAvailable() >> false + 0 * getId() >> "id" + 0 * getIngressReader() >> Mock(Reader) + } + def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, 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() + .verify(Duration.ofSeconds(3)) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 0e245913..32c13c3f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -25,17 +25,12 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.CacheConfig import io.emeraldpay.dshackle.config.MainConfig import io.emeraldpay.dshackle.quorum.AlwaysQuorum -import io.emeraldpay.dshackle.quorum.QuorumReader -import io.emeraldpay.dshackle.quorum.QuorumReaderFactory -import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.reader.RpcReader +import io.emeraldpay.dshackle.reader.RpcReaderFactory import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Multistream -import io.emeraldpay.dshackle.upstream.MultistreamHolder -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError @@ -132,9 +127,9 @@ class NativeCallSpec extends Specification { } def nativeCall = nativeCall() - nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { - 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, ups)) + nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { + 1 * read(_) >> Mono.just(new RpcReader.Result("\"foo\"".bytes, null, 1, ups)) } } def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, @@ -153,8 +148,8 @@ class NativeCallSpec extends Specification { def quorum = new AlwaysQuorum() def nativeCall = nativeCall() - nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * attempts() >> new AtomicInteger(1) 1 * read(new JsonRpcRequest("eth_test", [], 10)) >> Mono.empty() } @@ -178,8 +173,8 @@ class NativeCallSpec extends Specification { def quorum = new AlwaysQuorum() def nativeCall = nativeCall() - nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_test", [], 10)) >> Mono.error( new JsonRpcException(JsonRpcResponse.Id.from(12), new JsonRpcError(-32123, "Foo Bar", "Foo Bar Baz"), null, true) ) @@ -615,9 +610,9 @@ class NativeCallSpec extends Specification { _ * it.observeChains() >> Flux.empty() } def nativeCall = nativeCall(multistreamHolder) - nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { - 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, ups)) + nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { + 1 * read(_) >> Mono.just(new RpcReader.Result("\"0xab\"".bytes, null, 1, ups)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, @@ -650,9 +645,9 @@ class NativeCallSpec extends Specification { _ * it.observeChains() >> Flux.empty() } def nativeCall = nativeCall(multistreamHolder) - nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { - 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, ups)) + nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { + 1 * read(_) >> Mono.just(new RpcReader.Result("\"0xab\"".bytes, null, 1, ups)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, 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 53218002..ff4d1704 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -5,22 +5,17 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CurrentBlockCache import io.emeraldpay.dshackle.data.DefaultContainer -import io.emeraldpay.dshackle.quorum.QuorumReader -import io.emeraldpay.dshackle.quorum.QuorumReaderFactory -import io.emeraldpay.dshackle.quorum.QuorumRpcReader +import io.emeraldpay.dshackle.reader.RpcReader +import io.emeraldpay.dshackle.reader.RpcReaderFactory import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.ApiSource -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Multistream -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods +import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.domain.Wei -import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import org.apache.commons.collections4.Factory @@ -47,19 +42,16 @@ class EthereumDirectReaderSpec extends Specification { parentHash = BlockHash.from(hash1) transactions = [] } - def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _,) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolver) ) } @@ -77,19 +69,16 @@ class EthereumDirectReaderSpec extends Specification { def "Produce empty result on non-existing block"() { setup: - def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(null), null, 1, resolver ) ) @@ -113,21 +102,16 @@ class EthereumDirectReaderSpec extends Specification { parentHash = BlockHash.from(hash1) transactions = [] } - def up = Mock(Multistream) { - 1 * getApiSource( - new Selector.Builder().withMatcher(new Selector.HeightMatcher(100)).forMethod("eth_getBlockByNumber").build() - ) >> Stub(ApiSource) - } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolver ) ) @@ -151,19 +135,16 @@ class EthereumDirectReaderSpec extends Specification { blockNumber = 100 blockHash = BlockHash.from(hash1) } - def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolver ) ) @@ -187,19 +168,16 @@ class EthereumDirectReaderSpec extends Specification { blockNumber = 100 blockHash = BlockHash.from(hash1) } - def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolver ) ) @@ -220,9 +198,6 @@ class EthereumDirectReaderSpec extends Specification { blockNumber = 100 blockHash = BlockHash.from(hash1) } - def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } @@ -231,12 +206,12 @@ class EthereumDirectReaderSpec extends Specification { 1 * cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer data -> data.txId.toHex() == hash1.substring(2) && data.height == 100 }) } EthereumDirectReader reader = new EthereumDirectReader( - up, caches, new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), caches, new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolver ) ) @@ -252,19 +227,16 @@ class EthereumDirectReaderSpec extends Specification { def "Produce empty on non-existing tx"() { setup: - def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } EthereumDirectReader reader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(null), null, 1, resolver ) ) @@ -281,7 +253,6 @@ class EthereumDirectReaderSpec extends Specification { def "Reads balance - height is unknown"() { setup: def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) 1 * getHead() >> Mock(Head) { 1 * getCurrentHeight() >> null } @@ -292,10 +263,10 @@ class EthereumDirectReaderSpec extends Specification { EthereumDirectReader reader = new EthereumDirectReader( up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver ) ) @@ -313,7 +284,6 @@ class EthereumDirectReaderSpec extends Specification { def "Reads balance - height is known"() { setup: def up = Mock(Multistream) { - 1 * getApiSource(_) >> Stub(ApiSource) 1 * getHead() >> Mock(Head) { 1 * getCurrentHeight() >> 11_061_691 } @@ -324,10 +294,10 @@ class EthereumDirectReaderSpec extends Specification { EthereumDirectReader reader = new EthereumDirectReader( up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver ) ) @@ -352,25 +322,22 @@ class EthereumDirectReaderSpec extends Specification { totalDifficulty = BigInteger.ONE transactions = [] } - def up = Mock(Multistream) { - 3 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } def result = Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolver) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 2 * create(_, _, _, _) >> Mock(QuorumReader) { + ethereumDirectReader.rpcReaderFactory = Mock(RpcReaderFactory) { + 2 * create(_) >> Mock(RpcReader) { 2 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())] } - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> result } } @@ -395,25 +362,22 @@ class EthereumDirectReaderSpec extends Specification { parentHash = BlockHash.from(hash1) transactions = [] } - def up = Mock(Multistream) { - 3 * getApiSource(_) >> Stub(ApiSource) - } def calls = Mock(Factory) { 3 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET) } def result = Mono.just( - new QuorumRpcReader.Result( + new RpcReader.Result( Global.objectMapper.writeValueAsBytes(json), null, 1, resolver) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( - up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() + Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 2 * create(_, _, _, _) >> Mock(QuorumReader) { + ethereumDirectReader.rpcReaderFactory = Mock(RpcReaderFactory) { + 2 * create(_) >> Mock(RpcReader) { 2 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException())] } - 1 * create(_, _, _, _) >> Mock(QuorumReader) { + 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> result } } @@ -431,7 +395,6 @@ class EthereumDirectReaderSpec extends Specification { def "Reads balance with retries - expects an error within 1 sec"() { setup: def up = Mock(Multistream) { - 4 * getApiSource(_) >> Stub(ApiSource) 1 * getHead() >> Mock(Head) { 1 * getCurrentHeight() >> null } @@ -442,8 +405,8 @@ class EthereumDirectReaderSpec extends Specification { EthereumDirectReader reader = new EthereumDirectReader( up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() ) - reader.quorumReaderFactory = Mock(QuorumReaderFactory) { - 4 * create(_, _, _, _) >> Mock(QuorumReader) { + reader.rpcReaderFactory = Mock(RpcReaderFactory) { + 4 * create(_) >> Mock(RpcReader) { 4 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >>> [Mono.error(new RuntimeException()), Mono.error(new RuntimeException()), Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]