From 2b7109cea33fc2e219bb59ddd0dbfe5e0e02f6d2 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 4 Aug 2019 18:13:18 -0400 Subject: [PATCH] solution: use quorum for native calls --- build.gradle | 1 + .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 135 +++++++----- .../dshackle/upstream/AggregatedUpstreams.kt | 6 +- .../dshackle/upstream/AlwaysQuorum.kt | 26 +++ .../dshackle/upstream/AvailableChains.kt | 16 +- .../dshackle/upstream/BroadcastQuorum.kt | 43 ++++ .../dshackle/upstream/CallQuorum.kt | 31 +++ .../dshackle/upstream/ChainUpstreams.kt | 11 +- .../dshackle/upstream/ConfiguredUpstreams.kt | 13 +- .../dshackle/upstream/EthereumApi.kt | 103 +-------- .../dshackle/upstream/EthereumTargets.kt | 137 ++++++++++++ .../dshackle/upstream/EthereumUpstream.kt | 6 +- .../dshackle/upstream/FilteringApiIterator.kt | 35 +++- .../dshackle/upstream/GrpcUpstream.kt | 18 +- .../dshackle/upstream/GrpcUpstreams.kt | 2 +- .../dshackle/upstream/NonEmptyQuorum.kt | 36 ++++ .../dshackle/upstream/NonceQuorum.kt | 53 +++++ .../dshackle/upstream/NotLaggingQuorum.kt | 45 ++++ .../dshackle/upstream/ValueAwareQuorum.kt | 33 +++ .../dshackle/rpc/NativeCallSpec.groovy | 195 ++++++++++++++++++ .../dshackle/rpc/TrackAddressSpec.groovy | 6 +- .../dshackle/rpc/TrackTxSpec.groovy | 6 +- .../dshackle/test/EthereumApiMock.groovy | 7 +- .../dshackle/test/TestingCommons.groovy | 18 ++ .../upstream/BroadcastQuorumSpec.groovy | 78 +++++++ .../upstream/EthereumGrpcTransportSpec.groovy | 13 +- .../upstream/FilteringApiIteratorSpec.groovy | 68 +++--- .../dshackle/upstream/GrpcUpstreamSpec.groovy | 13 +- .../dshackle/upstream/NonceQuorumSpec.groovy | 80 +++++++ 29 files changed, 999 insertions(+), 235 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/AlwaysQuorum.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/BroadcastQuorum.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/CallQuorum.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/NonEmptyQuorum.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/NonceQuorum.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/NotLaggingQuorum.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ValueAwareQuorum.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/BroadcastQuorumSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/NonceQuorumSpec.groovy diff --git a/build.gradle b/build.gradle index c855d1b4..ac4fd827 100644 --- a/build.gradle +++ b/build.gradle @@ -98,6 +98,7 @@ dependencies { testCompile "org.spockframework:spock-core:$spockVersion" testCompile "io.grpc:grpc-testing:${grpcVersion}" testCompile "io.projectreactor:reactor-test:$reactorVersion" + testCompile 'org.objenesis:objenesis:3.0.1' } compileKotlin { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 2b7f89d3..9cd34ad0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -3,13 +3,9 @@ package io.emeraldpay.dshackle.rpc import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams -import io.emeraldpay.dshackle.upstream.EthereumApi -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.grpc.Chain -import io.grpc.stub.StreamObserver -import org.apache.commons.lang3.StringUtils +import io.infinitape.etherjar.rpc.RpcException import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -17,8 +13,10 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.toFlux import reactor.core.publisher.toMono +import reactor.util.function.Tuple2 import reactor.util.function.Tuples import java.lang.Exception +import java.util.function.Predicate @Service class NativeCall( @@ -29,63 +27,98 @@ class NativeCall( private val log = LoggerFactory.getLogger(NativeCall::class.java) open fun nativeCall(requestMono: Mono): Flux { - return requestMono.flatMapMany { request -> - val chain= Chain.byId(request.chain.number) - if (chain == Chain.UNSPECIFIED) { - // TODO send error to all requests? - throw Exception("Invalid chain id: ${request.chain.number}") - } - val matcher = Selector.convertToMatcher(request.selector) - val upstream = upstreams.getUpstream(chain)?.getApi(matcher) ?: throw Exception("Chain ${chain.id} is unavailable") - request.itemsList.toFlux().map { - val method = it.target - val params = it.payload.toStringUtf8() - CallContext(it.id, upstream, Tuples.of(method, params)) - } + return requestMono.flatMapMany(this::prepareCall) + .map(this::setupCallParams) + .flatMap(this::executeOnRemote) + .map(this::buildResponse) + .doOnError { e -> log.warn("Error during native call", e) } + .onErrorResume(this::processException) + } + + fun setupCallParams(it: CallContext>): CallContext>> { + val params = extractParams(it.payload.t2) + return it.withPayload(Tuples.of(it.payload.t1, params)) + } + + fun buildResponse(it: CallContext): BlockchainOuterClass.NativeCallReplyItem { + return BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setSucceed(true) + .setId(it.id) + .setPayload(ByteString.copyFrom(it.payload)) + .build() + } + + fun processException(it: Throwable?): Mono { + val id: Int = if (it != null && CallFailure::class.isInstance(it)) { + (it as CallFailure).id + } else { + log.error("Lost context for a native call", it) + 0 } - .map { - val params = extractParams(it.payload.t2) - it.withPayload(Tuples.of(it.payload.t1, params)) + return BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setSucceed(false) + .setId(id) + .build() + .toMono() + } + + fun prepareCall(request: BlockchainOuterClass.NativeCallRequest): Flux>> { + val chain = Chain.byId(request.chain.number) + if (chain == Chain.UNSPECIFIED) { + return Flux.error>>(CallFailure(0, Exception("Invalid chain id: ${request.chain.number}"))) } - .flatMap { ctx -> - ctx.upstream.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { resp -> - ctx.withPayload(resp) - }.onErrorMap { - CallFailure(ctx.id, it) - } - } - .map { - BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setSucceed(true) - .setId(it.id) - .setPayload(ByteString.copyFrom(it.payload)) - .build() - } - .onErrorResume() { - val id: Int = if (it != null && CallFailure::class.isInstance(it)) { - (it as CallFailure).id - } else { - log.error("Lost context for a native call", it) - 0 - } - BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setSucceed(false) - .setId(id) - .build() - .toMono() + val upstream = upstreams.getUpstream(chain) + ?: return Flux.error>>(CallFailure(0, Exception("Chain ${chain.id} is unavailable"))) + + return prepareCall(request, upstream) + } + + fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstreams): Flux>> { + val matcher = Selector.convertToMatcher(request.selector) + val apis = upstream.getApis(matcher) + return request.itemsList.toFlux().map { + val method = it.target + val params = it.payload.toStringUtf8() + val callQuorum = upstream.targets?.getQuorumFor(method) ?: AlwaysQuorum() + callQuorum.init(upstream.getHead()) + + CallContext(it.id, apis, callQuorum, Tuples.of(method, params)) } } + fun executeOnRemote(ctx: CallContext>>): Mono> { + val p: Predicate = CallQuorum.untilResolved(ctx.callQuorum) + return ctx.apis.toFlux() + .takeWhile(p) + .flatMap { api -> + api.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { Tuples.of(it, api.upstream!!) } + } + .reduce(ctx.callQuorum, CallQuorum.asReducer()) + .filter { it.isResolved() } + .map { + val result = it.getResult() + ?: throw CallFailure(ctx.id, Exception("No response from upstream for ${ctx.payload.t1}")) + ctx.withPayload(result) + } + .onErrorMap { + if (it is CallFailure) it + else CallFailure(ctx.id, it) + } + .switchIfEmpty( + Mono.error>(CallFailure(ctx.id, Exception("No response or no available upstream for ${ctx.payload.t1}"))) + ) + } + private fun extractParams(jsonParams: String): List { val req = objectMapper.readValue(jsonParams, List::class.java) return req as List } - private class CallContext(val id: Int, val upstream: EthereumApi, val payload: T) { + open class CallContext(val id: Int, val apis: Iterator, val callQuorum: CallQuorum, val payload: T) { fun withPayload(payload: X): CallContext { - return CallContext(id, upstream, payload) + return CallContext(id, apis, callQuorum, payload) } } - class CallFailure(val id: Int, val reason: Throwable): Exception("Failed to call $id: ${reason.message}") + open class CallFailure(val id: Int, val reason: Throwable): Exception("Failed to call $id: ${reason.message}") } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt index 9a85f21b..d03efdb5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt @@ -7,11 +7,13 @@ import java.time.Instant import java.util.concurrent.atomic.AtomicReference import java.util.function.Predicate -abstract class AggregatedUpstreams: Upstream { +abstract class AggregatedUpstreams( + val targets: EthereumTargets +): Upstream { abstract fun getAll(): List abstract fun addUpstream(upstream: Upstream) - abstract fun getApis(quorum: Int, matcher: Selector.Matcher): Iterator + abstract fun getApis(matcher: Selector.Matcher): Iterator override fun observeStatus(): Flux { val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AlwaysQuorum.kt new file mode 100644 index 00000000..dcce7093 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AlwaysQuorum.kt @@ -0,0 +1,26 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson + +open class AlwaysQuorum: CallQuorum { + + private var resolved = false + private var result: ByteArray? = null + + override fun init(head: Head>) { + } + + override fun isResolved(): Boolean { + return resolved + } + + override fun record(response: ByteArray, upstream: Upstream) { + result = response + resolved = true + } + + override fun getResult(): ByteArray? { + return result + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt index c5ea605b..0b2ac871 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt @@ -1,6 +1,8 @@ package io.emeraldpay.dshackle.upstream +import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.grpc.Chain +import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Repository import reactor.core.publisher.Flux import reactor.core.publisher.TopicProcessor @@ -8,10 +10,13 @@ import java.util.* import kotlin.collections.LinkedHashSet @Repository -class AvailableChains { +class AvailableChains( + @Autowired private val objectMapper: ObjectMapper +) { private val all = LinkedHashSet() private val bus = TopicProcessor.create() + private val callTargets = HashMap() fun add(chain: Chain) { all.add(chain) @@ -29,4 +34,13 @@ class AvailableChains { fun getAll(): Set { return Collections.unmodifiableSet(all) } + + fun targetFor(chain: Chain): EthereumTargets { + var current = callTargets[chain] + if (current == null) { + current = EthereumTargets(objectMapper, chain) + callTargets[chain] = current + } + return current + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BroadcastQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BroadcastQuorum.kt new file mode 100644 index 00000000..caa3a6d8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BroadcastQuorum.kt @@ -0,0 +1,43 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.json.BlockJson + +open class BroadcastQuorum( + jacksonRpcConverter: JacksonRpcConverter, + val quorum: Int = 3 +): CallQuorum, ValueAwareQuorum(jacksonRpcConverter, String::class.java) { + + private var result: ByteArray? = null + private var txid: String? = null + private var calls = 0 + + override fun init(head: Head>) { + } + + override fun isResolved(): Boolean { + return calls >= quorum + } + + override fun getResult(): ByteArray? { + return result + } + + override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream) { + calls++ + if (txid == null && responseValue != null) { + txid = responseValue + result = response + } + } + + override fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) { + // can be "message: known transaction: TXID" or "message: Nonce too low" + calls++ + if (result == null) { + result = response + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallQuorum.kt new file mode 100644 index 00000000..cc8a572b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallQuorum.kt @@ -0,0 +1,31 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.util.function.Tuple2 +import java.util.function.BiFunction +import java.util.function.Predicate + +interface CallQuorum { + + fun init(head: Head>) + + fun isResolved(): Boolean + fun record(response: ByteArray, upstream: Upstream) + fun getResult(): ByteArray? + + companion object { + fun untilResolved(cq: CallQuorum): Predicate { + return Predicate { _ -> + !cq.isResolved() + } + } + + fun asReducer(): BiFunction, CallQuorum> { + return BiFunction, CallQuorum> { a, b -> + a.record(b.t1, b.t2) + return@BiFunction a + } + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index de4ab552..a5289cfa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -8,8 +8,9 @@ import java.time.Duration class ChainUpstreams ( val chain: Chain, - private val upstreams: MutableList -) : AggregatedUpstreams() { + private val upstreams: MutableList, + targets: EthereumTargets +) : AggregatedUpstreams(targets) { private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private var seq = 0 @@ -43,16 +44,16 @@ class ChainUpstreams ( head = updateHead() } - override fun getApis(quorum: Int, matcher: Selector.Matcher): Iterator { + override fun getApis(matcher: Selector.Matcher): Iterator { val i = seq++ if (seq >= Int.MAX_VALUE / 2) { seq = 0 } - return FilteringApiIterator(upstreams, 1, seq, matcher) + return FilteringApiIterator(upstreams, i, matcher) } override fun getApi(matcher: Selector.Matcher): EthereumApi { - return getApis(1, matcher).next() + return getApis(matcher).next() } override fun getHead(): EthereumHead { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index 92290155..8c26614e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -17,6 +17,7 @@ import java.io.File import java.net.URI import java.util.* import javax.annotation.PostConstruct +import kotlin.collections.HashMap @Repository open class ConfiguredUpstreams( @@ -100,7 +101,8 @@ open class ConfiguredUpstreams( rpcApi = EthereumApi( DefaultRpcClient(DefaultRpcTransport(endpoint.url)), objectMapper, - chain + chain, + availableChains.targetFor(chain) ) urls.add(endpoint.url) } @@ -114,7 +116,12 @@ open class ConfiguredUpstreams( } if (rpcApi != null) { log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - getOrCreateUpstream(chain).addUpstream(EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels))) + getOrCreateUpstream(chain) + .addUpstream( + EthereumUpstream( + chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), availableChains.targetFor(chain) + ) + ) } } @@ -146,7 +153,7 @@ open class ConfiguredUpstreams( override fun getOrCreateUpstream(chain: Chain): ChainUpstreams { val current = chainMapping[chain] if (current == null) { - val created = ChainUpstreams(chain, ArrayList()) + val created = ChainUpstreams(chain, ArrayList(), availableChains.targetFor(chain)) chainMapping[chain] = created availableChains.add(chain) return created diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index e40fc3fe..35d162c2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -5,18 +5,17 @@ import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.hex.HexQuantity import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.json.ResponseJson -import io.infinitape.etherjar.rpc.transport.BatchStatus import org.slf4j.LoggerFactory -import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.publisher.toFlux import java.time.Duration import java.util.* open class EthereumApi( val rpcClient: RpcClient, private val objectMapper: ObjectMapper, - private val chain: Chain + private val chain: Chain, + val targets: EthereumTargets, + var upstream: Upstream? = null ) { private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) @@ -28,43 +27,6 @@ open class EthereumApi( field = value } - private val allowedMethods = listOf( - "eth_gasPrice", - "eth_blockNumber", - "eth_getBalance", - "eth_getStorageAt", - "eth_getTransactionCount", - "eth_getBlockTransactionCountByHash", - "eth_getBlockTransactionCountByNumber", - "eth_getUncleCountByBlockHash", - "eth_getUncleCountByBlockNumber", - "eth_getCode", - "eth_sendRawTransaction", - "eth_call", - "eth_estimateGas", - "eth_getBlockByHash", - "eth_getBlockByNumber", - "eth_getTransactionByHash", - "eth_getTransactionByBlockHashAndIndex", - "eth_getTransactionByBlockNumberAndIndex", - "eth_getTransactionReceipt", - "eth_getUncleByBlockHashAndIndex", - "eth_getUncleByBlockNumberAndIndex" - ) - - private val hardcodedMethods = listOf( - "net_version", - "net_peerCount", - "net_listening", - "web3_clientVersion", - "eth_protocolVersion", - "eth_syncing", - "eth_coinbase", - "eth_mining", - "eth_hashrate", - "eth_accounts" - ) - open fun executeAndConvert(rpcCall: RpcCall): Mono { return execute(0, rpcCall.method, rpcCall.params as List) .flatMap { @@ -77,17 +39,17 @@ open class EthereumApi( } open fun execute(id: Int, method: String, params: List): Mono { - val result: Mono = if (hardcodedMethods.contains(method)) { + val result: Mono = if (targets.isHardcoded(method)) { Mono.just(method) - .map { hardcoded(it) } - } else if (allowedMethods.contains(method)) { + .map { targets.hardcoded(it) } + } else if (targets.isAllowed(method)) { callUpstream(method, params) } else { Mono.error(RpcException(-32601, "Method not allowed or not found")) } return result .doOnError { t -> - log.warn("Upstream error: ${t.message}") + log.warn("Upstream error: ${t.message} for ${method} on $chain") } .map { val resp = ResponseJson() @@ -120,56 +82,7 @@ open class EthereumApi( } return Mono.fromCompletionStage( rpcClient.execute(RpcCall.create(method, Any::class.java, params)) - ).timeout(timeout) + ).timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout"))) } - fun hardcoded(method: String): Any { - if ("net_version" == method) { - if (Chain.ETHEREUM == chain) { - return "1" - } - if (Chain.ETHEREUM_CLASSIC == chain) { - return "1" - } - if (Chain.TESTNET_MORDEN == chain) { - return "2" - } - if (Chain.TESTNET_KOVAN == chain) { - return "42" - } - throw RpcException(-32602, "Invalid chain") - } - if ("net_peerCount" == method) { - return "0x2a" - } - if ("net_listening" == method) { - return true - } - if ("web3_clientVersion" == method) { - return "EmeraldDshackle/v0.2" - } - if ("eth_protocolVersion" == method) { - return "0x3f" - } - if ("eth_syncing" == method) { - return false - } - if ("eth_coinbase" == method) { - return "0x0000000000000000000000000000000000000000" - } - if ("eth_mining" == method) { - return "false" - } - if ("eth_hashrate" == method) { - return "0x0" - } - if ("eth_accounts" == method) { - return Collections.emptyList() - } - throw RpcException(-32601, "Method not found") - } - - fun getSupportedMethods(): Set { - return allowedMethods.plus(hardcodedMethods).toSortedSet() - } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt new file mode 100644 index 00000000..27603006 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt @@ -0,0 +1,137 @@ +package io.emeraldpay.dshackle.upstream + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.RpcException +import java.util.* + +class EthereumTargets( + private val objectMapper: ObjectMapper, + private val chain: Chain +) { + + private val jacksonRpcConverter = JacksonRpcConverter(objectMapper) + + private val anyResponseMethods = listOf( + "eth_gasPrice", + "eth_call", + "eth_estimateGas" + ) + + private val firstValueMethods = listOf( + "eth_getBlockTransactionCountByHash", + "eth_getUncleCountByBlockHash", + "eth_getBlockByHash", + "eth_getTransactionByHash", + "eth_getTransactionByBlockHashAndIndex", + "eth_getStorageAt", + "eth_getCode", + "eth_getUncleByBlockHashAndIndex" + ) + + private val specialMethods = listOf( + "eth_getTransactionCount", + "eth_blockNumber", + "eth_getBalance", + "eth_sendRawTransaction" + ) + + private val headVerifiedMethods = listOf( + "eth_getBlockTransactionCountByNumber", + "eth_getUncleCountByBlockNumber", + "eth_getBlockByNumber", + "eth_getTransactionByBlockNumberAndIndex", + "eth_getTransactionReceipt", + "eth_getUncleByBlockNumberAndIndex" + ) + + private val allowedMethods = anyResponseMethods + firstValueMethods + specialMethods + headVerifiedMethods + + private val hardcodedMethods = listOf( + "net_version", + "net_peerCount", + "net_listening", + "web3_clientVersion", + "eth_protocolVersion", + "eth_syncing", + "eth_coinbase", + "eth_mining", + "eth_hashrate", + "eth_accounts" + ) + + open fun getQuorumFor(method: String): CallQuorum { + return when { + hardcodedMethods.contains(method) -> AlwaysQuorum() + anyResponseMethods.contains(method) -> NotLaggingQuorum(6) + headVerifiedMethods.contains(method) -> NotLaggingQuorum(1) + specialMethods.contains(method) -> { + when (method) { + "eth_getTransactionCount" -> NonceQuorum(jacksonRpcConverter) + "eth_getBalance" -> NotLaggingQuorum(1) + "eth_sendRawTransaction" -> BroadcastQuorum(jacksonRpcConverter) + else -> AlwaysQuorum() + } + } + else -> AlwaysQuorum() + } + } + + fun isAllowed(method: String): Boolean { + return allowedMethods.contains(method) + } + fun isHardcoded(method: String): Boolean { + return hardcodedMethods.contains(method) + } + + fun hardcoded(method: String): Any { + if ("net_version" == method) { + if (Chain.ETHEREUM == chain) { + return "1" + } + if (Chain.ETHEREUM_CLASSIC == chain) { + return "1" + } + if (Chain.TESTNET_MORDEN == chain) { + return "2" + } + if (Chain.TESTNET_KOVAN == chain) { + return "42" + } + throw RpcException(-32602, "Invalid chain") + } + if ("net_peerCount" == method) { + return "0x2a" + } + if ("net_listening" == method) { + return true + } + if ("web3_clientVersion" == method) { + return "EmeraldDshackle/v0.2" + } + if ("eth_protocolVersion" == method) { + return "0x3f" + } + if ("eth_syncing" == method) { + return false + } + if ("eth_coinbase" == method) { + return "0x0000000000000000000000000000000000000000" + } + if ("eth_mining" == method) { + return "false" + } + if ("eth_hashrate" == method) { + return "0x0" + } + if ("eth_accounts" == method) { + return Collections.emptyList() + } + throw RpcException(-32601, "Method not found") + } + + fun getSupportedMethods(): Set { + return allowedMethods.plus(hardcodedMethods).toSortedSet() + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 73a69792..3f526729 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -14,11 +14,12 @@ open class EthereumUpstream( private val api: EthereumApi, private val ethereumWs: EthereumWs? = null, private val options: UpstreamsConfig.Options, - val node: NodeDetailsList.NodeDetails + val node: NodeDetailsList.NodeDetails, + private val targets: EthereumTargets ): Upstream { override fun getSupportedTargets(): Set { - return api.getSupportedMethods() + return targets.getSupportedMethods() } private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) @@ -37,6 +38,7 @@ open class EthereumUpstream( init { log.info("Configured for ${chain.chainName}") + api.upstream = this validator.start() .subscribe { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt index ac3c9ff0..423a14e6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt @@ -2,25 +2,40 @@ package io.emeraldpay.dshackle.upstream class FilteringApiIterator( private val apis: List, - private val quorum: Int, private var pos: Int, - private val matcher: Selector.Matcher + private val matcher: Selector.Matcher, + private val repeatLimit: Int = 3 ): Iterator { + private var nextApi: Upstream? = null private var consumed = 0 + private fun nextInternal(): Boolean { + if (nextApi != null) { + return true + } + while (nextApi == null) { + consumed++ + if (consumed > apis.size * repeatLimit) { + return false + } + val api = apis[pos++ % apis.size] + if (api.isAvailable(matcher)) { + nextApi = api + } + } + return nextApi != null + } + override fun hasNext(): Boolean { - return consumed < quorum + return nextInternal() } override fun next(): EthereumApi { - val start = pos - while (pos < start + apis.size) { - val api = apis[pos++ % apis.size] - if (api.isAvailable(matcher)) { - consumed++ - return api.getApi(matcher) - } + if (nextInternal()) { + val curr = nextApi!! + nextApi = null + return curr.getApi(matcher) } throw IllegalStateException("No upstream API available") } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index cb5c8b36..64e68c51 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -28,11 +28,12 @@ open class GrpcUpstream( private val chain: Chain, private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper, - private val options: UpstreamsConfig.Options + private val options: UpstreamsConfig.Options, + private val targets: EthereumTargets ): Upstream { - constructor(chain: Chain, client: ReactorBlockchainGrpc.ReactorBlockchainStub, objectMapper: ObjectMapper) - : this(chain, client, objectMapper, UpstreamsConfig.Options.getDefaults()) + constructor(chain: Chain, client: ReactorBlockchainGrpc.ReactorBlockchainStub, objectMapper: ObjectMapper, targets: EthereumTargets) + : this(chain, client, objectMapper, UpstreamsConfig.Options.getDefaults(), targets) private val log = LoggerFactory.getLogger(GrpcUpstream::class.java) @@ -47,7 +48,7 @@ open class GrpcUpstream( open fun createApi(matcher: Selector.Matcher): EthereumApi { val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher)) - return EthereumApi(rpcClient, objectMapper, chain) + return EthereumApi(rpcClient, objectMapper, chain, targets, this) } open fun connect() { @@ -81,10 +82,13 @@ open class GrpcUpstream( .flatMap { getApi(Selector.EmptyMatcher()) .executeAndConvert(Commands.eth().getBlock(it.hash)) - .timeout(Duration.ofSeconds(15)) + .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream"))) + .doOnError { t -> + log.warn("Failed to download block data", t) + } } - .doOnError { err -> - log.error("Head subscription error", err) + .onErrorContinue { err, _ -> + log.error("Head subscription error: ${err.message}") } .subscribe { block -> log.debug("New block ${block.number} on ${chain}") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt index 5591bfef..243b1da5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt @@ -92,7 +92,7 @@ class GrpcUpstreams( lock.withLock { val current = known[chain] return if (current == null) { - val created = GrpcUpstream(chain, client!!, objectMapper, options) + val created = GrpcUpstream(chain, client!!, objectMapper, options, availableChains.targetFor(chain)) known[chain] = created availableChains.add(chain) created.connect() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonEmptyQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonEmptyQuorum.kt new file mode 100644 index 00000000..15f336d4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonEmptyQuorum.kt @@ -0,0 +1,36 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.json.BlockJson + +open class NonEmptyQuorum( + jacksonRpcConverter: JacksonRpcConverter, + val maxTries: Int = 3 +): CallQuorum, ValueAwareQuorum(jacksonRpcConverter, Any::class.java) { + + private var result: ByteArray? = null + private var tries: Int = 0 + + override fun init(head: Head>) { + } + + override fun isResolved(): Boolean { + return result != null || tries >= maxTries + } + + override fun recordValue(response: ByteArray, responseValue: Any?, upstream: Upstream) { + tries++ + if (responseValue != null) { + result = response + } + } + + override fun getResult(): ByteArray? { + return result + } + + override fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) { + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonceQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonceQuorum.kt new file mode 100644 index 00000000..df33b453 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonceQuorum.kt @@ -0,0 +1,53 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.hex.HexQuantity +import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.json.BlockJson +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +open class NonceQuorum( + jacksonRpcConverter: JacksonRpcConverter, + val tries: Int = 3 +): CallQuorum, ValueAwareQuorum(jacksonRpcConverter, String::class.java) { + + private val lock = ReentrantLock() + private var resultValue = 0L + private var result: ByteArray? = null + private var receivedTimes = 0 + private var errors = 0 + + override fun init(head: Head>) { + } + + override fun isResolved(): Boolean { + lock.withLock { + return receivedTimes >= tries || errors >= tries + } + } + + override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream) { + val value = responseValue?.let { str -> + HexQuantity.from(str).value.toLong() + } + lock.withLock { + receivedTimes++ + if (value != null && value > resultValue) { + resultValue = value + result = response + } else if (result == null) { + result = response + } + } + } + + override fun getResult(): ByteArray? { + return result + } + + override fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) { + errors++ + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NotLaggingQuorum.kt new file mode 100644 index 00000000..f9d6a271 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NotLaggingQuorum.kt @@ -0,0 +1,45 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { + + private var head: Flux> = Flux.empty>() + private val lock = ReentrantLock() + private var resolved = false + private var result: ByteArray? = null + + override fun init(head: Head>) { + this.head = head.getFlux() + } + + override fun isResolved(): Boolean { + return resolved && result != null + } + + override fun record(response: ByteArray, upstream: Upstream) { + Mono.from(head) + .zipWith(upstream.getHead().getHead()) + .map { + val top = it.t1 + val current = it.t2 + return@map (top.number - current.number) < maxLag + }.subscribe { fresh -> + if (fresh) { + lock.withLock { + result = response + resolved = true + } + } + } + } + + override fun getResult(): ByteArray { + return result!! + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ValueAwareQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ValueAwareQuorum.kt new file mode 100644 index 00000000..27ba2c32 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ValueAwareQuorum.kt @@ -0,0 +1,33 @@ +package io.emeraldpay.dshackle.upstream + +import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.RpcException +import org.slf4j.LoggerFactory + +abstract class ValueAwareQuorum( + val jacksonRpcConverter: JacksonRpcConverter, + val clazz: Class +): CallQuorum { + + private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java) + + fun extractValue(response: ByteArray, clazz: Class): T? { + return jacksonRpcConverter.fromJson(response.inputStream(), clazz) + } + + override fun record(response: ByteArray, upstream: Upstream) { + try { + val value = extractValue(response, clazz) + recordValue(response, value, upstream) + } catch (e: RpcException) { + recordError(response, e.rpcMessage, upstream) + } catch (e: Exception) { + recordError(response, e.message, upstream) + } + } + + abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream) + + abstract fun recordError(response: ByteArray, errorMessage: String?, upstream: Upstream) + +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy new file mode 100644 index 00000000..3f621a7d --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -0,0 +1,195 @@ +package io.emeraldpay.dshackle.rpc + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.test.EthereumApiMock +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.AlwaysQuorum +import io.emeraldpay.dshackle.upstream.CallQuorum +import io.emeraldpay.dshackle.upstream.EthereumApi +import io.emeraldpay.dshackle.upstream.NonEmptyQuorum +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.RpcClient +import reactor.test.StepVerifier +import reactor.util.function.Tuples +import spock.lang.Specification + +import java.time.Duration + +class NativeCallSpec extends Specification { + + def objectMapper = TestingCommons.objectMapper() + + def "Quorum is applied"() { + setup: + def quorum = Spy(new AlwaysQuorum()) + def upstreams = Stub(Upstreams) + RpcClient rpcClient = Stub(RpcClient) + def upstream = Stub(Upstream) + def apiMock = TestingCommons.api(rpcClient, upstream) + + apiMock.answer("eth_test", [], "foo") + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def call = new NativeCall.CallContext(1, [apiMock].multiply(5).iterator(), quorum, Tuples.of("eth_test", [])) + + + when: + def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) + def act = objectMapper.readValue(resp.payload, Map) + then: + act == [jsonrpc:"2.0", id:1, result: "foo"] + (2..3) * quorum.isResolved() // 2 times during api call (before and after) + 1 time in a filter after + 1 * quorum.record(_, _) + 1 * quorum.getResult() + } + + def "Quorum may return not first received value"() { + setup: + def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) + + def upstreams = Stub(Upstreams) + RpcClient rpcClient = Stub(RpcClient) + def upstream = Stub(Upstream) + def apiMock = TestingCommons.api(rpcClient, upstream) + + apiMock.answerOnce("eth_test", [], null) + apiMock.answerOnce("eth_test", [], "bar") + apiMock.answerOnce("eth_test", [], null) + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def call = new NativeCall.CallContext(1, [apiMock].multiply(5).iterator(), quorum, Tuples.of("eth_test", [])) + + + when: + def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) + def act = objectMapper.readValue(resp.payload, Map) + then: + act == [jsonrpc:"2.0", id:1, result: "bar"] + (3..4) * quorum.isResolved() + 2 * quorum.record(_, _) + 1 * quorum.getResult() + } + + def "Returns error if no quorum"() { + setup: + def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) + + def upstreams = Stub(Upstreams) + RpcClient rpcClient = Stub(RpcClient) + def upstream = Stub(Upstream) + def apiMock = TestingCommons.api(rpcClient, upstream) + + apiMock.answer("eth_test", [], null, 3) + apiMock.answerOnce("eth_test", [], "foo") + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def call = new NativeCall.CallContext(1, [apiMock].multiply(5).iterator(), quorum, Tuples.of("eth_test", [])) + + (4..5) * quorum.isResolved() + 3 * quorum.record(_, _) + 1 * quorum.getResult() + + when: + def resp = nativeCall.executeOnRemote(call) + then: + StepVerifier.create(resp) + .expectErrorMatches({t -> t instanceof NativeCall.CallFailure && t.id == 1}) + .verify(Duration.ofSeconds(1)) + } + + def "Packs call exception into response with id"() { + setup: + def upstreams = Stub(Upstreams) + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + when: + def resp = nativeCall.processException(new NativeCall.CallFailure(5, new IllegalArgumentException("test test"))) + then: + StepVerifier.create(resp) + .expectNext(BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setSucceed(false) + .setId(5) + .build()) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Packs unknown exception into response"() { + setup: + def upstreams = Stub(Upstreams) + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + when: + def resp = nativeCall.processException(new IllegalArgumentException("test test")) + then: + StepVerifier.create(resp) + .expectNext(BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setSucceed(false) + .build()) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Builds normal response"() { + setup: + def upstreams = Stub(Upstreams) + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def json = [jsonrpc:"2.0", id:1, result: "foo"] + when: + def resp = nativeCall.buildResponse( + new NativeCall.CallContext(1561, [].iterator(), new AlwaysQuorum(), objectMapper.writeValueAsBytes(json)) + ) + then: + resp.id == 1561 + resp.succeed + objectMapper.readValue(resp.payload.toByteArray(), Map.class) == [jsonrpc:"2.0", id:1, result: "foo"] + } + + def "Returns error for invalid chain"() { + setup: + def upstreams = Stub(Upstreams) + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + + def req = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChainValue(0) + .addAllItems([1, 2].collect { id -> + return BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(id) + .setTarget("eth_test") + .build() + }) + .build() + when: + def resp = nativeCall.prepareCall(req) + then: + StepVerifier.create(resp) + .expectErrorMatches({t -> t instanceof NativeCall.CallFailure && t.id == 0}) +// .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Returns error for unsupported chain"() { + setup: + def upstreams = Mock(Upstreams) + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + + def req = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChainValue(Chain.TESTNET_MORDEN.id) + .addAllItems([1, 2].collect { id -> + return BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(id) + .setTarget("eth_test") + .build() + }) + .build() + 1 * upstreams.getUpstream(Chain.TESTNET_MORDEN) >> null + when: + def resp = nativeCall.prepareCall(req) + then: + StepVerifier.create(resp) + .expectErrorMatches({t -> t instanceof NativeCall.CallFailure && t.id == 0}) +// .expectComplete() + .verify(Duration.ofSeconds(1)) + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy index cad8e867..da4d9f1d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackAddressSpec.groovy @@ -37,7 +37,7 @@ class TrackAddressSpec extends Specification { def setup() { - availableChains = new AvailableChains() + availableChains = new AvailableChains(TestingCommons.objectMapper()) upstreams = Mock(Upstreams) trackAddress = new TrackAddress(upstreams, availableChains, Schedulers.immediate()) } @@ -61,7 +61,7 @@ class TrackAddressSpec extends Specification { .build() def upstreamMock = Mock(AggregatedUpstreams) - def apiMock = new EthereumApiMock(Mock(RpcClient), TestingCommons.objectMapper(), Chain.ETHEREUM) + def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) apiMock.answer("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock _ * upstreamMock.getApi(_) >> apiMock @@ -103,7 +103,7 @@ class TrackAddressSpec extends Specification { def blocksBus = TopicProcessor.create() def upstreamMock = Mock(AggregatedUpstreams) def headMock = Mock(EthereumHead) - def apiMock = new EthereumApiMock(Mock(RpcClient), TestingCommons.objectMapper(), Chain.ETHEREUM) + def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0x499602D2") apiMock.answerOnce("eth_getBalance", ["0xe2c8fa8120d813cd0b5e6add120295bf20cfa09f", "latest"], "0xff98") _ * upstreams.getUpstream(Chain.ETHEREUM) >> upstreamMock diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy index cc38d7f0..c166695f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackTxSpec.groovy @@ -25,7 +25,7 @@ import java.time.Duration class TrackTxSpec extends Specification { - AvailableChains availableChains = new AvailableChains() + AvailableChains availableChains = new AvailableChains(TestingCommons.objectMapper()) Upstreams upstreams TrackTx trackTx @@ -94,7 +94,7 @@ class TrackTxSpec extends Specification { def blocksBus = TopicProcessor.create() def headMock = Mock(EthereumHead) - def apiMock = new EthereumApiMock(Mock(RpcClient), TestingCommons.objectMapper(), Chain.ETHEREUM) + def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) apiMock.answer("eth_getTransactionByHash", [txId], txJson) apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson) @@ -171,7 +171,7 @@ class TrackTxSpec extends Specification { def blocksBus = TopicProcessor.create() def headMock = Mock(EthereumHead) - def apiMock = new EthereumApiMock(Mock(RpcClient), TestingCommons.objectMapper(), Chain.ETHEREUM) + def apiMock = TestingCommons.api(Stub(RpcClient), upstreamMock) apiMock.answerOnce("eth_getTransactionByHash", [txId], null) apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted) apiMock.answer("eth_getTransactionByHash", [txId], txJsonMined) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index 8e1aa72d..91a0318b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -4,6 +4,9 @@ import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.EthereumApi +import io.emeraldpay.dshackle.upstream.EthereumTargets +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import io.infinitape.etherjar.rpc.RpcClient @@ -20,8 +23,8 @@ class EthereumApiMock extends EthereumApi { List predefined = [] private ObjectMapper objectMapper - EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain) { - super(rpcClient, objectMapper, chain) + EthereumApiMock(@NotNull RpcClient rpcClient, @NotNull ObjectMapper objectMapper, @NotNull Chain chain, Upstream upstream) { + super(rpcClient, objectMapper, chain, new EthereumTargets(objectMapper, chain), upstream) this.objectMapper = objectMapper } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index c3d4cf0c..c1595530 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -4,8 +4,18 @@ import com.fasterxml.jackson.core.Version import com.fasterxml.jackson.databind.DeserializationFeature import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.module.SimpleModule +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.Batch +import io.infinitape.etherjar.rpc.ExecutableBatch +import io.infinitape.etherjar.rpc.JacksonRpcConverter +import io.infinitape.etherjar.rpc.RpcCall +import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.transport.BatchStatus +import spock.mock.MockingApi import java.text.SimpleDateFormat +import java.util.concurrent.CompletableFuture class TestingCommons { @@ -21,4 +31,12 @@ class TestingCommons { return objectMapper } + + static EthereumApiMock api(RpcClient rpcClient, Upstream upstream) { + return new EthereumApiMock(rpcClient, objectMapper(), Chain.ETHEREUM, upstream) + } + + static JacksonRpcConverter rpcConverter() { + return new JacksonRpcConverter(objectMapper()) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/BroadcastQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/BroadcastQuorumSpec.groovy new file mode 100644 index 00000000..9bc1d28c --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/BroadcastQuorumSpec.groovy @@ -0,0 +1,78 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.test.TestingCommons +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import spock.lang.Specification + +class BroadcastQuorumSpec extends Specification { + + def rpcConverted = TestingCommons.rpcConverter() + def objectMapper = TestingCommons.objectMapper() + + def "Resolved with first after 3 tries"() { + setup: + def q = Spy(new BroadcastQuorum(rpcConverted, 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + + when: + q.record(objectMapper.writeValueAsBytes([result: "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"]), upstream1) + then: + !q.isResolved() + 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _) + + when: + q.record(objectMapper.writeValueAsBytes([result: "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"]), upstream2) + then: + !q.isResolved() + 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _) + + when: + q.record(objectMapper.writeValueAsBytes([error: [message: "Nonce too low"]]), upstream3) + then: + 1 * q.recordError(_, _, _) + q.isResolved() + objectMapper.readValue(q.result, Map) == [result: "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"] + } + + def "Remembers first response"() { + setup: + def q = Spy(new BroadcastQuorum(rpcConverted, 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + + when: + q.record(objectMapper.writeValueAsBytes([error: [message: "Internal error"]]), upstream1) + then: + !q.isResolved() + 1 * q.recordError(_, _, _) + + when: + q.record(objectMapper.writeValueAsBytes([result: "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"]), upstream2) + then: + !q.isResolved() + 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _) + + when: + q.record(objectMapper.writeValueAsBytes([error: [message: "Nonce too low"]]), upstream3) + then: + 1 * q.recordError(_, _, _) + q.isResolved() + objectMapper.readValue(q.result, Map) == [result: "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"] + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy index 145fc802..f9cdba8b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy @@ -19,6 +19,7 @@ class EthereumGrpcTransportSpec extends Specification { MockServer mockServer = new MockServer() ObjectMapper objectMapper = TestingCommons.objectMapper() + def ethereumTargets = new EthereumTargets(objectMapper, Chain.ETHEREUM) def "Make simple call"() { setup: @@ -26,7 +27,7 @@ class EthereumGrpcTransportSpec extends Specification { def otherSideUpstreams = Mock(Upstreams) def otherSideAggr = Mock(AggregatedUpstreams) def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) - def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) + def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM, otherSideAggr) def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @Override @@ -45,7 +46,9 @@ class EthereumGrpcTransportSpec extends Specification { then: 1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr - 1 * otherSideAggr.getApi(_) >> otherSideApi + 1 * otherSideAggr.getApis(_) >> [otherSideApi].iterator() + _ * otherSideAggr.getHead() >> Stub(EthereumHead) + _ * otherSideAggr.getTargets() >> ethereumTargets status.failed == 0 status.succeed == 1 status.total == 1 @@ -67,7 +70,7 @@ class EthereumGrpcTransportSpec extends Specification { def otherSideUpstreams = Mock(Upstreams) def otherSideAggr = Mock(AggregatedUpstreams) def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) - def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) + def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM, otherSideAggr) def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @Override @@ -90,7 +93,9 @@ class EthereumGrpcTransportSpec extends Specification { then: 1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr - 1 * otherSideAggr.getApi(_) >> otherSideApi + 1 * otherSideAggr.getApis(_) >> [otherSideApi].multiply(3).iterator() + _ * otherSideAggr.getHead() >> Stub(EthereumHead) + _ * otherSideAggr.getTargets() >> ethereumTargets status.failed == 0 status.succeed == 2 status.total == 2 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy index 74cc9767..dbdb39b8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteringApiIteratorSpec.groovy @@ -10,52 +10,32 @@ class FilteringApiIteratorSpec extends Specification { def rpcClient = new DefaultRpcClient(null) def objectMapper = TestingCommons.objectMapper() + def ethereumTargets = new EthereumTargets(objectMapper, Chain.ETHEREUM) def "Verifies labels"() { setup: - def upstreams = [ - new EthereumUpstream( - Chain.ETHEREUM, - new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM), - (EthereumWs)null, - new UpstreamsConfig.Options(), - new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "foo"])) - ), - new EthereumUpstream( - Chain.ETHEREUM, - new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM), - (EthereumWs)null, - new UpstreamsConfig.Options(), - new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "bar"])) - ), - new EthereumUpstream( - Chain.ETHEREUM, - new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM), - (EthereumWs)null, - new UpstreamsConfig.Options(), - new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "foo", test2: "baz"])) - ), - new EthereumUpstream( - Chain.ETHEREUM, - new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM), - (EthereumWs)null, - new UpstreamsConfig.Options(), - new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "foo"])) - ), - new EthereumUpstream( - Chain.ETHEREUM, - new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM), - (EthereumWs)null, - new UpstreamsConfig.Options(), - new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap([test: "baz"])) - ) - ] + List upstreams = [ + [test: "foo"], + [test: "bar"], + [test: "foo", test2: "baz"], + [test: "foo"], + [test: "baz"] + ].collect { + new EthereumUpstream( + Chain.ETHEREUM, + new EthereumApi(rpcClient, objectMapper, Chain.ETHEREUM, ethereumTargets, null), + (EthereumWs) null, + new UpstreamsConfig.Options(), + new NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels.fromMap(it)), + ethereumTargets + ) + } def matcher = new Selector.LabelMatcher("test", ["foo"]) upstreams.forEach { it.setStatus(UpstreamAvailability.OK) } when: - def iter = new FilteringApiIterator(upstreams, 3, 0, matcher) + def iter = new FilteringApiIterator(upstreams, 0, matcher, 1) then: iter.hasNext() iter.next() == upstreams[0].api @@ -66,16 +46,18 @@ class FilteringApiIteratorSpec extends Specification { !iter.hasNext() when: - iter = new FilteringApiIterator(upstreams, 2, 1, matcher) + iter = new FilteringApiIterator(upstreams, 1, matcher, 1) then: iter.hasNext() iter.next() == upstreams[2].api iter.hasNext() iter.next() == upstreams[3].api + iter.hasNext() + iter.next() == upstreams[0].api !iter.hasNext() when: - iter = new FilteringApiIterator(upstreams, 3, 1, matcher) + iter = new FilteringApiIterator(upstreams, 1, matcher, 2) then: iter.hasNext() iter.next() == upstreams[2].api @@ -83,6 +65,12 @@ class FilteringApiIteratorSpec extends Specification { iter.next() == upstreams[3].api iter.hasNext() iter.next() == upstreams[0].api + iter.hasNext() + iter.next() == upstreams[2].api + iter.hasNext() + iter.next() == upstreams[3].api + iter.hasNext() + iter.next() == upstreams[0].api !iter.hasNext() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy index 6e16e45f..04107e0a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy @@ -23,12 +23,13 @@ class GrpcUpstreamSpec extends Specification { MockServer mockServer = new MockServer() ObjectMapper objectMapper = TestingCommons.objectMapper() + def ethereumTargets = new EthereumTargets(objectMapper, Chain.ETHEREUM) def "Subscribe to head"() { setup: def callData = [:] def chain = Chain.ETHEREUM - def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain) + def api = TestingCommons.api(Stub(RpcClient), Stub(Upstream)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -54,7 +55,7 @@ class GrpcUpstreamSpec extends Specification { ) } }) - def upstream = new GrpcUpstream(chain, client, objectMapper) + def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) when: upstream.connect() def h = upstream.head.head.block(Duration.ofSeconds(1)) @@ -69,7 +70,7 @@ class GrpcUpstreamSpec extends Specification { def callData = [:] def finished = new CompletableFuture() def chain = Chain.ETHEREUM - def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain) + def api = TestingCommons.api(Stub(RpcClient), Stub(Upstream)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -109,7 +110,7 @@ class GrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new GrpcUpstream(chain, client, objectMapper) + def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) when: upstream.connect() finished.get() @@ -125,7 +126,7 @@ class GrpcUpstreamSpec extends Specification { def callData = [:] def finished = new CompletableFuture() def chain = Chain.ETHEREUM - def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain) + def api = TestingCommons.api(Stub(RpcClient), Stub(Upstream)) def block1 = new BlockJson().with { it.number = 650246 it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -165,7 +166,7 @@ class GrpcUpstreamSpec extends Specification { finished.complete(true) } }) - def upstream = new GrpcUpstream(chain, client, objectMapper) + def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) when: upstream.connect() finished.get() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/NonceQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/NonceQuorumSpec.groovy new file mode 100644 index 00000000..34535afd --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/NonceQuorumSpec.groovy @@ -0,0 +1,80 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.test.TestingCommons +import spock.lang.Specification + +class NonceQuorumSpec extends Specification { + + def rpcConverted = TestingCommons.rpcConverter() + def objectMapper = TestingCommons.objectMapper() + + def "Gets max value"() { + setup: + def q = Spy(new NonceQuorum(rpcConverted, 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + + when: + q.record(objectMapper.writeValueAsBytes([result: "0x10"]), upstream1) + then: + !q.isResolved() + 1 * q.recordValue(_, "0x10", _) + + when: + q.record(objectMapper.writeValueAsBytes([result: "0x11"]), upstream2) + then: + !q.isResolved() + 1 * q.recordValue(_, "0x11", _) + + when: + q.record(objectMapper.writeValueAsBytes([result: "0x10"]), upstream3) + then: + 1 * q.recordValue(_, "0x10", _) + q.isResolved() + objectMapper.readValue(q.result, Map) == [result: "0x11"] + } + + def "Ignores errors"() { + setup: + def q = Spy(new NonceQuorum(rpcConverted, 3)) + def upstream1 = Stub(Upstream) + def upstream2 = Stub(Upstream) + def upstream3 = Stub(Upstream) + + when: + q.init(Stub(Head)) + then: + !q.isResolved() + + when: + q.record(objectMapper.writeValueAsBytes([error: [error: "Internal"]]), upstream1) + then: + !q.isResolved() + 1 * q.recordError(_, _, _) + + when: + q.record(objectMapper.writeValueAsBytes([result: "0x11"]), upstream2) + then: + !q.isResolved() + 1 * q.recordValue(_, "0x11", _) + + when: + q.record(objectMapper.writeValueAsBytes([result: "0x10"]), upstream3) + then: + 1 * q.recordValue(_, "0x10", _) + !q.isResolved() + + when: + q.record(objectMapper.writeValueAsBytes([result: "0x11"]), upstream1) + then: + 1 * q.recordValue(_, "0x11", _) + q.isResolved() + objectMapper.readValue(q.result, Map) == [result: "0x11"] + } +}