fix head liveness validator memmory leak (#289)

gradle/libs.versions.toml

@@ -96,7 +96,7 @@ zeromq = "org.zeromq:jeromq:0.5.2"
 objgenesis = "org.objenesis:objenesis:3.1"
 
 reactor-core = { module = "io.projectreactor:reactor-core", version.ref = "reactor" }
-reactor-netty = { module = "io.projectreactor.netty:reactor-netty", version = "1.0.24" }
+reactor-netty = { module = "io.projectreactor.netty:reactor-netty", version = "1.0.11" }
 reactor-extra = { module = "io.projectreactor.addons:reactor-extra", version = "3.4.5" }
 reactor-kotlin = { module = "io.projectreactor.kotlin:reactor-kotlin-extensions", version = "1.1.4" }
 reactor-test = { module = "io.projectreactor:reactor-test", version.ref = "reactor" }

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

@@ -56,6 +56,7 @@ open class EthereumLikeRpcUpstream(
     private var hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false)
 
     private var validatorSubscription: Disposable? = null
+    private var livenessSubscription: Disposable? = null
 
     override fun getCapabilities(): Set<Capability> {
         return if (hasLiveSubscriptionHead.get()) {
@@ -88,7 +89,7 @@ open class EthereumLikeRpcUpstream(
             validatorSubscription = validator.start()
                 .subscribe(this::setStatus)
         }
-        connector.hasLiveSubscriptionHead().subscribe {
+        livenessSubscription = connector.hasLiveSubscriptionHead().subscribe {
             hasLiveSubscriptionHead.set(it)
             eventPublisher?.publishEvent(UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED))
         }
@@ -122,6 +123,8 @@ open class EthereumLikeRpcUpstream(
     override fun stop() {
         validatorSubscription?.dispose()
         validatorSubscription = null
+        livenessSubscription?.dispose()
+        livenessSubscription = null
         connector.stop()
     }
 

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

@@ -7,9 +7,9 @@ import reactor.core.scheduler.Scheduler
 import java.time.Duration
 
 class HeadLivenessValidator(
-    val head: Head,
+    private val head: Head,
-    val expectedBlockTime: Duration,
+    private val expectedBlockTime: Duration,
-    val scheduler: Scheduler,
+    private val scheduler: Scheduler,
     private val upstreamId: String
 ) {
     companion object {
@@ -17,17 +17,10 @@ class HeadLivenessValidator(
         private val log = LoggerFactory.getLogger(HeadLivenessValidator::class.java)
     }
 
-    private fun fallback(): Flux<Boolean> {
-        return Flux.defer {
-            log.debug("head liveness check broken with timeout in $upstreamId")
-            Flux.just(false).concatWith(getFlux()) // emit false and then restart the Flux
-        }
-    }
-
     fun getFlux(): Flux<Boolean> {
         // first we have moving window of 2 blocks and check that they are consecutive ones
-        return head.getFlux().buffer(2, 1).map {
+        return head.getFlux().map { it.height }.buffer(2, 1).map {
-            it.last().height - it.first().height == 1L
+            it.last() - it.first() == 1L
         }.scan(Pair(0, true)) { acc, value ->
             // then we accumulate consecutive true events, false resets counter
             if (value) {
@@ -44,10 +37,11 @@ class HeadLivenessValidator(
                 !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),
-            fallback()
+            Flux.just(false).doOnNext {
-        ).subscribeOn(scheduler)
+                log.debug("head liveness check broken with timeout in $upstreamId")
+            }
+        ).repeat().subscribeOn(scheduler)
     }
 }