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

more logs for heads and better understanding of what they belong to

Browse Source
This commit is contained in:
Termina1
2022-12-03 15:52:54 +02:00
parent d1021dcf63
commit afa3745478
5 changed files with 22 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
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
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()) {
try {
start()
@@ -82,10 +82,10 @@ abstract class AbstractHead @JvmOverloads constructor(
// but technically it should never happen during normal work, only when the Head
// is 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
} else {
log.warn("Received signal $it - stop emit new head!!!")
log.warn("Received signal $upstreamId $it - stop emit new head!!!")
completed = true
stream.tryEmitComplete()
}
@@ -95,7 +95,7 @@ abstract class AbstractHead @JvmOverloads constructor(
val valid = runCatching {
blockValidator.isValid(forkChoice.getHead(), block)
}.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 }
if (valid) {
notifyBeforeBlock()
@@ -113,7 +113,7 @@ abstract class AbstractHead @JvmOverloads constructor(
is ForkChoice.ChoiceResult.Same -> {}
}
} 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.publisher.Flux
class MergedHead(
class MergedHead @JvmOverloads constructor(
private val sources: Iterable<Head>,
forkChoice: ForkChoice
) : AbstractHead(forkChoice), Lifecycle, CachesEnabled {
forkChoice: ForkChoice,
private val label: String = ""
) : AbstractHead(forkChoice, upstreamId = label), Lifecycle, CachesEnabled {
companion object {
private val log = LoggerFactory.getLogger(MergedHead::class.java)
@@ -48,7 +49,9 @@ class MergedHead(
}
subscription?.dispose()
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 {
val heads = upstreams.map { it.getHead() }
val newHead = MergedHead(heads, MostWorkForkChoice()).apply {
val newHead = MergedHead(heads, MostWorkForkChoice(), "ETH Multistream").apply {
this.start()
}
val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream>)
@@ -166,13 +166,12 @@ open class EthereumMultistream(
upstreams.filter { mather.matches(it) }
.apply {
log.debug("Found $size upstreams matching [${mather.describeInternal()}]")
}
.map { it.getHead() }
.let {
}.let {
val selected = it.map { it.getHead() }
when (it.size) {
0 -> EmptyHead()
1 -> it.first()
else -> MergedHead(it, MostWorkForkChoice()).apply {
1 -> selected.first()
else -> MergedHead(selected, MostWorkForkChoice(), "Eth head ${it.map { it.getId() }}").apply {
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)
// 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))
head = MergedHead(listOf(rpcHead, wsHead), forkChoice)
head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id")
} else {
conn = null
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 {
val heads = upstreams.map { it.getHead() }
val newHead = MergedHead(heads, PriorityForkChoice()).apply {
val newHead = MergedHead(heads, PriorityForkChoice(), "ETH Pos Multistream").apply {
this.start()
}
val lagObserver = EthereumPosHeadLagObserver(newHead, upstreams as Collection<Upstream>)
@@ -161,12 +161,12 @@ open class EthereumPosMultiStream(
.apply {
log.debug("Found $size upstreams matching [${mather.describeInternal()}]")
}
.map { it.getHead() }
.let {
val selected = it.map { it.getHead() }
when (it.size) {
0 -> EmptyHead()
1 -> it.first()
else -> MergedHead(it, PriorityForkChoice()).apply {
1 -> selected.first()
else -> MergedHead(selected, PriorityForkChoice(), "ETH head for ${it.map { it.getId() }}").apply {
start()
}
}