From 74d666a0bdd965cbdb92c3f993844148d18ddd0b Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Tue, 21 Mar 2023 16:41:47 +0400 Subject: [PATCH] Fix reconnect to newHeads (#178) * Fix reconnect to newHeads --- .../dshackle/upstream/AbstractHead.kt | 13 ++---- .../emeraldpay/dshackle/upstream/EmptyHead.kt | 3 ++ .../io/emeraldpay/dshackle/upstream/Head.kt | 2 + .../upstream/ethereum/EthereumWsHead.kt | 10 ++++ .../dshackle/test/EthereumHeadMock.groovy | 5 ++ .../dshackle/upstream/AbstractHeadSpec.groovy | 6 +-- .../ethereum/EthereumWsHeadSpec.groovy | 46 +++++++++++++++++++ 7 files changed, 74 insertions(+), 11 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 0c6baa8b..59c2b28b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -67,11 +67,6 @@ abstract class AbstractHead @JvmOverloads constructor( } fun follow(source: Flux): Disposable { - if (completed) { - // if stream was already completed it cannot accept messages (with FAIL_TERMINATED), so needs to be recreated - stream = Sinks.many().multicast().directBestEffort() - completed = false - } return source .filter { log.trace("Filtering block $upstreamId block $it") @@ -85,9 +80,7 @@ abstract class AbstractHead @JvmOverloads constructor( log.warn("Received signal $upstreamId $it unexpectedly - restart head") lastHeadUpdated = 0L } else { - log.warn("Received signal $upstreamId $it - stop emit new head") - completed = true - stream.tryEmitComplete() + log.warn("Received signal $upstreamId $it, continue emit heads") } } .subscribeOn(Schedulers.boundedElastic()) @@ -155,6 +148,9 @@ abstract class AbstractHead @JvmOverloads constructor( future = null } + override fun onNoHeadUpdates() { + } + override fun start() { stopping = false log.debug("Start ${this.javaClass.simpleName} $upstreamId") @@ -165,6 +161,7 @@ abstract class AbstractHead @JvmOverloads constructor( delayed.set(delay > awaitHeadTimeoutMs) if (delayed.get()) { log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass.simpleName}") + onNoHeadUpdates() } }, 180, 30, TimeUnit.SECONDS ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt index 371fc95f..30dad649 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt @@ -36,4 +36,7 @@ class EmptyHead : Head { override fun stop() { } + + override fun onNoHeadUpdates() { + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt index 3337e44a..4a899a85 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt @@ -41,4 +41,6 @@ interface Head { fun start() fun stop() + + fun onNoHeadUpdates() } 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 8666ee90..877636d3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -28,6 +28,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson +import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -44,6 +45,10 @@ class EthereumWsHead( private val skipEnhance: Boolean ) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle { + companion object { + private val log = LoggerFactory.getLogger(EthereumWsHead::class.java) + } + private var subscription: Disposable? = null override fun isRunning(): Boolean { @@ -61,6 +66,11 @@ class EthereumWsHead( this.subscription = super.follow(heads) } + override fun onNoHeadUpdates() { + log.warn("Restart ws head, upstreamId: $upstreamId") + start() + } + fun listenNewHeads(): Flux { return wsSubscriptions.subscribe("newHeads") .map { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy index 142002b4..c41daba3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy @@ -77,4 +77,9 @@ class EthereumHeadMock implements Head { void stop() { } + + @Override + void onNoHeadUpdates() { + + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy index e95bca1c..7d0b766f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/AbstractHeadSpec.groovy @@ -60,7 +60,7 @@ class AbstractHeadSpec extends Specification { head.stop() source.tryEmitComplete() } - .expectComplete() + .thenCancel() .verify(Duration.ofSeconds(1)) } @@ -85,7 +85,7 @@ class AbstractHeadSpec extends Specification { head.stop() source.tryEmitComplete() } - .expectComplete() + .thenCancel() .verify(Duration.ofSeconds(1)) } @@ -115,7 +115,7 @@ class AbstractHeadSpec extends Specification { head.stop() source.tryEmitComplete() } - .expectComplete() + .thenCancel() .verify(Duration.ofSeconds(1)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy index 9bf4286f..db7d2451 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy @@ -25,8 +25,11 @@ import io.emeraldpay.etherjar.domain.TransactionId import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.test.StepVerifier import spock.lang.Specification +import java.time.Duration import java.time.Instant import java.time.temporal.ChronoUnit @@ -71,4 +74,47 @@ class EthereumWsHeadSpec extends Specification { headBlock ]) } + + def "Restart ethereum ws head"() { + setup: + def block = new BlockJson() + block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS) + block.number = 103 + block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") + def secondBlock = new BlockJson() + secondBlock.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS) + secondBlock.number = 105 + secondBlock.hash = BlockHash.from("0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8") + + def firstHeadBlock = block.with { + Global.objectMapper.writeValueAsBytes(it) + } + def secondHeadBlock = secondBlock.with { + Global.objectMapper.writeValueAsBytes(it) + } + + def apiMock = TestingCommons.api() + apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], null) + apiMock.answerOnce("eth_getBlockByHash", ["0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8", false], null) + apiMock.answerOnce("eth_blockNumber", [], Mono.empty()) + apiMock.answerOnce("eth_blockNumber", [], Mono.empty()) + + def ws = Mock(WsSubscriptions) { + 2 * subscribe("newHeads") >>> [Flux.fromIterable([firstHeadBlock]), Flux.fromIterable([secondHeadBlock])] + } + + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true) + + when: + def act = head.getFlux() + + then: + StepVerifier.create(act) + .then { head.start() } + .expectNext(BlockContainer.from(block)) + .then { head.onNoHeadUpdates() } + .expectNext(BlockContainer.from(secondBlock)) + .thenCancel() + .verify(Duration.ofSeconds(1)) + } }