StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactoring fork choice rules out of abstract head and priority fork choice support for PoS Ethereum

Browse Source
This commit is contained in:
terminal
2022-07-29 15:32:48 +04:00
parent 92218623f2
commit eb95c591f7
51 changed files with 995 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt
View File

@@ -30,7 +30,8 @@ class BlockContainer(
val full: Boolean,
json: ByteArray?,
val parsed: Any?,
val transactions: List<TxId> = emptyList()
val transactions: List<TxId> = 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()

44
src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt Normal file
View File

@@ -0,0 +1,44 @@
package io.emeraldpay.dshackle.data
import java.util.concurrent.atomic.AtomicReference
class RingSet<T>(
private val maxSize: Int
): Set<T> {
private var seqValues: AtomicReference<List<T>> = AtomicReference(emptyList())
private var set: Set<T> = 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<T> {
return set.iterator()
}
override fun containsAll(elements: Collection<@UnsafeVariance T>): Boolean {
return set.containsAll(elements)
}
}

60
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -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<UpstreamsConfig.EthereumPosConnection>,
// 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<UpstreamsConfig.EthereumPosConnection>,
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<URI>()
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<UpstreamsConfig.BitcoinConnection>,
@@ -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<URI>()
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<URI>): EthereumConnectorFactory? {
private fun buildEthereumConnectorFactory(conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList<URI>, 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

37
src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt
View File

@@ -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<BlockContainer>(null)
private var stream = Sinks.many().multicast().directBestEffort<BlockContainer>()
private var completed = false
private val beforeBlockHandlers = ArrayList<Runnable>()
@@ -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<BlockContainer> {
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? {

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -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<Multistream> {
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

28
src/main/kotlin/io/emeraldpay/dshackle/upstream/DistanceExtractor.kt Normal file
View File

@@ -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)
}
}
}
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt
View File

@@ -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<Upstream>
private val followers: Collection<Upstream>,
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)
}
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/MergedHead.kt
View File

@@ -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<Head>
) : AbstractHead(), Lifecycle, CachesEnabled {
private val sources: Iterable<Head>,
forkChoice: ForkChoice
) : AbstractHead(forkChoice), Lifecycle, CachesEnabled {
private var subscription: Disposable? = null

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinHeadLagObserver.kt
View File

@@ -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<Upstream>
) : HeadLagObserver(master, followers) {
) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) {
companion object {
private val log = LoggerFactory.getLogger(BitcoinHeadLagObserver::class.java)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
View File

@@ -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)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
View File

@@ -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<JsonRpcRequest, JsonRpcResponse>,
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)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
View File

@@ -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<JsonRpcRequest, JsonRpcResponse>,
private val extractBlock: ExtractBlock,
) : Head, AbstractHead(), Lifecycle {
) : Head, AbstractHead(MostWorkForkChoice()), Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinZMQHead::class.java)

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
View File

@@ -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)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFees.kt
View File

@@ -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<EthereumFees.EthereumFee, BlockJson<TransactionRefJson>, TransactionRefJson, TransactionJson>(heightLimit, upstreams, extractTx), ChainFees {

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt
View File

@@ -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<Upstream>
) : HeadLagObserver(master, followers) {
) : HeadLagObserver(master, followers, DistanceExtractor::extractPowDistance) {
companion object {
private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
View File

@@ -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<Upstream>)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumPriorityFees.kt
View File

@@ -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 {

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
View File

@@ -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<JsonRpcRequest, JsonRpcResponse>,
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 =

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
View File

@@ -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)
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
View File

@@ -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 {

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -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)

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
View File

@@ -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)
}
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
View File

@@ -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<JsonRpcRequest, JsonRpcResponse>,
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)
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
View File

@@ -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<JsonRpcRequest, JsonRpcResponse>
@@ -46,7 +48,7 @@ class EthereumWsConnector(
)
conn = wsFactory.create(upstream, validator, metrics)
head = EthereumWsHead(conn)
head = EthereumWsHead(conn, forkChoice)
api = JsonRpcWsClient(conn)
}

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosHeadLagObserver.kt Normal file
View File

