Fix reconnect to newHeads (#178)

* Fix reconnect to newHeads

src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt

@@ -67,11 +67,6 @@ abstract class AbstractHead @JvmOverloads constructor(
     }
 
     fun follow(source: Flux<BlockContainer>): 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<BlockContainer>()
-            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
             )

src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt

@@ -36,4 +36,7 @@ class EmptyHead : Head {
 
     override fun stop() {
     }
+
+    override fun onNoHeadUpdates() {
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt

@@ -41,4 +41,6 @@ interface Head {
     fun start()
 
     fun stop()
+
+    fun onNoHeadUpdates()
 }

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<BlockContainer> {
         return wsSubscriptions.subscribe("newHeads")
             .map {

src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy

@@ -77,4 +77,9 @@ class EthereumHeadMock implements Head {
     void stop() {
 
     }
+
+    @Override
+    void onNoHeadUpdates() {
+
+    }
 }

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

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<TransactionRefJson>()
+        block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
+        block.number = 103
+        block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
+        def secondBlock = new BlockJson<TransactionRefJson>()
+        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))
+    }
 }