Fixes slot, directReader and wsHead bugs (#411)

src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt

@@ -105,7 +105,7 @@ class BlockContainer @JvmOverloads constructor(
 
     fun copyWithRating(nodeRating: Int): BlockContainer {
         return BlockContainer(
-            height, hash, difficulty, timestamp, full, json, parsed, parentHash, transactions, nodeRating,
+            height, hash, difficulty, timestamp, full, json, parsed, parentHash, transactions, nodeRating, upstreamId, slot,
         )
     }
 

src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt

@@ -42,7 +42,6 @@ class CompoundReader<K, D> (
                 rdr.read(key)
                     .timeout(Defaults.timeoutInternal, Mono.empty())
                     .doOnError { t -> log.warn("Failed to read from $rdr", t) }
-                    .onErrorResume { Mono.empty() }
             }, 1,)
             .next()
     }

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

@@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
 import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.domain.BlockHash
@@ -31,6 +32,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionLogJson
 import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.apache.commons.collections4.Factory
+import org.apache.commons.lang3.exception.ExceptionUtils
 import org.slf4j.LoggerFactory
 import org.springframework.cloud.sleuth.Tracer
 import reactor.core.publisher.Mono
@@ -81,7 +83,7 @@ class EthereumDirectReader(
             override fun read(key: TransactionId): Mono<Result<TxContainer>> {
                 val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
                 return readWithQuorum(request) // retries were removed because we use NotNullQuorum which handle errors too
-                    .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
+                    .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Tx not read $key")))
                     .flatMap { result ->
                         val tx = objectMapper.readValue(result.data, TransactionJsonSnapshot::class.java)
                         if (tx == null) {
@@ -96,6 +98,8 @@ class EthereumDirectReader(
                         if (tx.data.blockId != null) {
                             caches.cache(Caches.Tag.REQUESTED, tx.data)
                         }
+                    }.onErrorResume {
+                        Mono.error(JsonRpcException(request.id, ExceptionUtils.getRootCauseMessage(it)))
                     }
             }
         }
@@ -128,7 +132,7 @@ class EthereumDirectReader(
             override fun read(key: TransactionId): Mono<Result<ByteArray>> {
                 val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex()))
                 return readWithQuorum(request)
-                    .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Receipt not read $key")))
+                    .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Receipt not read $key")))
                     .flatMap { result ->
                         val receipt = objectMapper.readValue(result.data, TransactionReceiptJson::class.java)
                         if (receipt == null) {
@@ -149,6 +153,8 @@ class EthereumDirectReader(
                                 result,
                             )
                         }
+                    }.onErrorResume {
+                        Mono.error(JsonRpcException(request.id, ExceptionUtils.getRootCauseMessage(it)))
                     }
             }
         }
@@ -188,7 +194,7 @@ class EthereumDirectReader(
         matcher: Selector.Matcher = Selector.empty,
     ): Mono<Result<BlockContainer>> {
         return readWithQuorum(request, matcher)
-            .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id")))
+            .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Block not read $id")))
             .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200)))
             .flatMap { result ->
                 val block = objectMapper.readValue(result.data, BlockJson::class.java) as BlockJson<TransactionRefJson>?
@@ -208,6 +214,8 @@ class EthereumDirectReader(
             }
             .doOnNext { block ->
                 caches.cache(Caches.Tag.REQUESTED, block.data)
+            }.onErrorResume {
+                Mono.error(JsonRpcException(request.id, ExceptionUtils.getRootCauseMessage(it)))
             }
     }
 

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

