solution: use quorum for native calls

2019-08-04 18:13:18 -04:00
parent 90d46d15a6
commit 2b7109cea3
29 changed files with 999 additions and 235 deletions
--- 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<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
-        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<Tuple2<String, String>>): CallContext<Tuple2<String, List<Any>>> {
+        val params = extractParams(it.payload.t2)
+        return it.withPayload(Tuples.of(it.payload.t1, params))
+    }
+
+    fun buildResponse(it: CallContext<ByteArray>): BlockchainOuterClass.NativeCallReplyItem {
+        return BlockchainOuterClass.NativeCallReplyItem.newBuilder()
+                .setSucceed(true)
+                .setId(it.id)
+                .setPayload(ByteString.copyFrom(it.payload))
+                .build()
+    }
+
+    fun processException(it: Throwable?): Mono<BlockchainOuterClass.NativeCallReplyItem> {
+        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<CallContext<Tuple2<String, String>>> {
+        val chain = Chain.byId(request.chain.number)
+        if (chain == Chain.UNSPECIFIED) {
+            return Flux.error<CallContext<Tuple2<String, String>>>(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<CallContext<Tuple2<String, String>>>(CallFailure(0, Exception("Chain ${chain.id} is unavailable")))
+
+        return prepareCall(request, upstream)
+    }
+
+    fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstreams): Flux<CallContext<Tuple2<String, String>>> {
+        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<Tuple2<String, List<Any>>>): Mono<CallContext<ByteArray>> {
+        val p: Predicate<Any> = 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<CallContext<ByteArray>>(CallFailure(ctx.id, Exception("No response or no available upstream for ${ctx.payload.t1}")))
+                )
+    }
+
    private fun extractParams(jsonParams: String): List<Any> {
        val req = objectMapper.readValue(jsonParams, List::class.java)
        return req as List<Any>
    }

-    private class CallContext<T>(val id: Int, val upstream: EthereumApi, val payload: T) {
+    open class CallContext<T>(val id: Int, val apis: Iterator<EthereumApi>, val callQuorum: CallQuorum, val payload: T) {
        fun <X> withPayload(payload: X): CallContext<X> {
-            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}")
 }
--- 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<Upstream>
    abstract fun addUpstream(upstream: Upstream)
-    abstract fun getApis(quorum: Int, matcher: Selector.Matcher): Iterator<EthereumApi>
+    abstract fun getApis(matcher: Selector.Matcher): Iterator<EthereumApi>

    override fun observeStatus(): Flux<UpstreamAvailability> {
        val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AlwaysQuorum.kt
+++ 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<BlockJson<TransactionId>>) {
+    }
+
+    override fun isResolved(): Boolean {
+        return resolved
+    }
+
+    override fun record(response: ByteArray, upstream: Upstream) {
+        result = response
+        resolved = true
+    }
+
+    override fun getResult(): ByteArray? {
+        return result
+    }
+}
--- 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<Chain>()
    private val bus = TopicProcessor.create<Chain>()
+    private val callTargets = HashMap<Chain, EthereumTargets>()

    fun add(chain: Chain) {
        all.add(chain)
@@ -29,4 +34,13 @@ class AvailableChains {
    fun getAll(): Set<Chain> {
        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
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BroadcastQuorum.kt
+++ 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<String>(jacksonRpcConverter, String::class.java) {
+
+    private var result: ByteArray? = null
+    private var txid: String? = null
+    private var calls = 0
+
+    override fun init(head: Head<BlockJson<TransactionId>>) {
+    }
+
+    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
+        }
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallQuorum.kt
+++ 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<BlockJson<TransactionId>>)
+
+    fun isResolved(): Boolean
+    fun record(response: ByteArray, upstream: Upstream)
+    fun getResult(): ByteArray?
+
+    companion object {
+        fun untilResolved(cq: CallQuorum): Predicate<Any> {
+            return Predicate { _ ->
+                !cq.isResolved()
+            }
+        }
+
+        fun asReducer(): BiFunction<CallQuorum, Tuple2<ByteArray, Upstream>, CallQuorum> {
+            return BiFunction<CallQuorum, Tuple2<ByteArray, Upstream>, CallQuorum> { a, b ->
+                a.record(b.t1, b.t2)
+                return@BiFunction a
+            }
+        }
+    }
+}
--- 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<Upstream>
-) : AggregatedUpstreams() {
+        private val upstreams: MutableList<Upstream>,
+        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<EthereumApi> {
+    override fun getApis(matcher: Selector.Matcher): Iterator<EthereumApi> {
        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 {
--- 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<Upstream>())
+            val created = ChainUpstreams(chain, ArrayList<Upstream>(), availableChains.targetFor(chain))
            chainMapping[chain] = created
            availableChains.add(chain)
            return created
--- 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 <JS, RS> executeAndConvert(rpcCall: RpcCall<JS, RS>): Mono<RS> {
        return execute(0, rpcCall.method, rpcCall.params as List<Any>)
                .flatMap {
@@ -77,17 +39,17 @@ open class EthereumApi(
    }

    open fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
-        val result: Mono<Any> = if (hardcodedMethods.contains(method)) {
+        val result: Mono<Any> = 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<Any, Int>()
@@ -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<String>()
-        }
-        throw RpcException(-32601, "Method not found")
-    }
-
-    fun getSupportedMethods(): Set<String> {
-        return allowedMethods.plus(hardcodedMethods).toSortedSet()
-    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumTargets.kt
+++ 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<String>()
+        }
+        throw RpcException(-32601, "Method not found")
+    }
+
+    fun getSupportedMethods(): Set<String> {
+        return allowedMethods.plus(hardcodedMethods).toSortedSet()
+    }
+}
--- 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<String> {
-        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 {
--- 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<Upstream>,
-        private val quorum: Int,
        private var pos: Int,
-        private val matcher: Selector.Matcher
+        private val matcher: Selector.Matcher,
+        private val repeatLimit: Int = 3
 ): Iterator<EthereumApi> {

+    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")
    }
--- 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}")
--- 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()
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonEmptyQuorum.kt
+++ 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<Any>(jacksonRpcConverter, Any::class.java) {
+
+    private var result: ByteArray? = null
+    private var tries: Int = 0
+
+    override fun init(head: Head<BlockJson<TransactionId>>) {
+    }
+
+    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) {
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NonceQuorum.kt
+++ 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<String>(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<BlockJson<TransactionId>>) {
+    }
+
+    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++
+    }
+
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NotLaggingQuorum.kt
+++ 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<BlockJson<TransactionId>> = Flux.empty<BlockJson<TransactionId>>()
+    private val lock = ReentrantLock()
+    private var resolved = false
+    private var result: ByteArray? = null
+
+    override fun init(head: Head<BlockJson<TransactionId>>) {
+        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!!
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ValueAwareQuorum.kt
+++ 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<T>(
+        val jacksonRpcConverter: JacksonRpcConverter,
+        val clazz: Class<T>
+): CallQuorum {
+
+    private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
+
+    fun extractValue(response: ByteArray, clazz: Class<T>): 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)
+
+}