@@ -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<Upstream>
) : 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
}
}

142
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt Normal file
View File

@@ -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<EthereumPosUpstream>,
caches: Caches
) : Multistream(chain, upstreams as MutableList<Upstream>, 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<Upstream>)
this.lagObserver = lagObserver
lagObserver.start()
newHead
}
onHeadUpdated(head)
return head
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return upstreams.flatMap { it.getLabels() }
}
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): 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<Reader<JsonRpcRequest, JsonRpcResponse>> {
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
}
}

101
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt
View File

@@ -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<Capability> {
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<UpstreamsConfig.Labels> {
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<JsonRpcRequest, JsonRpcResponse> {
return connector.getApi()
}
override fun isGrpc(): Boolean {
return false
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
return ethereumUpstream.getApi()
private val capabilities = if (options.providesBalance != false) {
setOf(Capability.RPC, Capability.BALANCE)
} else {
setOf(Capability.RPC)
}
override fun getCapabilities(): Set<Capability> {
return capabilities
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return node?.let { listOf(it.labels) } ?: emptyList()
}
@Suppress("UNCHECKED_CAST")
@@ -47,4 +110,4 @@ class EthereumPosUpstream(
}
return this as T
}
}
}

17
src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt Normal file
View File

@@ -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
}

32
src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoice.kt Normal file
View File

@@ -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<BlockContainer>(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)
}
}

36
src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt Normal file
View File

@@ -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<BlockContainer>(null)
private val seenBlocks = RingSet<BlockId>(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)
}
}

35
src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt Normal file
View File

@@ -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<BlockContainer>(null)
private val seenBlocks = RingSet<BlockId>(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)
}
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -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<Capability> = emptySet()

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
View File

@@ -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<Capability> = emptySet()
private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)

162
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt Normal file
View File

@@ -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<BlockchainOuterClass.ChainHead, BlockContainer> = 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<BlockContainer, Publisher<BlockContainer>> = 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<Capability> = emptySet()
private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = 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<UpstreamsConfig.Labels> {
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<JsonRpcRequest, JsonRpcResponse> {
return defaultReader
}
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return this as T
}
override fun getCapabilities(): Set<Capability> {
return capabilities
}
override fun isGrpc(): Boolean {
return true
}
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt
View File

@@ -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<BlockContainer, Publisher<BlockContainer>>?
) : AbstractHead(), Lifecycle {
private val enhancer: Function<BlockContainer, Publisher<BlockContainer>>?,
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)
}

2
src/test/groovy/io/emeraldpay/dshackle/cache/HeightByHashAddingSpec.groovy vendored
View File

@@ -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"() {

3
src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy vendored
View File

@@ -86,7 +86,8 @@ class ReceiptMemCacheSpec extends Specification {
false,
"{}".bytes,
null,
[TxId.from(receipt.transactionHash)]
[TxId.from(receipt.transactionHash)],
0
)
when:

4
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinAddressSpec.groovy
View File

@@ -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 {

6
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackBitcoinTxSpec.groovy
View File

@@ -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

5
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy
View File

@@ -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)

3
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
View File

@@ -111,7 +111,8 @@ class TestingCommons {
false,
null,
null,
[]
[],
0
)
}

26
src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy
View File

@@ -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<BlockContainer> 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
}
})
}
}
}

74
src/test/groovy/io/emeraldpay/dshackle/upstream/DistanceExtractorSpec.groovy Normal file
View File

@@ -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
}
}

3
src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
View File

@@ -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,

37
src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy
View File

@@ -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<? extends Upstream> followers) {
super(master, followers)
super(master, followers, DistanceExtractor.@Companion::extractPowDistance)
}
@Override

11
src/test/groovy/io/emeraldpay/dshackle/upstream/MergedHeadSpec.groovy
View File

@@ -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() {

3
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy
View File

@@ -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 ->

37
src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/MostWorkForkChoiceSpec.groovy Normal file
View File

@@ -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]
}
}

51
src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoiceSpec.groovy Normal file
View File

@@ -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
}
}

41
src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy Normal file
View File

@@ -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]
}
}

5
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy
View File

@@ -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()