@@ -50,15 +50,12 @@ class GenericWsHead(
     private var isSyncing = false
 
     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 var subscriptionId = AtomicReference("")
 
-    init {
-        registerHeadResubscribeFlux()
-    }
-
     override fun isRunning(): Boolean {
         return subscription != null
     }
@@ -73,6 +70,10 @@ class GenericWsHead(
             listenNewHeads(),
         )
         this.subscription = super.follow(heads)
+
+        if (headResubSubscription == null) {
+            headResubSubscription = registerHeadResubscribeFlux()
+        }
     }
 
     override fun onNoHeadUpdates() {
@@ -86,7 +87,7 @@ class GenericWsHead(
         this.isSyncing = isSyncing
     }
 
-    fun listenNewHeads(): Flux<BlockContainer> {
+    private fun listenNewHeads(): Flux<BlockContainer> {
         return subscribe()
             .map {
                 chainSpecific.parseHeader(it, "unknown")
@@ -102,7 +103,8 @@ class GenericWsHead(
     override fun stop() {
         super.stop()
         cancelSub()
-        noHeadUpdatesSink.tryEmitComplete()
+        headResubSubscription?.dispose()
+        headResubSubscription = null
     }
 
     override fun headLiveness(): Flux<Boolean> = headLivenessSink.asFlux()
@@ -135,7 +137,7 @@ class GenericWsHead(
         }
     }
 
-    private fun registerHeadResubscribeFlux() {
+    private fun registerHeadResubscribeFlux(): Disposable {
         val connectionStates = wsSubscriptions.connectionInfoFlux()
             .map {
                 if (it.connectionId == connectionId && it.connectionState == WsConnection.ConnectionState.DISCONNECTED) {
@@ -150,10 +152,10 @@ class GenericWsHead(
                 return@map false
             }
 
-        Flux.merge(
+        return Flux.merge(
             noHeadUpdatesSink.asFlux(),
             connectionStates,
-        ).subscribeOn(wsConnectionResubscribeScheduler)
+        ).publishOn(wsConnectionResubscribeScheduler)
             .filter { it && !subscribed && connected && !isSyncing }
             .subscribe {
                 log.warn("Restart ws head, upstreamId: $upstreamId")

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.test.GenericUpstreamMock
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.BlockValidator
@@ -59,24 +60,32 @@ class GenericWsHeadSpec extends Specification {
             Global.objectMapper.writeValueAsBytes(it)
         }
 
-        def apiMock = TestingCommons.api()
+        def reader = Mock(Reader) {
+            1 * it.read(new JsonRpcRequest("eth_getBlockByNumber", List.of("latest", false))) >> Mono.empty()
+        }
 
         def ws = Mock(WsSubscriptions) {
             1 * it.connectionInfoFlux() >> Flux.empty()
         }
 
-        def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
-
-        def res = BlockContainer.from(block)
-        when:
-        def act = head.listenNewHeads().blockFirst()
-
-        then:
-        act == res
-
         1 * ws.subscribe(_) >> new WsSubscriptions.SubscribeData(
                 Flux.fromIterable([headBlock]), "id", new AtomicReference<String>("")
         )
+
+        def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, reader, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
+
+        def res = BlockContainer.from(block)
+        when:
+        def act = head.getFlux()
+
+        then:
+        StepVerifier.create(act)
+                .then {
+                    head.start()
+                }
+                .expectNext(res)
+                .thenCancel()
+                .verify(Duration.ofSeconds(3))
     }
 
     def "Restart ethereum ws head"() {
@@ -328,7 +337,9 @@ class GenericWsHeadSpec extends Specification {
         block.uncles = []
         block.totalDifficulty = BigInteger.ONE
 
-        def apiMock = TestingCommons.api()
+        def reader = Mock(Reader) {
+            1 * it.read(new JsonRpcRequest("eth_getBlockByNumber", List.of("latest", false))) >> Mono.empty()
+        }
         def subId = "subId"
         def ws = Mock(WsSubscriptions) {
             1 * it.connectionInfoFlux() >> Flux.empty()
@@ -339,15 +350,19 @@ class GenericWsHeadSpec extends Specification {
                     Mono.just(new JsonRpcResponse("".bytes, null))
         }
 
-        def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
+        def head = new GenericWsHead(new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, reader, ws, Schedulers.boundedElastic(), Schedulers.boundedElastic(), upstream, EthereumChainSpecific.INSTANCE)
 
         when:
-        def act = head.listenNewHeads()
+        def act = head.getFlux()
 
         then:
         StepVerifier.create(act)
-            .expectComplete()
+            .then {
-            .verify(Duration.ofSeconds(1))
+                head.start()
+            }
+            .expectNoEvent(Duration.ofMillis(100))
+            .thenCancel()
+            .verify(Duration.ofSeconds(3))
     }
 
     def "If there is ws disconnect then head must emit false its liveness state"() {