Check chain settings via ws (#544)

2024-08-12 18:27:47 +04:00
parent e8410d7638
commit 96487b6c5e
20 changed files with 449 additions and 59 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt
@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream

 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.micrometer.core.instrument.Gauge
 import io.micrometer.core.instrument.Meter
@@ -164,7 +165,7 @@ abstract class AbstractHead @JvmOverloads constructor(
        metrics.forEach { Metrics.globalRegistry.remove(it) }
    }

-    protected open fun onNoHeadUpdates() {
+    open fun onNoHeadUpdates() {
        // NOOP
    }

@@ -172,7 +173,7 @@ abstract class AbstractHead @JvmOverloads constructor(
        // NOOP
    }

-    override fun headLiveness(): Flux<Boolean> {
+    override fun headLiveness(): Flux<HeadLivenessState> {
        return Flux.empty()
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt
@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.upstream

 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import reactor.core.publisher.Flux

 class EmptyHead : Head {
@@ -44,7 +45,7 @@ class EmptyHead : Head {
    override fun onSyncingNode(isSyncing: Boolean) {
    }

-    override fun headLiveness(): Flux<Boolean> = Flux.empty()
+    override fun headLiveness(): Flux<HeadLivenessState> = Flux.empty()

    override fun getCurrent(): BlockContainer? {
        return null
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt
@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.upstream

 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import reactor.core.publisher.Flux

 /**
@@ -46,7 +47,7 @@ interface Head {

    fun onSyncingNode(isSyncing: Boolean)

-    fun headLiveness(): Flux<Boolean>
+    fun headLiveness(): Flux<HeadLivenessState>

    fun getCurrent(): BlockContainer?
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt
@@ -163,6 +163,17 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
        }
    }

+    override fun chainSettingsValidator(
+        chain: Chain,
+        upstream: Upstream,
+        reader: ChainReader,
+    ): SingleValidator<ValidateUpstreamSettingsResult>? {
+        if (upstream.getOptions().disableUpstreamValidation) {
+            return null
+        }
+        return ChainIdValidator(upstream, chain, reader)
+    }
+
    override fun upstreamRpcModulesDetector(upstream: Upstream): UpstreamRpcModulesDetector {
        return BasicEthUpstreamRpcModulesDetector(upstream)
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
 import io.emeraldpay.dshackle.foundation.ChainOptions
+import io.emeraldpay.dshackle.reader.ChainReader
 import io.emeraldpay.dshackle.upstream.ChainRequest
 import io.emeraldpay.dshackle.upstream.ChainResponse
 import io.emeraldpay.dshackle.upstream.SingleValidator
@@ -37,6 +38,8 @@ import reactor.kotlin.extra.retry.retryRandomBackoff
 import java.math.BigInteger
 import java.time.Duration
 import java.util.concurrent.TimeoutException
+import java.util.function.Supplier
+
 interface CallLimitValidator : SingleValidator<ValidateUpstreamSettingsResult> {
    fun isEnabled(): Boolean
 }
@@ -144,7 +147,11 @@ fun callLimitValidatorFactory(
 class ChainIdValidator(
    private val upstream: Upstream,
    private val chain: Chain,
+    private val customReader: ChainReader? = null,
 ) : SingleValidator<ValidateUpstreamSettingsResult> {
+    private val validatorReader: Supplier<ChainReader> = Supplier {
+        customReader ?: upstream.getIngressReader()
+    }

    companion object {
        @JvmStatic
@@ -186,7 +193,7 @@ class ChainIdValidator(
    }

    private fun chainId(): Mono<String> {
-        return upstream.getIngressReader()
+        return validatorReader.get()
            .read(ChainRequest("eth_chainId", ListParams()))
            .retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
                log.warn(
@@ -199,7 +206,7 @@ class ChainIdValidator(
    }

    private fun netVersion(): Mono<String> {
-        return upstream.getIngressReader()
+        return validatorReader.get()
            .read(ChainRequest("net_version", ListParams()))
            .retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
                log.warn(
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHead.kt
@@ -21,14 +21,18 @@ import io.emeraldpay.dshackle.reader.ChainReader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Lifecycle
+import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR
+import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult.UPSTREAM_VALID
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.generic.ChainSpecific
 import io.emeraldpay.dshackle.upstream.generic.GenericHead
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.Sinks
 import reactor.core.scheduler.Scheduler
+import reactor.kotlin.core.publisher.switchIfEmpty
 import java.time.Duration
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.atomic.AtomicReference
@@ -42,6 +46,7 @@ class GenericWsHead(
    headScheduler: Scheduler,
    upstream: DefaultUpstream,
    private val chainSpecific: ChainSpecific,
+    jsonRpcWsClient: JsonRpcWsClient,
    timeout: Duration,
 ) : GenericHead(upstream.getId(), forkChoice, blockValidator, headScheduler, chainSpecific), Lifecycle {
    private val wsHeadTimeout = run {
@@ -54,6 +59,7 @@ class GenericWsHead(
    }.also {
        log.info("WS head timeout for ${upstream.getId()} is $it")
    }
+    private val chainIdValidator = chainSpecific.chainSettingsValidator(upstream.getChain(), upstream, jsonRpcWsClient)

    private var connectionId: String? = null
    private var subscribed = false
@@ -63,7 +69,7 @@ class GenericWsHead(
    private var subscription: Disposable? = null
    private var headResubSubscription: Disposable? = null
    private val noHeadUpdatesSink = Sinks.many().multicast().directBestEffort<Boolean>()
-    private val headLivenessSink = Sinks.many().multicast().directBestEffort<Boolean>()
+    private val headLivenessSink = Sinks.many().multicast().directBestEffort<HeadLivenessState>()

    private var subscriptionId = AtomicReference("")

@@ -99,15 +105,37 @@ class GenericWsHead(
    }

    private fun listenNewHeads(): Flux<BlockContainer> {
-        return subscribe()
+        return Mono.justOrEmpty(chainIdValidator)
            .flatMap {
-                chainSpecific.getFromHeader(it, "unknown", api)
+                it!!.validate(UPSTREAM_SETTINGS_ERROR)
            }
-            .timeout(wsHeadTimeout, Mono.error(RuntimeException("No response from subscribe to newHeads")))
-            .onErrorResume {
-                log.error("Error getting heads for $upstreamId", it)
-                subscribed = false
-                unsubscribe()
+            .switchIfEmpty {
+                Mono.just(UPSTREAM_VALID)
+            }
+            .flatMapMany {
+                when (it) {
+                    UPSTREAM_VALID -> {
+                        subscribe()
+                            .flatMap { data ->
+                                chainSpecific.getFromHeader(data, "unknown", api)
+                            }
+                            .timeout(wsHeadTimeout, Mono.error(RuntimeException("No response from subscribe to newHeads")))
+                            .onErrorResume { err ->
+                                log.error("Error getting heads for {}, message {}", upstreamId, err.message)
+                                unsubscribe()
+                            }
+                    }
+                    UPSTREAM_SETTINGS_ERROR -> {
+                        log.warn("Couldn't check chain settings via ws connection for {}, ws sub will be recreated", upstreamId)
+                        subscribed = false
+                        Mono.empty()
+                    }
+                    else -> {
+                        log.error("Chain settings check hasn't been passed via ws connection, upstream {} will be removed", upstreamId)
+                        headLivenessSink.emitNext(HeadLivenessState.FATAL_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                        Mono.empty()
+                    }
+                }
            }
    }

@@ -118,9 +146,10 @@ class GenericWsHead(
        headResubSubscription = null
    }

-    override fun headLiveness(): Flux<Boolean> = headLivenessSink.asFlux()
+    override fun headLiveness(): Flux<HeadLivenessState> = headLivenessSink.asFlux()

    private fun unsubscribe(): Mono<BlockContainer> {
+        subscribed = false
        return wsSubscriptions.unsubscribe(chainSpecific.unsubscribeNewHeadsRequest(subscriptionId.get()).copy(id = ids.getAndIncrement()))
            .flatMap { it.requireResult() }
            .doOnNext { log.warn("{} has just unsubscribed from newHeads", upstreamId) }
@@ -152,7 +181,7 @@ class GenericWsHead(
        val connectionStates = wsSubscriptions.connectionInfoFlux()
            .map {
                if (it.connectionId == connectionId && it.connectionState == WsConnection.ConnectionState.DISCONNECTED) {
-                    headLivenessSink.emitNext(false) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                    headLivenessSink.emitNext(HeadLivenessState.DISCONNECTED) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
                    subscribed = false
                    connected = false
                    connectionId = null
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidator.kt
@@ -2,19 +2,23 @@ package io.emeraldpay.dshackle.upstream.ethereum

 import reactor.core.publisher.Flux

+enum class HeadLivenessState {
+    OK, NON_CONSECUTIVE, DISCONNECTED, FATAL_ERROR
+}
+
 interface HeadLivenessValidator {
-    fun getFlux(): Flux<Boolean>
+    fun getFlux(): Flux<HeadLivenessState>
 }

 class NoHeadLivenessValidator : HeadLivenessValidator {

-    override fun getFlux(): Flux<Boolean> {
-        return Flux.just(false)
+    override fun getFlux(): Flux<HeadLivenessState> {
+        return Flux.just(HeadLivenessState.NON_CONSECUTIVE)
    }
 }

 class AlwaysHeadLivenessValidator : HeadLivenessValidator {
-    override fun getFlux(): Flux<Boolean> {
-        return Flux.just(true)
+    override fun getFlux(): Flux<HeadLivenessState> {
+        return Flux.just(HeadLivenessState.OK)
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidatorImpl.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidatorImpl.kt
@@ -18,7 +18,7 @@ class HeadLivenessValidatorImpl(
        private val log = LoggerFactory.getLogger(HeadLivenessValidatorImpl::class.java)
    }

-    override fun getFlux(): Flux<Boolean> {
+    override fun getFlux(): Flux<HeadLivenessState> {
        val headLiveness = head.headLiveness()
        // first we have moving window of 2 blocks and check that they are consecutive ones
        val headFlux = head.getFlux().map { it.height }.buffer(2, 1).map {
@@ -39,13 +39,13 @@ class HeadLivenessValidatorImpl(
            // 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)
-                !value -> Flux.just(false)
+                count >= (CHECKED_BLOCKS_UNTIL_LIVE - 1) -> Flux.just(HeadLivenessState.OK)
+                !value -> Flux.just(HeadLivenessState.NON_CONSECUTIVE)
                else -> Flux.empty()
            }
        }.timeout(
            expectedBlockTime.multipliedBy(CHECKED_BLOCKS_UNTIL_LIVE.toLong() * 2),
-            Flux.just(false).doOnNext {
+            Flux.just(HeadLivenessState.NON_CONSECUTIVE).doOnNext {
                if (log.isDebugEnabled) {
                    log.debug("head liveness check broken with timeout in $upstreamId")
                } else {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt
@@ -60,6 +60,14 @@ abstract class AbstractChainSpecific : ChainSpecific {
        return null
    }

+    override fun chainSettingsValidator(
+        chain: Chain,
+        upstream: Upstream,
+        reader: ChainReader,
+    ): SingleValidator<ValidateUpstreamSettingsResult>? {
+        return null
+    }
+
    override fun upstreamRpcModulesDetector(upstream: Upstream): UpstreamRpcModulesDetector? {
        return null
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
@@ -23,10 +23,12 @@ import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.IngressSubscription
 import io.emeraldpay.dshackle.upstream.LogsOracle
 import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.SingleValidator
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamRpcModulesDetector
 import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector
 import io.emeraldpay.dshackle.upstream.UpstreamValidator
+import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
 import io.emeraldpay.dshackle.upstream.beaconchain.BeaconChainSpecific
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.CallSelector
@@ -82,6 +84,8 @@ interface ChainSpecific {

    fun upstreamSettingsDetector(chain: Chain, upstream: Upstream): UpstreamSettingsDetector?

+    fun chainSettingsValidator(chain: Chain, upstream: Upstream, reader: ChainReader): SingleValidator<ValidateUpstreamSettingsResult>?
+
    fun upstreamRpcModulesDetector(upstream: Upstream): UpstreamRpcModulesDetector?

    fun makeIngressSubscription(ws: WsSubscriptions): IngressSubscription
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt
@@ -23,7 +23,11 @@ import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetectorBuilder
 import io.emeraldpay.dshackle.upstream.UpstreamValidator
 import io.emeraldpay.dshackle.upstream.UpstreamValidatorBuilder
 import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
+import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR
+import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR
+import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult.UPSTREAM_VALID
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import io.emeraldpay.dshackle.upstream.finalization.FinalizationData
 import io.emeraldpay.dshackle.upstream.generic.connectors.ConnectorFactory
 import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnector
@@ -33,6 +37,7 @@ import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Sinks
 import java.time.Duration
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
@@ -95,6 +100,8 @@ open class GenericUpstream(
    private val finalizationDetector = finalizationDetectorBuilder()
    private var finalizationDetectorSubscription: Disposable? = null

+    private val headLivenessState = Sinks.many().multicast().directBestEffort<ValidateUpstreamSettingsResult>()
+
    override fun getHead(): Head {
        return connector.getHead()
    }
@@ -152,16 +159,16 @@ open class GenericUpstream(
        if (validator != null) {
            val validSettingsResult = validator.validateUpstreamSettingsOnStartup()
            when (validSettingsResult) {
-                ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR -> {
+                UPSTREAM_FATAL_SETTINGS_ERROR -> {
                    log.warn("Upstream ${getId()} couldn't start, invalid upstream settings")
                    connector.stop()
                    return
                }
-                ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR -> {
+                UPSTREAM_SETTINGS_ERROR -> {
                    log.warn("Non fatal upstream settings error, continue validation...")
                    connector.getHead().stop()
                }
-                ValidateUpstreamSettingsResult.UPSTREAM_VALID -> {
+                UPSTREAM_VALID -> {
                    isUpstreamValid.set(true)
                    upstreamStart()
                }
@@ -177,15 +184,18 @@ open class GenericUpstream(

    private fun validateUpstreamSettings() {
        if (validator != null) {
-            validationSettingsSubscription = Flux.interval(
-                Duration.ofSeconds(20),
-            ).flatMap {
-                validator.validateUpstreamSettings()
-            }
+            validationSettingsSubscription = Flux.merge(
+                Flux.interval(
+                    Duration.ofSeconds(20),
+                ).flatMap {
+                    validator.validateUpstreamSettings()
+                },
+                headLivenessState.asFlux(),
+            )
                .distinctUntilChanged()
                .subscribe {
                    when (it) {
-                        ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR -> {
+                        UPSTREAM_FATAL_SETTINGS_ERROR -> {
                            if (isUpstreamValid.get()) {
                                log.warn("There is a fatal error after upstream settings validation, removing ${getId()}...")
                                partialStop()
@@ -194,7 +204,7 @@ open class GenericUpstream(
                            isUpstreamValid.set(false)
                        }

-                        ValidateUpstreamSettingsResult.UPSTREAM_VALID -> {
+                        UPSTREAM_VALID -> {
                            if (!isUpstreamValid.get()) {
                                log.warn("Upstream ${getId()} is now valid, adding to the multistream...")
                                upstreamStart()
@@ -232,8 +242,6 @@ open class GenericUpstream(
    }

    private fun detectRpcModules(config: UpstreamsConfig.Upstream<*>, buildMethods: (UpstreamsConfig.Upstream<*>, Chain) -> CallMethods) {
-        rpcModulesDetector?.detectRpcModules()
-
        val rpcDetector = rpcModulesDetector?.detectRpcModules()?.block() ?: HashMap<String, String>()
        log.info("Upstream rpc detector for  ${getId()} returned  $rpcDetector ")
        if (rpcDetector.size != 0) {
@@ -266,9 +274,14 @@ open class GenericUpstream(
            validatorSubscription = validator?.start()
                ?.subscribe(this::setStatus)
        }
-        livenessSubscription = connector.hasLiveSubscriptionHead().subscribe({
-            hasLiveSubscriptionHead.set(it)
-            sendUpstreamStateEvent(UPDATED)
+        livenessSubscription = connector.headLivenessEvents().subscribe({
+            val hasSub = it == HeadLivenessState.OK
+            hasLiveSubscriptionHead.set(hasSub)
+            if (it == HeadLivenessState.FATAL_ERROR) {
+                headLivenessState.emitNext(UPSTREAM_FATAL_SETTINGS_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+            } else {
+                sendUpstreamStateEvent(UPDATED)
+            }
        }, {
            log.debug("Error while checking live subscription for ${getId()}", it)
        },)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnector.kt
@@ -4,12 +4,13 @@ import io.emeraldpay.dshackle.reader.ChainReader
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.IngressSubscription
 import io.emeraldpay.dshackle.upstream.Lifecycle
+import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import reactor.core.publisher.Flux

 interface GenericConnector : Lifecycle {
    fun getHead(): Head

-    fun hasLiveSubscriptionHead(): Flux<Boolean>
+    fun headLivenessEvents(): Flux<HeadLivenessState>

    fun getIngressReader(): ChainReader

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericRpcConnector.kt
@@ -14,6 +14,7 @@ import io.emeraldpay.dshackle.upstream.MergedHead
 import io.emeraldpay.dshackle.upstream.NoIngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.AlwaysHeadLivenessValidator
 import io.emeraldpay.dshackle.upstream.ethereum.GenericWsHead
+import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidator
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidatorImpl
 import io.emeraldpay.dshackle.upstream.ethereum.NoHeadLivenessValidator
@@ -30,6 +31,7 @@ import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactor
 import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD
 import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD
 import io.emeraldpay.dshackle.upstream.generic.connectors.GenericConnectorFactory.ConnectorMode.WS_ONLY
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.scheduler.Scheduler
@@ -55,18 +57,20 @@ class GenericRpcConnector(
    private val ingressSubscription: IngressSubscription?
    private val head: Head
    private val liveness: HeadLivenessValidator
+    private val jsonRpcWsClient: JsonRpcWsClient?

    companion object {
        private val log = LoggerFactory.getLogger(GenericRpcConnector::class.java)
    }

-    override fun hasLiveSubscriptionHead(): Flux<Boolean> {
+    override fun headLivenessEvents(): Flux<HeadLivenessState> {
        return liveness.getFlux().distinctUntilChanged()
    }

    init {
        pool = wsFactory?.create(upstream)
        wsSubs = pool?.let { WsSubscriptionsImpl(it) }
+        jsonRpcWsClient = pool?.let { JsonRpcWsClient(pool) }
        ingressSubscription = wsSubs?.let { chainSpecific.makeIngressSubscription(it) }

        head = when (connectorType) {
@@ -98,6 +102,7 @@ class GenericRpcConnector(
                        headScheduler,
                        upstream,
                        chainSpecific,
+                        jsonRpcWsClient!!,
                        expectedBlockTime,
                    )
                // receive all new blocks through WebSockets, but also periodically verify with RPC in case if WS failed
@@ -124,6 +129,7 @@ class GenericRpcConnector(
                    headScheduler,
                    upstream,
                    chainSpecific,
+                    jsonRpcWsClient!!,
                    expectedBlockTime,
                )
            }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericWsConnector.kt
@@ -6,6 +6,7 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.IngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.GenericWsHead
+import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessState
 import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidatorImpl
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPoolFactory
@@ -46,13 +47,14 @@ class GenericWsConnector(
            headScheduler,
            upstream,
            chainSpecific,
+            reader,
            expectedBlockTime,
        )
        liveness = HeadLivenessValidatorImpl(head, expectedBlockTime, headLivenessScheduler, upstream.getId())
        subscriptions = chainSpecific.makeIngressSubscription(wsSubscriptions)
    }

-    override fun hasLiveSubscriptionHead(): Flux<Boolean> {
+    override fun headLivenessEvents(): Flux<HeadLivenessState> {
        return liveness.getFlux().distinctUntilChanged()
    }
    override fun start() {