Merge pull request #90 from p2p-org/fix_stuck_provider

fix handling ws + prior rpc connections

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

@@ -93,8 +93,9 @@ abstract class DefaultUpstream(
     open fun setStatus(avail: UpstreamAvailability) {
         status.updateAndGet { curr ->
             Status(curr.lag, avail, statusByLag(curr.lag, avail))
+        }.also {
+            statusStream.tryEmitNext(it.status)
         }
-        statusStream.tryEmitNext(status.get().status)
     }
 
     fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
@@ -118,13 +119,12 @@ abstract class DefaultUpstream(
     }
 
     override fun setLag(lag: Long) {
-        if (lag < 0) {
-            setLag(0)
-        } else {
+        lag.coerceAtLeast(0).let { nLag ->
             status.updateAndGet { curr ->
-                Status(lag, curr.avail, statusByLag(lag, curr.avail))
+                Status(nLag, curr.avail, statusByLag(nLag, curr.avail))
+            }.also {
+                statusStream.tryEmitNext(it.status)
             }
-            statusStream.tryEmitNext(status.get().status)
         }
     }
 

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

@@ -61,7 +61,7 @@ open class DefaultEthereumHead(
                 BlockContainer.fromEthereumJson(it.getResult(), upstreamId)
             }
             .onErrorResume { err ->
-                log.error("Failed to fetch latest block: ${err.message}")
+                log.error("Failed to fetch latest block: ${err.message} $upstreamId", err)
                 Mono.empty()
             }
     }

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

@@ -11,6 +11,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
+import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -35,9 +36,9 @@ class EthereumRpcConnector(
         if (wsFactory != null) {
             // do not set upstream to the WS, since it doesn't control the RPC upstream
             conn = wsFactory.create(null, null)
-            val wsHead = EthereumWsHead(conn, id, forkChoice, blockValidator)
+            val wsHead = EthereumWsHead(conn, id, AlwaysForkChoice(), blockValidator)
             // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
-            val rpcHead = EthereumRpcHead(directReader, forkChoice, id, blockValidator, Duration.ofSeconds(60))
+            val rpcHead = EthereumRpcHead(directReader, AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
             head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id")
         } else {
             conn = null
@@ -53,10 +54,10 @@ class EthereumRpcConnector(
     }
 
     override fun start() {
+        conn?.connect()
         if (head is Lifecycle) {
             head.start()
         }
-        conn?.connect()
     }
 
     override fun isRunning(): Boolean {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -136,8 +136,8 @@ open class EthereumPosMultiStream(
                     }
                 }
             }
-        }.apply {
-            onHeadUpdated(this)
+        }.also {
+            onHeadUpdated(it)
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/AlwaysForkChoice.kt

@@ -0,0 +1,22 @@
+package io.emeraldpay.dshackle.upstream.forkchoice
+
+import io.emeraldpay.dshackle.data.BlockContainer
+import java.util.concurrent.atomic.AtomicReference
+
+class AlwaysForkChoice : ForkChoice {
+    private val head = AtomicReference<BlockContainer>(null)
+
+    override fun getHead(): BlockContainer? = head.get()
+
+    override fun filter(block: BlockContainer): Boolean =
+        head.get()?.let { it.hash != block.hash } ?: true
+
+    override fun choose(block: BlockContainer): ForkChoice.ChoiceResult =
+        head.updateAndGet { block }.let {
+            if (it.hash == block.hash) {
+                ForkChoice.ChoiceResult.Updated(it)
+            } else {
+                ForkChoice.ChoiceResult.Same(it)
+            }
+        }
+}