From eb95c591f786452e143f788f51c9031983b8e2a8 Mon Sep 17 00:00:00 2001 From: terminal Date: Fri, 29 Jul 2022 15:32:48 +0400 Subject: [PATCH] refactoring fork choice rules out of abstract head and priority fork choice support for PoS Ethereum --- .../dshackle/data/BlockContainer.kt | 7 +- .../io/emeraldpay/dshackle/data/RingSet.kt | 44 +++++ .../dshackle/startup/ConfiguredUpstreams.kt | 60 ++++--- .../dshackle/upstream/AbstractHead.kt | 37 ++-- .../upstream/CurrentMultistreamHolder.kt | 11 ++ .../dshackle/upstream/DistanceExtractor.kt | 28 +++ .../dshackle/upstream/HeadLagObserver.kt | 11 +- .../dshackle/upstream/MergedHead.kt | 6 +- .../bitcoin/BitcoinHeadLagObserver.kt | 3 +- .../upstream/bitcoin/BitcoinMultistream.kt | 3 +- .../upstream/bitcoin/BitcoinRpcHead.kt | 3 +- .../upstream/bitcoin/BitcoinZMQHead.kt | 3 +- .../upstream/ethereum/DefaultEthereumHead.kt | 5 +- .../upstream/ethereum/EthereumFees.kt | 3 +- .../ethereum/EthereumHeadLagObserver.kt | 3 +- .../upstream/ethereum/EthereumMultistream.kt | 3 +- .../upstream/ethereum/EthereumPriorityFees.kt | 3 +- .../upstream/ethereum/EthereumRpcHead.kt | 6 +- .../upstream/ethereum/EthereumUpstream.kt | 1 - .../ethereum/EthereumUpstreamValidator.kt | 4 +- .../upstream/ethereum/EthereumWsHead.kt | 6 +- .../connectors/EthereumConnectorFactory.kt | 8 +- .../connectors/EthereumRpcConnector.kt | 11 +- .../connectors/EthereumWsConnector.kt | 4 +- .../EthereumPosHeadLagObserver.kt | 22 +++ .../ethereum_pos/EthereumPosMultiStream.kt | 142 +++++++++++++++ .../ethereum_pos/EthereumPosUpstream.kt | 101 +++++++++-- .../upstream/forkchoice/ForkChoice.kt | 17 ++ .../upstream/forkchoice/MostWorkForkChoice.kt | 32 ++++ .../NoChoiceWithPriorityForkChoice.kt | 36 ++++ .../upstream/forkchoice/PriorityForkChoice.kt | 35 ++++ .../upstream/grpc/BitcoinGrpcUpstream.kt | 3 +- .../upstream/grpc/EthereumGrpcUpstream.kt | 3 +- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 162 ++++++++++++++++++ .../dshackle/upstream/grpc/GrpcHead.kt | 11 +- .../cache/HeightByHashAddingSpec.groovy | 2 +- .../dshackle/cache/ReceiptMemCacheSpec.groovy | 3 +- .../rpc/TrackBitcoinAddressSpec.groovy | 4 +- .../dshackle/rpc/TrackBitcoinTxSpec.groovy | 6 +- .../dshackle/rpc/TrackEthereumTxSpec.groovy | 5 +- .../dshackle/test/TestingCommons.groovy | 3 +- .../dshackle/upstream/AbstractHeadSpec.groovy | 26 ++- .../upstream/DistanceExtractorSpec.groovy | 74 ++++++++ .../dshackle/upstream/FilteredApisSpec.groovy | 3 +- .../upstream/HeadLagObserverSpec.groovy | 37 +--- .../dshackle/upstream/MergedHeadSpec.groovy | 11 +- .../ethereum/DefaultEthereumHeadSpec.groovy | 3 +- .../forkchoice/MostWorkForkChoiceSpec.groovy | 37 ++++ .../NoChoiceWithPriorityForkChoiceSpec.groovy | 51 ++++++ .../forkchoice/PriorityForkChoiceSpec.groovy | 41 +++++ .../upstream/grpc/GrpcHeadSpec.groovy | 5 +- 51 files changed, 995 insertions(+), 153 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/DistanceExtractor.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoice.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/DistanceExtractorSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoiceSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoiceSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt index f5c47d3f..ac411dbb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt @@ -30,7 +30,8 @@ class BlockContainer( val full: Boolean, json: ByteArray?, val parsed: Any?, - val transactions: List = emptyList() + val transactions: List = emptyList(), + val nodeRating: Int = 0 ) : SourceContainer(json, parsed) { companion object { @@ -82,6 +83,10 @@ class BlockContainer( return true } + fun copyWithRating(nodeRating: Int): BlockContainer { + return BlockContainer(height, hash, difficulty, timestamp, full, json, parsed, transactions, nodeRating) + } + override fun hashCode(): Int { var result = super.hashCode() result = 31 * result + height.hashCode() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt new file mode 100644 index 00000000..a3e1815e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt @@ -0,0 +1,44 @@ +package io.emeraldpay.dshackle.data + +import java.util.concurrent.atomic.AtomicReference + +class RingSet( + private val maxSize: Int +): Set { + private var seqValues: AtomicReference> = AtomicReference(emptyList()) + private var set: Set = emptySet() + override val size: Int + get() = set.size + + fun add(element: T) { + if (set.contains(element)) { + return + } + seqValues.getAndUpdate { vals -> + vals.let { + if (vals.size > maxSize) { + vals.drop(1) + } else { + vals + } + }.plus(element).let { + set = HashSet(it) + it + } + } + } + + override fun isEmpty(): Boolean { + return set.isEmpty() + } + override fun contains(element: @UnsafeVariance T): Boolean { + return set.contains(element) + } + override fun iterator(): Iterator { + return set.iterator() + } + + override fun containsAll(elements: Collection<@UnsafeVariance T>): Boolean { + return set.containsAll(elements) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index bcf9e216..10e320ff 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -29,9 +29,13 @@ import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice +import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -82,9 +86,9 @@ open class ConfiguredUpstreams( BlockchainType.BITCOIN -> { buildBitcoinUpstream(up.cast(UpstreamsConfig.BitcoinConnection::class.java), chain, options) } -// BlockchainType.ETHEREUM_POS -> { -// buildEthereumPosUpstream(up.cast(UpstreamsConfig.EthereumPosConnection::class.java), chain, options) -// } + BlockchainType.ETHEREUM_POS -> { + buildEthereumPosUpstream(up.cast(UpstreamsConfig.EthereumPosConnection::class.java), chain, options) + } else -> { log.error("Chain is unsupported: ${up.chain}") return@forEach @@ -139,20 +143,34 @@ open class ConfiguredUpstreams( } } -// private fun buildEthereumPosUpstream( -// config: UpstreamsConfig.Upstream, -// chain: Chain, -// options: UpstreamsConfig.Options -// ) : Upstream? { -// val conn = config.connection!! -// val execution = conn.execution -// if (execution == null) { -// log.warn("Upstream doesn't have execution layer configuration") -// return null -// } -// -// val connectorFactory = buildEthereumConnectorFactory(execution, chain) -// } + private fun buildEthereumPosUpstream( + config: UpstreamsConfig.Upstream, + chain: Chain, + options: UpstreamsConfig.Options + ) : Upstream? { + val conn = config.connection!! + val execution = conn.execution + if (execution == null) { + log.warn("Upstream doesn't have execution layer configuration") + return null + } + val urls = ArrayList() + val connectorFactory = buildEthereumConnectorFactory(execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.blockPriority)) + val methods = buildMethods(config, chain) + if (connectorFactory == null) { + return null + } + val upstream = EthereumPosUpstream( + config.id!!, + chain, + options, config.role, + methods, + QuorumForLabels.QuorumItem(1, config.labels), + connectorFactory + ) + upstream.start() + return upstream + } private fun buildBitcoinUpstream( config: UpstreamsConfig.Upstream, @@ -180,7 +198,7 @@ open class ConfiguredUpstreams( val head: Head = conn.zeroMq?.let { zeroMq -> val server = ZMQServer(zeroMq.host, zeroMq.port, "hashblock") val zeroMqHead = BitcoinZMQHead(server, directApi, extractBlock) - MergedHead(listOf(rpcHead, zeroMqHead)) + MergedHead(listOf(rpcHead, zeroMqHead), MostWorkForkChoice()) } ?: rpcHead val methods = buildMethods(config, chain) @@ -206,7 +224,7 @@ open class ConfiguredUpstreams( val urls = ArrayList() val methods = buildMethods(config, chain) - val connectorFactory = buildEthereumConnectorFactory(conn, chain, urls) + val connectorFactory = buildEthereumConnectorFactory(conn, chain, urls, MostWorkForkChoice()) if (connectorFactory == null) { return null } @@ -272,11 +290,11 @@ open class ConfiguredUpstreams( } } - private fun buildEthereumConnectorFactory(conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList): EthereumConnectorFactory? { + private fun buildEthereumConnectorFactory(conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList, forkChoice: ForkChoice): EthereumConnectorFactory? { val wsFactoryApi = buildWsFactory(conn, urls) val httpFactory = buildHttpFactory(conn, urls) log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory) + val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice) if (!connectorFactory.isValid()) { log.warn("Upstream configuration is invalid (probably no http endpoint)") return null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 9f85e7f6..ce4e7931 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -16,21 +16,24 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.Sinks import reactor.core.scheduler.Schedulers +import reactor.kotlin.core.publisher.toMono import java.util.concurrent.atomic.AtomicReference -abstract class AbstractHead : Head { +abstract class AbstractHead( + private val forkChoice: ForkChoice +) : Head { companion object { private val log = LoggerFactory.getLogger(AbstractHead::class.java) } - private val head = AtomicReference(null) private var stream = Sinks.many().multicast().directBestEffort() private var completed = false private val beforeBlockHandlers = ArrayList() @@ -44,10 +47,8 @@ abstract class AbstractHead : Head { return source .distinctUntilChanged { it.hash - }.filter { block -> - val curr = head.get() - curr == null || curr.difficulty < block.difficulty } + .filter { forkChoice.filter(it) } .doFinally { // close internal stream if upstream is finished, otherwise it gets stuck, // but technically it should never happen during normal work, only when the Head @@ -58,19 +59,16 @@ abstract class AbstractHead : Head { .subscribeOn(Schedulers.boundedElastic()) .subscribe { block -> notifyBeforeBlock() - val prev = head.getAndUpdate { curr -> - if (curr == null || curr.difficulty < block.difficulty) { - block - } else { - curr - } - } - if (prev == null || prev.hash != block.hash) { - log.debug("New block ${block.height} ${block.hash}") - val result = stream.tryEmitNext(block) - if (result.isFailure && result != Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) { - log.warn("Failed to dispatch block: $result as ${this.javaClass}") + when (val choiceResult = forkChoice.choose(block)) { + is ForkChoice.ChoiceResult.Updated -> { + val newHead = choiceResult.nwhead + log.debug("New block ${newHead.height} ${newHead.hash}") + val result = stream.tryEmitNext(newHead) + if (result.isFailure && result != Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER) { + log.warn("Failed to dispatch block: $result as ${this.javaClass}") + } } + is ForkChoice.ChoiceResult.Same -> {} } } } @@ -90,14 +88,15 @@ abstract class AbstractHead : Head { } override fun getFlux(): Flux { + val curHead = forkChoice.getHead() return Flux.concat( - Mono.justOrEmpty(head.get()), + forkChoice.getHead().toMono(), stream.asFlux() ).onBackpressureLatest() } fun getCurrent(): BlockContainer? { - return head.get() + return forkChoice.getHead() } override fun getCurrentHeight(): Long? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt index 9a99d6de..4de9bbef 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt @@ -25,6 +25,8 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultistream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.grpc.BlockchainType import io.emeraldpay.grpc.Chain @@ -68,6 +70,14 @@ open class CurrentMultistreamHolder( } processUpdate(change, up, current, factory) } + BlockchainType.ETHEREUM_POS -> { + val up = change.upstream.cast(EthereumPosUpstream::class.java) + val current = chainMapping[chain] + val factory = Callable { + EthereumPosMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) + } + processUpdate(change, up, current, factory) + } BlockchainType.BITCOIN -> { val up = change.upstream.cast(BitcoinUpstream::class.java) val current = chainMapping[chain] @@ -137,6 +147,7 @@ open class CurrentMultistreamHolder( val created = when (BlockchainType.from(chain)) { BlockchainType.ETHEREUM -> DefaultEthereumMethods(chain) BlockchainType.BITCOIN -> DefaultBitcoinMethods() + BlockchainType.ETHEREUM_POS -> DefaultEthereumMethods(chain) else -> throw IllegalStateException("Unsupported chain: $chain") } callTargets[chain] = created diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DistanceExtractor.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DistanceExtractor.kt new file mode 100644 index 00000000..77dc9b61 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DistanceExtractor.kt @@ -0,0 +1,28 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.data.BlockContainer + +class DistanceExtractor { + sealed class ChainDistance { + data class Distance(val dist: Long): ChainDistance() + object Fork: ChainDistance() + } + + companion object { + fun extractPowDistance(top: BlockContainer, curr: BlockContainer): ChainDistance { + return when { + curr.height > top.height -> if (curr.difficulty >= top.difficulty) ChainDistance.Distance(0) else ChainDistance.Fork + curr.height == top.height -> if (curr.difficulty == top.difficulty) ChainDistance.Distance(0) else ChainDistance.Fork + else -> ChainDistance.Distance(top.height - curr.height) + } + } + + fun extractPriorityDistance(top: BlockContainer, curr: BlockContainer): ChainDistance { + return when { + curr.height > top.height -> ChainDistance.Fork + curr.height == top.height -> if (curr.hash == top.hash) ChainDistance.Distance(0) else ChainDistance.Fork + else -> ChainDistance.Distance(top.height - curr.height) + } + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt index 748e729d..a1c61c2f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt @@ -30,9 +30,11 @@ import java.time.Duration * Observer group of upstreams and defined a distance in blocks (lag) between a leader (best height/difficulty) and * other upstreams. */ +typealias Extractor = (top: BlockContainer, curr: BlockContainer) -> DistanceExtractor.ChainDistance abstract class HeadLagObserver( private val master: Head, - private val followers: Collection + private val followers: Collection, + private val distanceExtractor: Extractor ) : Lifecycle { private val log = LoggerFactory.getLogger(HeadLagObserver::class.java) @@ -85,10 +87,9 @@ abstract class HeadLagObserver( } open fun extractDistance(top: BlockContainer, curr: BlockContainer): Long { - return when { - curr.height > top.height -> if (curr.difficulty >= top.difficulty) 0 else forkDistance(top, curr) - curr.height == top.height -> if (curr.difficulty == top.difficulty) 0 else forkDistance(top, curr) - else -> top.height - curr.height + return when (val distance = distanceExtractor(top, curr)) { + is DistanceExtractor.ChainDistance.Distance -> distance.dist + is DistanceExtractor.ChainDistance.Fork -> forkDistance(top, curr) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt index 57a65ca1..0c87de88 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt @@ -18,13 +18,15 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesEnabled +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux class MergedHead( - private val sources: Iterable -) : AbstractHead(), Lifecycle, CachesEnabled { + private val sources: Iterable, + forkChoice: ForkChoice +) : AbstractHead(forkChoice), Lifecycle, CachesEnabled { private var subscription: Disposable? = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt index ffc3740b..fbfb663d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt @@ -16,6 +16,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.upstream.DistanceExtractor import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.HeadLagObserver import io.emeraldpay.dshackle.upstream.Upstream @@ -24,7 +25,7 @@ import org.slf4j.LoggerFactory class BitcoinHeadLagObserver( master: Head, followers: Collection -) : HeadLagObserver(master, followers) { +) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) { companion object { private val log = LoggerFactory.getLogger(BitcoinHeadLagObserver::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index c7740f1e..8a9054b7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.upstream.RequestPostprocessor import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.ethereum.LocalCallRouter +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain @@ -84,7 +85,7 @@ open class BitcoinMultistream( } } } else { - val newHead = MergedHead(upstreams.map { it.getHead() }).apply { + val newHead = MergedHead(upstreams.map { it.getHead() }, MostWorkForkChoice()).apply { this.start() } val lagObserver = BitcoinHeadLagObserver(newHead, upstreams) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt index f0353f44..982d4fb2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt @@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.slf4j.LoggerFactory @@ -35,7 +36,7 @@ class BitcoinRpcHead( private val api: Reader, private val extractBlock: ExtractBlock, private val interval: Duration = Duration.ofSeconds(15) -) : Head, AbstractHead(), Lifecycle { +) : Head, AbstractHead(MostWorkForkChoice()), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinRpcHead::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt index fa0d206d..ca916408 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt @@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.apache.commons.codec.binary.Hex @@ -20,7 +21,7 @@ class BitcoinZMQHead( private val server: ZMQServer, private val api: Reader, private val extractBlock: ExtractBlock, -) : Head, AbstractHead(), Lifecycle { +) : Head, AbstractHead(MostWorkForkChoice()), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinZMQHead::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt index 8fbe804d..ded52346 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt @@ -20,13 +20,16 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.hex.HexQuantity import org.slf4j.LoggerFactory import reactor.core.publisher.Mono -open class DefaultEthereumHead : Head, AbstractHead() { +open class DefaultEthereumHead( + forkChoice: ForkChoice +) : Head, AbstractHead(forkChoice) { companion object { private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt index 3e798ecf..0cf7c42c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.upstream.AbstractChainFees import io.emeraldpay.dshackle.upstream.ChainFees +import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.etherjar.domain.Wei import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson @@ -28,7 +29,7 @@ import reactor.util.function.Tuples import java.util.function.Function abstract class EthereumFees( - upstreams: EthereumMultistream, + upstreams: Multistream, private val reader: EthereumReader, heightLimit: Int, ) : AbstractChainFees, TransactionRefJson, TransactionJson>(heightLimit, upstreams, extractTx), ChainFees { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt index 9d9d3b87..c933214b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.upstream.DistanceExtractor import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.HeadLagObserver import io.emeraldpay.dshackle.upstream.Upstream @@ -25,7 +26,7 @@ import org.slf4j.LoggerFactory class EthereumHeadLagObserver( master: Head, followers: Collection -) : HeadLagObserver(master, followers) { +) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) { companion object { private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 0e4fcdfe..f9557cb7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.MergedHead import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.grpc.Chain @@ -109,7 +110,7 @@ open class EthereumMultistream( } } else { val heads = upstreams.map { it.getHead() } - val newHead = MergedHead(heads).apply { + val newHead = MergedHead(heads, MostWorkForkChoice()).apply { this.start() } val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt index f3138799..274f6c61 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt @@ -16,6 +16,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.etherjar.domain.Wei import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson @@ -23,7 +24,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory import java.util.function.Function -class EthereumPriorityFees(upstreams: EthereumMultistream, reader: EthereumReader, heightLimit: Int) : +class EthereumPriorityFees(upstreams: Multistream, reader: EthereumReader, heightLimit: Int) : EthereumFees(upstreams, reader, heightLimit) { companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index 572f4fcf..104a9bba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.slf4j.LoggerFactory @@ -30,8 +31,9 @@ import java.util.concurrent.Executors class EthereumRpcHead( private val api: Reader, - private val interval: Duration = Duration.ofSeconds(10) -) : DefaultEthereumHead(), Lifecycle { + forkChoice: ForkChoice, + private val interval: Duration = Duration.ofSeconds(10), +) : DefaultEthereumHead(forkChoice), Lifecycle { companion object { val scheduler = diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index 4248f78f..ef226380 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -63,7 +63,6 @@ open class EthereumUpstream( this.setStatus(UpstreamAvailability.OK) } else { log.debug("Start validation for upstream ${this.getId()}") - val validator = EthereumUpstreamValidator(this, getOptions()) validatorSubscription = validator.start() .subscribe(this::setStatus) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt index cc7feca8..d8cbc7bd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt @@ -20,7 +20,9 @@ import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.rpc.json.SyncingJson @@ -34,7 +36,7 @@ import java.util.concurrent.Executors import java.util.concurrent.TimeoutException open class EthereumUpstreamValidator( - private val upstream: EthereumUpstream, + private val upstream: Upstream, private val options: UpstreamsConfig.Options ) { companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 117e901b..42c7b717 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -16,6 +16,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle @@ -23,8 +24,9 @@ import reactor.core.Disposable import reactor.core.publisher.Flux class EthereumWsHead( - private val ws: WsConnection -) : DefaultEthereumHead(), Lifecycle { + private val ws: WsConnection, + forkChoice: ForkChoice +) : DefaultEthereumHead(forkChoice), Lifecycle { private val log = LoggerFactory.getLogger(EthereumWsHead::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt index 68c11b43..78f69556 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt @@ -5,13 +5,15 @@ import io.emeraldpay.dshackle.upstream.HttpFactory import io.emeraldpay.dshackle.upstream.HttpRpcFactory import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory open class EthereumConnectorFactory( private val preferHttp: Boolean, private val wsFactory: EthereumWsFactory?, - private val httpFactory: HttpFactory? + private val httpFactory: HttpFactory?, + private val forkChoice: ForkChoice ): ConnectorFactory { private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java) @@ -24,11 +26,11 @@ open class EthereumConnectorFactory( override fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector { if (wsFactory!= null && !preferHttp) { - return EthereumWsConnector(wsFactory, upstream, validator, chain) + return EthereumWsConnector(wsFactory, upstream, validator, chain, forkChoice) } if (httpFactory == null) { throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set") } - return EthereumRpcConnector(httpFactory.create(upstream.getId(), chain), wsFactory, upstream.getId()) + return EthereumRpcConnector(httpFactory.create(upstream.getId(), chain), wsFactory, upstream.getId(), forkChoice) } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index 1dca02ba..8ca9d135 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt @@ -6,6 +6,8 @@ import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.MergedHead import io.emeraldpay.dshackle.upstream.ethereum.* +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.slf4j.LoggerFactory @@ -16,6 +18,7 @@ class EthereumRpcConnector( private val directReader : Reader, wsFactory: EthereumWsFactory?, id : String, + forkChoice: ForkChoice ) : EthereumConnector, CachesEnabled { private val conn : WsConnection? private val head : Head @@ -28,14 +31,14 @@ class EthereumRpcConnector( if (wsFactory != null) { // do not set upstream to the WS, since it doesn't control the RPC upstream conn = wsFactory.create(null, null, null) - val wsHead = EthereumWsHead(conn) + val wsHead = EthereumWsHead(conn, forkChoice) // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed - val rpcHead = EthereumRpcHead(directReader, Duration.ofSeconds(60)) - head = MergedHead(listOf(rpcHead, wsHead)) + val rpcHead = EthereumRpcHead(directReader, forkChoice, Duration.ofSeconds(60)) + head = MergedHead(listOf(rpcHead, wsHead), forkChoice) } else { conn = null log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC") - head = EthereumRpcHead(directReader) + head = EthereumRpcHead(directReader, forkChoice) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index fdb2e834..46bd437b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -7,6 +7,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead import io.emeraldpay.dshackle.upstream.ethereum.WsConnection +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient @@ -22,6 +23,7 @@ class EthereumWsConnector( upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain, + forkChoice: ForkChoice ) : EthereumConnector { private val conn: WsConnection private val api: Reader @@ -46,7 +48,7 @@ class EthereumWsConnector( ) conn = wsFactory.create(upstream, validator, metrics) - head = EthereumWsHead(conn) + head = EthereumWsHead(conn, forkChoice) api = JsonRpcWsClient(conn) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt new file mode 100644 index 00000000..87b4b42e --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt @@ -0,0 +1,22 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.upstream.DistanceExtractor +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.HeadLagObserver +import io.emeraldpay.dshackle.upstream.Upstream +import org.slf4j.LoggerFactory + +class EthereumPostHeadLagObserver( + master: Head, + followers: Collection +) : HeadLagObserver(master, followers, DistanceExtractor::extractPriorityDistance) { + + companion object { + private val log = LoggerFactory.getLogger(EthereumPostHeadLagObserver::class.java) + } + + override fun forkDistance(top: BlockContainer, curr: BlockContainer): Long { + return 6 + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt new file mode 100644 index 00000000..aab4ba37 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -0,0 +1,142 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * Copyright (c) 2020 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.ChainFees +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.MergedHead +import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice +import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.publisher.Mono + +@Suppress("UNCHECKED_CAST") +open class EthereumPosMultistream( + chain: Chain, + val upstreams: MutableList, + caches: Caches +) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)) { + + companion object { + private val log = LoggerFactory.getLogger(EthereumPosMultistream::class.java) + } + + private var head: Head? = null + + private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory()) + private val feeEstimation = EthereumPriorityFees(this, reader, 256) + init { + this.init() + } + + override fun init() { + if (upstreams.size > 0) { + head = updateHead() + } + super.init() + } + + override fun start() { + super.start() + reader.start() + } + + override fun stop() { + super.stop() + reader.stop() + } + + override fun isRunning(): Boolean { + return super.isRunning() || reader.isRunning + } + + open fun getReader(): EthereumReader { + return reader + } + + override fun getHead(): Head { + return head!! + } + + override fun setHead(head: Head) { + this.head = head + } + + override fun updateHead(): Head { + head?.let { + if (it is Lifecycle) { + it.stop() + } + } + lagObserver?.stop() + lagObserver = null + val head = if (upstreams.size == 1) { + val upstream = upstreams.first() + upstream.setLag(0) + upstream.getHead().apply { + if (this is Lifecycle) { + this.start() + } + } + } else { + val heads = upstreams.map { it.getHead() } + val newHead = MergedHead(heads, PriorityForkChoice()).apply { + this.start() + } + val lagObserver = EthereumPostHeadLagObserver(newHead, upstreams as Collection) + this.lagObserver = lagObserver + lagObserver.start() + newHead + } + onHeadUpdated(head) + return head + } + + override fun getLabels(): Collection { + return upstreams.flatMap { it.getLabels() } + } + + @Suppress("UNCHECKED_CAST") + override fun cast(selfType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + return this as T + } + + override fun getRoutedApi(matcher: Selector.Matcher): Mono> { + return Mono.just(LocalCallRouter(reader, getMethods(), getHead())) + } + + open fun getSubscribe(): EthereumSubscribe { + throw Error("Does not supports subscription for PoS ethereum") + } + + override fun getFeeEstimation(): ChainFees { + return feeEstimation + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt index 298b784a..1a75cf21 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt @@ -1,43 +1,106 @@ -package io.emeraldpay.dshackle.upstream.ethereum_pos +/** + * Copyright (c) 2020 EmeraldPay, Inc + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.Capability -import io.emeraldpay.dshackle.upstream.DefaultUpstream -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.grpc.Chain +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable -class EthereumPosUpstream( +open class EthereumPosUpstream( id: String, + val chain: Chain, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, - node: QuorumForLabels.QuorumItem?, - private val ethereumUpstream: EthereumUpstream -) : DefaultUpstream(id, options, role, targets, node) { - override fun getCapabilities(): Set { - return ethereumUpstream.getCapabilities() + private val node: QuorumForLabels.QuorumItem?, + connectorFactory: ConnectorFactory +) : DefaultUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled { + private val log = LoggerFactory.getLogger(EthereumPosUpstream::class.java) + private val validator : EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions()) + private val connector : EthereumConnector = connectorFactory.create(this, validator, chain) + + private var validatorSubscription: Disposable? = null + + override fun setCaches(caches: Caches) { + if (connector is CachesEnabled) { + connector.setCaches(caches) + } } - override fun getLabels(): Collection { - return ethereumUpstream.getLabels() + override fun start() { + log.info("Configured for ${chain.chainName}") + connector.start() + if (getOptions().disableValidation != null && getOptions().disableValidation!!) { + log.warn("Disable validation for upstream ${this.getId()}") + this.setLag(0) + this.setStatus(UpstreamAvailability.OK) + } else { + log.debug("Start validation for upstream ${this.getId()}") + validatorSubscription = validator.start() + .subscribe(this::setStatus) + } + } + override fun getHead(): Head { + return connector.getHead() } - override fun getHead() : Head { - return ethereumUpstream.getHead() + override fun stop() { + validatorSubscription?.dispose() + validatorSubscription = null + connector.stop() + } + + override fun isRunning(): Boolean { + return connector.isRunning + } + + override fun getApi(): Reader { + return connector.getApi() } override fun isGrpc(): Boolean { return false } - override fun getApi(): Reader { - return ethereumUpstream.getApi() + private val capabilities = if (options.providesBalance != false) { + setOf(Capability.RPC, Capability.BALANCE) + } else { + setOf(Capability.RPC) + } + + override fun getCapabilities(): Set { + return capabilities + } + + override fun getLabels(): Collection { + return node?.let { listOf(it.labels) } ?: emptyList() } @Suppress("UNCHECKED_CAST") @@ -47,4 +110,4 @@ class EthereumPosUpstream( } return this as T } -} \ No newline at end of file +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt new file mode 100644 index 00000000..aa664241 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt @@ -0,0 +1,17 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer + +interface ForkChoice { + + sealed class ChoiceResult { + data class Updated(val nwhead: BlockContainer): ChoiceResult() + data class Same(val head: BlockContainer?): ChoiceResult() + } + + fun getHead(): BlockContainer? + + fun filter(block: BlockContainer): Boolean + + fun choose(block: BlockContainer): ChoiceResult +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoice.kt new file mode 100644 index 00000000..d4c5e9c0 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoice.kt @@ -0,0 +1,32 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer +import java.util.concurrent.atomic.AtomicReference + +class MostWorkForkChoice : ForkChoice { + private val head = AtomicReference(null) + + override fun getHead() : BlockContainer? { + return head.get() + } + + override fun filter(block: BlockContainer): Boolean { + val curr = head.get() + return curr == null || curr.difficulty < block.difficulty + } + + override fun choose(block: BlockContainer): ForkChoice.ChoiceResult { + val nwhead = head.updateAndGet { curr -> + if (filter(block)) { + block + } else { + curr + } + } + if (nwhead.hash == block.hash) { + return ForkChoice.ChoiceResult.Updated(nwhead) + } + return ForkChoice.ChoiceResult.Same(nwhead) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt new file mode 100644 index 00000000..523f4547 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt @@ -0,0 +1,36 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.RingSet +import java.util.concurrent.atomic.AtomicReference + +class NoChoiceWithPriorityForkChoice( + private val nodeRating: Int +): ForkChoice { + private val head = AtomicReference(null) + private val seenBlocks = RingSet(10) + + override fun getHead(): BlockContainer? { + return head.get() + } + + override fun filter(block: BlockContainer): Boolean { + return !seenBlocks.contains(block.hash) + } + + override fun choose(block: BlockContainer): ForkChoice.ChoiceResult { + val nwhead = head.updateAndGet { curr -> + if (!filter(block)) { + curr + } else { + seenBlocks.add(block.hash) + block.copyWithRating(nodeRating) + } + } + if (nwhead.hash == block.hash) { + return ForkChoice.ChoiceResult.Updated(nwhead) + } + return ForkChoice.ChoiceResult.Same(nwhead) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt new file mode 100644 index 00000000..e9ec338b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt @@ -0,0 +1,35 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.RingSet +import java.util.concurrent.atomic.AtomicReference + +class PriorityForkChoice: ForkChoice { + private val head = AtomicReference(null) + private val seenBlocks = RingSet(10) + + override fun getHead(): BlockContainer? { + return head.get() + } + + override fun filter(block: BlockContainer): Boolean { + val curr = head.get() + return (curr == null || curr.nodeRating <= block.nodeRating) && !seenBlocks.contains(block.hash) + } + + override fun choose(block: BlockContainer): ForkChoice.ChoiceResult { + val nwhead = head.updateAndGet { curr -> + if (!filter(block)) { + curr + } else { + seenBlocks.add(block.hash) + block + } + } + if (nwhead.hash == block.hash) { + return ForkChoice.ChoiceResult.Updated(nwhead) + } + return ForkChoice.ChoiceResult.Same(nwhead) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 542688d3..e4205af9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -29,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -95,7 +96,7 @@ class BitcoinGrpcUpstream( } } private val upstreamStatus = GrpcUpstreamStatus() - private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock) + private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) var timeout = Defaults.timeout private var capabilities: Set = emptySet() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 7a3c5006..b7824b01 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -93,7 +94,7 @@ open class EthereumGrpcUpstream( private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) private val upstreamStatus = GrpcUpstreamStatus() - private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock) + private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) private var capabilities: Set = emptySet() private val defaultReader: Reader = client.forSelector(Selector.empty) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt new file mode 100644 index 00000000..ffa5ae06 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -0,0 +1,162 @@ +/** + * Copyright (c) 2020 EmeraldPay, Inc + * Copyright (c) 2019 ETCDEV GmbH + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.grpc + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.ReactorBlockchainGrpc +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.* +import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.rpc.RpcException +import io.emeraldpay.grpc.Chain +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.publisher.Mono +import java.math.BigInteger +import java.time.Instant +import java.util.Locale +import java.util.concurrent.TimeoutException +import java.util.function.Function + +open class EthereumPosGrpcUpstream( + private val parentId: String, + role: UpstreamsConfig.UpstreamRole, + private val chain: Chain, + private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, + private val client: JsonRpcGrpcClient +) : DefaultUpstream( + "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}", + UpstreamsConfig.Options.getDefaults(), + role, + null, null +), + GrpcUpstream, + Lifecycle { + + private val blockConverter: Function = Function { value -> + val block = BlockContainer( + value.height, + BlockId.from(BlockHash.from("0x" + value.blockId)), + BigInteger(1, value.weight.toByteArray()), + Instant.ofEpochMilli(value.timestamp), + false, + null, + null + ) + block + } + + private val reloadBlock: Function> = Function { existingBlock -> + // head comes without transaction data + // need to download transactions for the block + defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(existingBlock.hash.toHexWithPrefix(), false))) + .flatMap(JsonRpcResponse::requireResult) + .map { + BlockContainer.fromEthereumJson(it) + } + .timeout(timeout, Mono.error(TimeoutException("Timeout from upstream"))) + .doOnError { t -> + setStatus(UpstreamAvailability.UNAVAILABLE) + val msg = "Failed to download block data for chain $chain on $parentId" + if (t is RpcException || t is TimeoutException) { + log.warn("$msg. Message: ${t.message}") + } else { + log.error(msg, t) + } + } + } + + private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) + private val upstreamStatus = GrpcUpstreamStatus() + private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) + private var capabilities: Set = emptySet() + + private val defaultReader: Reader = client.forSelector(Selector.empty) + var timeout = Defaults.timeout + + override fun start() { + } + + override fun isRunning(): Boolean { + return true + } + + override fun stop() { + } + + override fun update(conf: BlockchainOuterClass.DescribeChain) { + upstreamStatus.update(conf) + capabilities = RemoteCapabilities.extract(conf) + conf.status?.let { status -> onStatus(status) } + } + + override fun getQuorumByLabel(): QuorumForLabels { + return upstreamStatus.getNodes() + } + + // ------------------------------------------------------------------------------------------ + + override fun getLabels(): Collection { + return upstreamStatus.getLabels() + } + + override fun getMethods(): CallMethods { + return upstreamStatus.getCallMethods() + } + + override fun isAvailable(): Boolean { + return super.isAvailable() && grpcHead.getCurrent() != null && getQuorumByLabel().getAll().any { + it.quorum > 0 + } + } + + override fun getHead(): Head { + return grpcHead + } + + override fun getApi(): Reader { + return defaultReader + } + + @Suppress("UNCHECKED_CAST") + override fun cast(selfType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + return this as T + } + + override fun getCapabilities(): Set { + return capabilities + } + + override fun isGrpc(): Boolean { + return true + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index 2e7e8bba..d8434fbe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.grpc.Chain import org.reactivestreams.Publisher import org.slf4j.LoggerFactory @@ -45,8 +46,9 @@ class GrpcHead( /** * Populate block data with all missing details, of any */ - private val enhancer: Function>? -) : AbstractHead(), Lifecycle { + private val enhancer: Function>?, + private val forkChoice: ForkChoice +) : AbstractHead(forkChoice), Lifecycle { companion object { private val log = LoggerFactory.getLogger(GrpcHead::class.java) @@ -94,10 +96,7 @@ class GrpcHead( var blocks = source.map(converter) .distinctUntilChanged { it.hash - }.filter { block -> - val curr = this.getCurrent() - curr == null || curr.difficulty < block.difficulty - } + }.filter { forkChoice.filter(it) } if (enhancer != null) { blocks = blocks.flatMap(enhancer) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy index 541a5e61..cdb5c8bd 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy @@ -27,7 +27,7 @@ class HeightByHashAddingSpec extends Specification { def block = new BlockContainer( 12079192L, BlockId.from("0xa6af163aab691919c595e2a466f0a7b01f1dff8cfd9631dee811df57064c2d32"), - BigInteger.ONE, Instant.now(), false, "".bytes, null, [] + BigInteger.ONE, Instant.now(), false, "".bytes, null, [], 0 ) def "use memory if available"() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy index 01c88ba7..d950a765 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy @@ -86,7 +86,8 @@ class ReceiptMemCacheSpec extends Specification { false, "{}".bytes, null, - [TxId.from(receipt.transactionHash)] + [TxId.from(receipt.transactionHash)], + 0 ) when: diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy index 59a9008d..143febc5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy @@ -271,7 +271,7 @@ class TrackBitcoinAddressSpec extends Specification { Head head = Mock(Head) { 1 * getFlux() >> Flux.concat( Flux.just( - new BlockContainer(0L, BlockId.from(hash1), BigInteger.ZERO, Instant.now(), false, null, null, []) + new BlockContainer(0L, BlockId.from(hash1), BigInteger.ZERO, Instant.now(), false, null, null, [], 0) ), blocks.asFlux() ) @@ -312,7 +312,7 @@ class TrackBitcoinAddressSpec extends Specification { StepVerifier.create(resp) .expectNext("0") .then { - blocks.tryEmitNext(new BlockContainer(1L, BlockId.from(hash1), BigInteger.ONE, Instant.now(), false, null, null, [])) + blocks.tryEmitNext(new BlockContainer(1L, BlockId.from(hash1), BigInteger.ONE, Instant.now(), false, null, null, [], 0)) } .expectNext("1230000") .then { diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy index 4a04dee5..b1f36eba 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy @@ -142,7 +142,7 @@ class TrackBitcoinTxSpec extends Specification { def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" // start with the current block def next = Flux.fromIterable([10, 12, 13, 14, 15]).map { h -> - new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, []) + new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0) } Head head = Mock(Head) { 1 * getFlux() >> next @@ -173,7 +173,7 @@ class TrackBitcoinTxSpec extends Specification { def txid = "69cd44d7c641db82e69824523c7ac0c5c1e5628f025474529cf5ffe64527efc9" // start with the current block def next = Flux.fromIterable([10, 12, 13]).map { h -> - new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, []) + new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0) } Head head = Mock(Head) { 1 * getFlux() >> next @@ -268,7 +268,7 @@ class TrackBitcoinTxSpec extends Specification { ]) } def next = Flux.fromIterable([10, 11, 12]).map { h -> - new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, []) + new BlockContainer(h.longValue(), BlockId.from("0000000000000000000895d1b9d3898700e1deecc3b0e69f439aa77875e6042f"), BigInteger.ONE, Instant.now(), false, null, null, [], 0) } Head head = Mock(Head) { _ * getFlux() >> next diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index d371d4a4..71fedb56 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy @@ -198,7 +198,7 @@ class TrackEthereumTxSpec extends Specification { def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) def block = new BlockContainer( 100, BlockId.from(txId), BigInteger.ONE, Instant.now(), false, "".bytes, null, - [TxId.from(txId)] + [TxId.from(txId)], 0 ) when: @@ -220,7 +220,8 @@ class TrackEthereumTxSpec extends Specification { def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) def block = new BlockContainer( 100, BlockId.from(txId), BigInteger.ONE, Instant.now(), false, "".bytes, null, - [TxId.from("0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22")] + [TxId.from("0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22")], + 0 ) apiMock.answer("eth_getTransactionByHash", [txId], null) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index a9203234..795995db 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -111,7 +111,8 @@ class TestingCommons { false, null, null, - [] + [], + 0 ) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy index 439b811d..48bc614b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy @@ -17,6 +17,9 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice +import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux import reactor.core.publisher.Sinks import reactor.test.StepVerifier @@ -31,7 +34,7 @@ class AbstractHeadSpec extends Specification { def blocks = [1L, 2, 3, 4].collect { i -> byte[] hash = new byte[32] hash[0] = i as byte - new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, []) + new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0) } def "Calls beforeBlock on each block"() { @@ -85,7 +88,7 @@ class AbstractHeadSpec extends Specification { .verify(Duration.ofSeconds(1)) } - def "Ignores block will less difficulty"() { + def "Ignores block that is filtered by forkchoice"() { setup: Sinks.Many source = Sinks.many().unicast().onBackpressureBuffer() def head = new TestHead() @@ -93,7 +96,7 @@ class AbstractHeadSpec extends Specification { blocks[1].height, BlockId.from(blocks[1].hash.value.clone().tap { it[1] = 0xff as byte }), blocks[1].difficulty - 1, Instant.now(), - false, null, null, [] + false, null, null, [], 0 ) when: head.follow(source.asFlux()) @@ -113,6 +116,23 @@ class AbstractHeadSpec extends Specification { } class TestHead extends AbstractHead { + TestHead() { + super(new ForkChoice() { + @Override + boolean filter(@NotNull BlockContainer block) { + return block.hash != BlockId.from("02ff000000000000000000000000000000000000000000000000000000000000") + } + @Override + ForkChoice.ChoiceResult choose(@NotNull BlockContainer block) { + return new ForkChoice.ChoiceResult.Updated(block) + } + + @Override + BlockContainer getHead() { + return null + } + }) + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/DistanceExtractorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/DistanceExtractorSpec.groovy new file mode 100644 index 00000000..0fe56a8a --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/DistanceExtractorSpec.groovy @@ -0,0 +1,74 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.rpc.json.BlockJson +import spock.lang.Specification + +import java.time.Instant + +class DistanceExtractorSpec extends Specification { + def "Correct distance for PoW"() { + expect: + def top = new BlockJson().with { + it.number = topHeight + it.totalDifficulty = topDiff + it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915123") + it.timestamp = Instant.now() + return it + } + def curr = new BlockJson().with { + it.number = currHeight + it.totalDifficulty = currDiff + it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915123") + it.timestamp = Instant.now() + return it + } + delta as DistanceExtractor.ChainDistance == DistanceExtractor.@Companion.extractPowDistance(BlockContainer.from(top), BlockContainer.from(curr)) + where: + topHeight | topDiff | currHeight | currDiff | delta + 100 | 1000 | 100 | 1000 | new DistanceExtractor.ChainDistance.Distance(0) + 101 | 1010 | 100 | 1000 | new DistanceExtractor.ChainDistance.Distance(1) + 102 | 1020 | 100 | 1000 | new DistanceExtractor.ChainDistance.Distance(2) + 103 | 1030 | 100 | 1000 | new DistanceExtractor.ChainDistance.Distance(3) + 150 | 1500 | 100 | 1000 | new DistanceExtractor.ChainDistance.Distance(50) + + 100 | 1000 | 101 | 1010 | new DistanceExtractor.ChainDistance.Distance(0) + 100 | 1000 | 102 | 1020 | new DistanceExtractor.ChainDistance.Distance(0) + 100 | 1000 | 100 | 1010 | DistanceExtractor.ChainDistance.Fork.INSTANCE + 100 | 1100 | 100 | 1000 | DistanceExtractor.ChainDistance.Fork.INSTANCE + } + + def "Correct distance for priority"() { + setup: + def hash1 = "0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915123" + def hash2 = "0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915124" + expect: + def top = new BlockJson().with { + it.number = topHeight + it.totalDifficulty = 0 + it.hash = BlockHash.from(hashA == 0 ? hash1 : hash2) + it.timestamp = Instant.now() + return it + } + def curr = new BlockJson().with { + it.number = currHeight + it.totalDifficulty = 0 + it.hash = BlockHash.from(hashB == 0 ? hash1 : hash2) + it.timestamp = Instant.now() + return it + } + delta as DistanceExtractor.ChainDistance == DistanceExtractor.@Companion.extractPriorityDistance(BlockContainer.from(top), BlockContainer.from(curr)) + where: + topHeight | hashA | currHeight | hashB || delta + 100 | 0 | 100 | 0 || new DistanceExtractor.ChainDistance.Distance(0) + 101 | 0 | 100 | 1 || new DistanceExtractor.ChainDistance.Distance(1) + 102 | 0 | 100 | 1 || new DistanceExtractor.ChainDistance.Distance(2) + 103 | 0 | 100 | 1 || new DistanceExtractor.ChainDistance.Distance(3) + 150 | 0 | 100 | 1 || new DistanceExtractor.ChainDistance.Distance(50) + + 100 | 0 | 101 | 1 || DistanceExtractor.ChainDistance.Fork.INSTANCE + 100 | 0 | 102 | 1 || DistanceExtractor.ChainDistance.Fork.INSTANCE + 100 | 0 | 100 | 1 || DistanceExtractor.ChainDistance.Fork.INSTANCE + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index eefdf46a..371fb461 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.grpc.Chain import reactor.test.StepVerifier import spock.lang.Retry @@ -47,7 +48,7 @@ class FilteredApisSpec extends Specification { def httpFactory = Mock(HttpFactory) { create(_, _) >> TestingCommons.api().tap { it.id = "${i++}" } } - def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory) + def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice()) new EthereumUpstream( "test", Chain.ETHEREUM, diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy index fa511ed1..0a235796 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy @@ -111,45 +111,10 @@ class HeadLagObserverSpec extends Specification { .verifyComplete() } - def "Correct distance"() { - setup: - Head master = Mock() - HeadLagObserver observer = new TestHeadLagObserver(master, []) - expect: - def top = new BlockJson().with { - it.number = topHeight - it.totalDifficulty = topDiff - it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915123") - it.timestamp = Instant.now() - return it - } - def curr = new BlockJson().with { - it.number = currHeight - it.totalDifficulty = currDiff - it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915123") - it.timestamp = Instant.now() - return it - } - delta as Long == observer.extractDistance(BlockContainer.from(top), BlockContainer.from(curr)) - where: - topHeight | topDiff | currHeight | currDiff | delta - 100 | 1000 | 100 | 1000 | 0 - 101 | 1010 | 100 | 1000 | 1 - 102 | 1020 | 100 | 1000 | 2 - 103 | 1030 | 100 | 1000 | 3 - 150 | 1500 | 100 | 1000 | 50 - - 100 | 1000 | 101 | 1010 | 0 - 100 | 1000 | 102 | 1020 | 0 - 100 | 1000 | 100 | 1010 | 11 - 100 | 1100 | 100 | 1000 | 11 - - } - class TestHeadLagObserver extends HeadLagObserver { TestHeadLagObserver(@NotNull Head master, @NotNull Collection followers) { - super(master, followers) + super(master, followers, DistanceExtractor.@Companion::extractPowDistance) } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy index 217f2f74..a5b747ef 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import org.springframework.context.Lifecycle import reactor.core.publisher.Flux import spock.lang.Specification @@ -36,7 +37,7 @@ class MergedHeadSpec extends Specification { } when: - def merged = new MergedHead([head1, head2, head3]) + def merged = new MergedHead([head1, head2, head3], new MostWorkForkChoice()) merged.start() then: @@ -44,11 +45,17 @@ class MergedHeadSpec extends Specification { } class TestHead1 extends AbstractHead { - + TestHead1() { + super(new MostWorkForkChoice()) + } } class TestHead2 extends AbstractHead implements Lifecycle { + TestHead2() { + super(new MostWorkForkChoice()) + } + @Override void start() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy index 943edf8c..94fe5fde 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy @@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.rpc.json.BlockJson import reactor.core.publisher.Flux @@ -30,7 +31,7 @@ import java.time.Instant class DefaultEthereumHeadSpec extends Specification { - DefaultEthereumHead head = new DefaultEthereumHead() + DefaultEthereumHead head = new DefaultEthereumHead(new MostWorkForkChoice()) ObjectMapper objectMapper = Global.objectMapper def blocks = (10L..20L).collect { i -> diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoiceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoiceSpec.groovy new file mode 100644 index 00000000..dac1ec85 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoiceSpec.groovy @@ -0,0 +1,37 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import spock.lang.Specification + +import java.time.Instant + +class MostWorkForkChoiceSpec extends Specification { + + def blocks = [1L, 2, 3, 4].collect { i -> + byte[] hash = new byte[32] + hash[0] = i as byte + new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0) + } + + def "filters blocks"() { + def choice = new MostWorkForkChoice() + choice.choose(blocks[1]) + expect: + !choice.filter(blocks[0]) + choice.filter(blocks[2]) + } + + def "chooses correct block as head"() { + def choice = new MostWorkForkChoice() + choice.choose(blocks[1]) + when: + choice.choose(blocks[0]) + then: + choice.getHead() == blocks[1] + when: + choice.choose(blocks[2]) + then: + choice.getHead() == blocks[2] + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoiceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoiceSpec.groovy new file mode 100644 index 00000000..20f90dae --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoiceSpec.groovy @@ -0,0 +1,51 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import spock.lang.Specification + +import java.time.Instant + +class NoChoiceWithPriorityForkChoiceSpec extends Specification { + def blocks = [1L, 2, 3, 4].collect { i -> + byte[] hash = new byte[32] + hash[0] = i as byte + new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], 0) + } + + def "filters blocks"() { + def blockR0 = blocks[0].copyWithRating(10) + def blockR1 = blocks[1].copyWithRating(10) + def choice = new NoChoiceWithPriorityForkChoice(10) + when: + choice.choose(blocks[0]) + then: + choice.getHead() == blockR0 + when: + choice.choose(blocks[1]) + then: + choice.getHead() == blockR1 + when: + choice.choose(blocks[0]) + then: + choice.getHead() == blocks[1] + } + + def "chooses blocks and adds rating"() { + def blockR0 = blocks[0].copyWithRating(10) + def blockR1 = blocks[1].copyWithRating(10) + def choice = new NoChoiceWithPriorityForkChoice(10) + when: + choice.choose(blocks[0]) + then: + choice.getHead() == blockR0 + when: + choice.choose(blocks[1]) + then: + choice.getHead() == blockR1 + when: + choice.choose(blocks[0]) + then: + choice.getHead() == blockR1 + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy new file mode 100644 index 00000000..25848281 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy @@ -0,0 +1,41 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import spock.lang.Specification + +import java.time.Instant + +class PriorityForkChoiceSpec extends Specification { + def blocks = [1L, 2, 3, 4].collect { i -> + byte[] hash = new byte[32] + hash[0] = i as byte + new BlockContainer(i, BlockId.from(hash), BigInteger.valueOf(i), Instant.now(), false, null, null, [], i.toInteger()) + } + def "filters blocks"() { + def choice = new PriorityForkChoice() + choice.choose(blocks[1]) + expect: + !choice.filter(blocks[0]) + choice.filter(blocks[2]) + !choice.filter(blocks[1]) + } + + def "chooses correct block according to node rating"() { + def choice = new PriorityForkChoice() + choice.choose(blocks[1]) + when: + choice.choose(blocks[0]) + then: + choice.getHead() == blocks[1] + when: + choice.choose(blocks[2]) + then: + choice.getHead() == blocks[2] + when: + def seenblock = blocks[1].copyWithRating(20) + choice.choose(seenblock) + then: + choice.getHead() == blocks[2] + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index 073c26d5..0f8a1e49 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy @@ -21,6 +21,7 @@ import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.test.MockGrpcServer import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import reactor.test.StepVerifier @@ -60,7 +61,7 @@ class GrpcHeadSpec extends Specification { Chain.BITCOIN, Stub(DefaultUpstream), client, - convert, null + convert, null, new MostWorkForkChoice() ) when: def act = head.getFlux() @@ -121,7 +122,7 @@ class GrpcHeadSpec extends Specification { Chain.BITCOIN, Stub(DefaultUpstream), client, - convert, null + convert, null, new MostWorkForkChoice() ) when: def act = head.getFlux()