diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 0f967fba..d9fb547d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -93,8 +93,9 @@ abstract class DefaultUpstream( open fun setStatus(avail: UpstreamAvailability) { status.updateAndGet { curr -> Status(curr.lag, avail, statusByLag(curr.lag, avail)) + }.also { + statusStream.tryEmitNext(it.status) } - statusStream.tryEmitNext(status.get().status) } fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability { @@ -118,13 +119,12 @@ abstract class DefaultUpstream( } override fun setLag(lag: Long) { - if (lag < 0) { - setLag(0) - } else { + lag.coerceAtLeast(0).let { nLag -> status.updateAndGet { curr -> - Status(lag, curr.avail, statusByLag(lag, curr.avail)) + Status(nLag, curr.avail, statusByLag(nLag, curr.avail)) + }.also { + statusStream.tryEmitNext(it.status) } - statusStream.tryEmitNext(status.get().status) } } 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 02fc4068..d0205dd9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt @@ -61,7 +61,7 @@ open class DefaultEthereumHead( BlockContainer.fromEthereumJson(it.getResult(), upstreamId) } .onErrorResume { err -> - log.error("Failed to fetch latest block: ${err.message}") + log.error("Failed to fetch latest block: ${err.message} $upstreamId", err) Mono.empty() } } 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 4f61b43d..41b9449a 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 @@ -11,6 +11,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead 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.AlwaysForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -35,9 +36,9 @@ 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) - val wsHead = EthereumWsHead(conn, id, forkChoice, blockValidator) + val wsHead = EthereumWsHead(conn, id, AlwaysForkChoice(), 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)) + val rpcHead = EthereumRpcHead(directReader, AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30)) head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id") } else { conn = null @@ -53,10 +54,10 @@ class EthereumRpcConnector( } override fun start() { + conn?.connect() if (head is Lifecycle) { head.start() } - conn?.connect() } override fun isRunning(): Boolean { 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 index 5070dcb0..c9fda3ed 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -136,8 +136,8 @@ open class EthereumPosMultiStream( } } } - }.apply { - onHeadUpdated(this) + }.also { + onHeadUpdated(it) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/AlwaysForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/AlwaysForkChoice.kt new file mode 100644 index 00000000..18d08303 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/AlwaysForkChoice.kt @@ -0,0 +1,22 @@ +package io.emeraldpay.dshackle.upstream.forkchoice + +import io.emeraldpay.dshackle.data.BlockContainer +import java.util.concurrent.atomic.AtomicReference + +class AlwaysForkChoice : ForkChoice { + private val head = AtomicReference(null) + + override fun getHead(): BlockContainer? = head.get() + + override fun filter(block: BlockContainer): Boolean = + head.get()?.let { it.hash != block.hash } ?: true + + override fun choose(block: BlockContainer): ForkChoice.ChoiceResult = + head.updateAndGet { block }.let { + if (it.hash == block.hash) { + ForkChoice.ChoiceResult.Updated(it) + } else { + ForkChoice.ChoiceResult.Same(it) + } + } +} \ No newline at end of file