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

Merge pull request #60 from p2p-org/fix-heads

Browse Source 
Fix head updates in case of priority fork choice
This commit is contained in:
Vyacheslav Shebanov
2022-12-03 16:10:41 +02:00
committed by GitHub
parent acaeb0c127 5e15096da5
commit 0fd4a597fb
5 changed files with 25 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -53,7 +53,7 @@ abstract class AbstractHead @JvmOverloads constructor(
{ {
val delay = System.currentTimeMillis() - lastHeadUpdated val delay = System.currentTimeMillis() - lastHeadUpdated
if (delay > awaitHeadTimeoutMs) { if (delay > awaitHeadTimeoutMs) {
log.warn("No head updates for $delay ms @ ${this.javaClass} - restart") log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass} - restart")
if (lock.tryLock()) { if (lock.tryLock()) {
try { try {
start() start()
@@ -73,19 +73,19 @@ abstract class AbstractHead @JvmOverloads constructor(
completed = false completed = false
} }
return source return source
.distinctUntilChanged { .filter {
it.hash log.debug("Filtering block $upstreamId block $it")
forkChoice.filter(it)
} }
.filter { forkChoice.filter(it) }
.doFinally { .doFinally {
// close internal stream if upstream is finished, otherwise it gets stuck, // close internal stream if upstream is finished, otherwise it gets stuck,
// but technically it should never happen during normal work, only when the Head // but technically it should never happen during normal work, only when the Head
// is stopping // is stopping
if (it == SignalType.ON_ERROR && !stopping) { if (it == SignalType.ON_ERROR && !stopping) {
log.warn("Received signal $it unexpectedly - restart head") log.warn("Received signal $upstreamId $it unexpectedly - restart head")
lastHeadUpdated = 0L lastHeadUpdated = 0L
} else { } else {
log.warn("Received signal $it - stop emit new head!!!") log.warn("Received signal $upstreamId $it - stop emit new head!!!")
completed = true completed = true
stream.tryEmitComplete() stream.tryEmitComplete()
} }
@@ -95,7 +95,7 @@ abstract class AbstractHead @JvmOverloads constructor(
val valid = runCatching { val valid = runCatching {
blockValidator.isValid(forkChoice.getHead(), block) blockValidator.isValid(forkChoice.getHead(), block)
}.onFailure { }.onFailure {
log.error("Block ${block.hash} validation failed with '${it.message}'", it) log.error("Block $upstreamId ${block.hash} validation failed with '${it.message}'", it)
}.getOrElse { false } }.getOrElse { false }
if (valid) { if (valid) {
notifyBeforeBlock() notifyBeforeBlock()
@@ -113,7 +113,7 @@ abstract class AbstractHead @JvmOverloads constructor(
is ForkChoice.ChoiceResult.Same -> {} is ForkChoice.ChoiceResult.Same -> {}
} }
} else { } else {
log.warn("Invalid block $block}") log.warn("Invalid block $upstreamId $block}")
} }
} }
} }

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

@@ -24,10 +24,11 @@ import org.slf4j.LoggerFactory
import reactor.core.Disposable import reactor.core.Disposable
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
class MergedHead( class MergedHead @JvmOverloads constructor(
private val sources: Iterable<Head>, private val sources: Iterable<Head>,
forkChoice: ForkChoice forkChoice: ForkChoice,
) : AbstractHead(forkChoice), Lifecycle, CachesEnabled { private val label: String = ""
) : AbstractHead(forkChoice, upstreamId = label), Lifecycle, CachesEnabled {
companion object { companion object {
private val log = LoggerFactory.getLogger(MergedHead::class.java) private val log = LoggerFactory.getLogger(MergedHead::class.java)
@@ -48,7 +49,9 @@ class MergedHead(
} }
subscription?.dispose() subscription?.dispose()
subscription = super.follow( subscription = super.follow(
Flux.merge(sources.map { it.getFlux() }).doOnNext { log.debug("New MERGED head $it") } Flux.merge(sources.map { it.getFlux() }).doOnNext {
log.debug("New MERGED $label head $it")
}
) )
} }

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

@@ -128,7 +128,7 @@ open class EthereumMultistream(
} }
} else { } else {
val heads = upstreams.map { it.getHead() } val heads = upstreams.map { it.getHead() }
val newHead = MergedHead(heads, MostWorkForkChoice()).apply { val newHead = MergedHead(heads, MostWorkForkChoice(), "ETH Multistream").apply {
this.start() this.start()
} }
val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream>) val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream>)
@@ -166,13 +166,12 @@ open class EthereumMultistream(
upstreams.filter { mather.matches(it) } upstreams.filter { mather.matches(it) }
.apply { .apply {
log.debug("Found $size upstreams matching [${mather.describeInternal()}]") log.debug("Found $size upstreams matching [${mather.describeInternal()}]")
} }.let {
.map { it.getHead() } val selected = it.map { it.getHead() }
.let {
when (it.size) { when (it.size) {
0 -> EmptyHead() 0 -> EmptyHead()
1 -> it.first() 1 -> selected.first()
else -> MergedHead(it, MostWorkForkChoice()).apply { else -> MergedHead(selected, MostWorkForkChoice(), "Eth head ${it.map { it.getId() }}").apply {
start() start()
} }
} }

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

@@ -38,7 +38,7 @@ class EthereumRpcConnector(
val wsHead = EthereumWsHead(conn, id, forkChoice, blockValidator) val wsHead = EthereumWsHead(conn, id, forkChoice, blockValidator)
// receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
val rpcHead = EthereumRpcHead(directReader, forkChoice, id, blockValidator, Duration.ofSeconds(60)) val rpcHead = EthereumRpcHead(directReader, forkChoice, id, blockValidator, Duration.ofSeconds(60))
head = MergedHead(listOf(rpcHead, wsHead), forkChoice) head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id")
} else { } else {
conn = null conn = null
log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC") log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC")

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
View File

@@ -123,7 +123,7 @@ open class EthereumPosMultiStream(
} }
} else { } else {
val heads = upstreams.map { it.getHead() } val heads = upstreams.map { it.getHead() }
val newHead = MergedHead(heads, PriorityForkChoice()).apply { val newHead = MergedHead(heads, PriorityForkChoice(), "ETH Pos Multistream").apply {
this.start() this.start()
} }
val lagObserver = EthereumPosHeadLagObserver(newHead, upstreams as Collection<Upstream>) val lagObserver = EthereumPosHeadLagObserver(newHead, upstreams as Collection<Upstream>)
@@ -161,12 +161,12 @@ open class EthereumPosMultiStream(
.apply { .apply {
log.debug("Found $size upstreams matching [${mather.describeInternal()}]") log.debug("Found $size upstreams matching [${mather.describeInternal()}]")
} }
.map { it.getHead() }
.let { .let {
val selected = it.map { it.getHead() }
when (it.size) { when (it.size) {
0 -> EmptyHead() 0 -> EmptyHead()
1 -> it.first() 1 -> selected.first()
else -> MergedHead(it, PriorityForkChoice()).apply { else -> MergedHead(selected, PriorityForkChoice(), "ETH head for ${it.map { it.getId() }}").apply {
start() start()
} }
} }