From e40c62a728f28907ee1af06fbb5055ec50d822b2 Mon Sep 17 00:00:00 2001
From: Igor Artamonov <igor@artamonov.ru>
Date: Mon, 20 Sep 2021 22:43:00 -0400
Subject: [PATCH] solution: check current height on connect

---
 .../upstream/ethereum/DefaultEthereumHead.kt  | 39 +++++++++++++++++++
 .../upstream/ethereum/EthereumRpcHead.kt      | 25 +-----------
 .../upstream/ethereum/EthereumWsHead.kt       |  9 ++++-
 3 files changed, 48 insertions(+), 25 deletions(-)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
index ea912f82..3ccd4f8d 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
@@ -15,9 +15,48 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.AbstractHead
 import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.etherjar.hex.HexQuantity
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
 
 open class DefaultEthereumHead : Head, AbstractHead() {
 
+    companion object {
+        private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
+    }
+
+    fun getLatestBlock(api: Reader<JsonRpcRequest, JsonRpcResponse>): Mono<BlockContainer> {
+        return api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
+                .subscribeOn(EthereumRpcHead.scheduler)
+                .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
+                .flatMap {
+                    if (it.error != null) {
+                        Mono.error(it.error.asException(null))
+                    } else {
+                        val value = it.getResultAsProcessedString()
+                        Mono.just(HexQuantity.from(value))
+                    }
+                }
+                .flatMap {
+                    //fetching by Block Height here, critical to use the same upstream as in previous call,
+                    //b/c different upstreams may have different blocks on the same height
+                    api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
+                            .subscribeOn(EthereumRpcHead.scheduler)
+                            .timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
+                }
+                .map {
+                    BlockContainer.fromEthereumJson(it.getResult())
+                }
+                .onErrorResume { err ->
+                    log.debug("Failed to fetch latest block: ${err.message}")
+                    Mono.empty()
+                }
+    }
 }
\ No newline at end of file
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
index d4b1971d..960e2266 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
@@ -49,30 +49,7 @@ class EthereumRpcHead(
         val base = Flux.interval(interval)
                 .publishOn(scheduler)
                 .flatMap {
-                    api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
-                            .subscribeOn(scheduler)
-                            .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
-                            .flatMap {
-                                if (it.error != null) {
-                                    Mono.error(it.error.asException(null))
-                                } else {
-                                    val value = it.getResultAsProcessedString()
-                                    Mono.just(HexQuantity.from(value))
-                                }
-                            }
-                }
-                .flatMap {
-                    //fetching by Block Height here, critical to use same upstream,
-                    //different upstreams may have different blocks on the same height
-                    api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
-                            .subscribeOn(scheduler)
-                            .timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
-                }
-                .map {
-                    BlockContainer.fromEthereumJson(it.getResult())
-                }
-                .onErrorContinue { err, _ ->
-                    log.debug("RPC error ${err.message}")
+                    getLatestBlock(api)
                 }
         refreshSubscription = super.follow(base)
     }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
index 3386611e..a24c0e0a 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
@@ -16,9 +16,11 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
+import reactor.core.publisher.Flux
 
 class EthereumWsHead(
         private val ws: EthereumWsFactory.EthereumWs
@@ -34,7 +36,12 @@ class EthereumWsHead(
 
     override fun start() {
         this.subscription?.dispose()
-        this.subscription = super.follow(ws.getBlocksFlux())
+        val heads = Flux.merge(
+                // get the current block, not just wait for the next update
+                getLatestBlock(JsonRpcWsClient(ws)),
+                ws.getBlocksFlux()
+        )
+        this.subscription = super.follow(heads)
     }
 
     override fun stop() {