Fixes head liveness check: timeout logic and periodic false negative events (#280)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidator.kt

@@ -1,6 +1,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.upstream.Head
+import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.scheduler.Scheduler
 import java.time.Duration
@@ -8,11 +9,19 @@ import java.time.Duration
 class HeadLivenessValidator(
     val head: Head,
     val expectedBlockTime: Duration,
-    val scheduler: Scheduler
+    val scheduler: Scheduler,
+    private val upstreamId: String
 ) {
-
     companion object {
         const val CHECKED_BLOCKS_UNTIL_LIVE = 3
+        private val log = LoggerFactory.getLogger(HeadLivenessValidator::class.java)
+    }
+
+    private fun fallback(): Flux<Boolean> {
+        return Flux.defer {
+            log.info("head liveness check broken with timeout in $upstreamId")
+            Flux.just(false).concatWith(getFlux()) // emit false and then restart the Flux
+        }
     }
 
     fun getFlux(): Flux<Boolean> {
@@ -24,18 +33,21 @@ class HeadLivenessValidator(
             if (value) {
                 Pair(acc.first + 1, true)
             } else {
+                log.info("non consecutive blocks in head for $upstreamId")
                 Pair(0, false)
             }
         }.flatMap { (count, value) ->
             // we emit when we have false or checked CHECKED_BLOCKS_UNTIL_LIVE blocks
             // CHECKED_BLOCKS_UNTIL_LIVE blocks == (CHECKED_BLOCKS_UNTIL_LIVE - 1) consecutive true
             when {
-                count == (CHECKED_BLOCKS_UNTIL_LIVE - 1) -> Flux.just(true)
+                count >= (CHECKED_BLOCKS_UNTIL_LIVE - 1) -> Flux.just(true)
                 !value -> Flux.just(false)
                 else -> Flux.empty()
             }
             // finally, we timeout after we waited for double the time we needed to emit those blocks
-        }.timeout(expectedBlockTime.multipliedBy(CHECKED_BLOCKS_UNTIL_LIVE.toLong() * 2), Flux.just(false))
-            .distinctUntilChanged().subscribeOn(scheduler)
+        }.timeout(
+            expectedBlockTime.multipliedBy(CHECKED_BLOCKS_UNTIL_LIVE.toLong() * 2),
+            fallback()
+        ).subscribeOn(scheduler)
     }
 }

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

@@ -99,7 +99,7 @@ class EthereumRpcConnector(
                 )
             }
         }
-        liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler)
+        liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler, id)
     }
 
     override fun setCaches(caches: Caches) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt

@@ -46,7 +46,7 @@ class EthereumWsConnector(
             wsConnectionResubscribeScheduler,
             headScheduler
         )
-        liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler)
+        liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler, upstream.getId())
         subscriptions = EthereumWsIngressSubscription(wsSubscriptions)
     }