solution: configurable timeout

README.adoc

@@ -45,7 +45,7 @@ upstreams:
 
 Which sets the following:
 
-- application listen on 0.0.0.0:2448
+- application listen on 0.0.0.0:2449
 - TLS security is disabled (_don't use in production!_)
 - read upstreams configuration from file `upstreams.yaml` in the current directory
 

docs/04-upstream-config.adoc

@@ -86,9 +86,15 @@ In the example above we have:
 
 Options (default or as part of upstream config):
 
-- `disable-validation` - if `true` then Dshackle would not try to verify status of the upstream (useful for a trusted cloud
+[cols="2,1,5a"]
-   provider such as Infura, but is not recommended for an own node)
+|===
-- `min-peers` - do not use upstream with less than specified connected peers
+| Option | Default | Description
+
+| `disable-validation` | false | if `true` then Dshackle will not try to verify status of the upstream (could be useful for a trusted cloud
+   provider such as Infura, but disabling it is not recommended for a normal node)
+| `min-peers` | 3 | specify minimum amount of connected peers, Dshackle will not use upstream with less than specified number
+| `timeout` | 60 | timeout in seconds after which request to the upstream will be discarded (and may be retried on an another upstream)
+|===
 
 === Connection type
 
@@ -153,12 +159,13 @@ upstreams:
         - name: eth_getBlockByNumber
 ----
 
-Such configuration option allows to execute methods `trace_transaction` and also disables `eth_getBlockByNumber` on that
+Such configuration option allows to execute method `trace_transaction` and also disables `eth_getBlockByNumber` on that
-particular upstream. If a client tries to execute method `trace_transaction` it will be executed on that upstream, or
+particular upstream. If a client requests to execute method `trace_transaction` then it will be scheduled to that upstream (or
-another upstream will have such method enabled.
+any upstream with such method enabled).
 
-Together with label `archive: true` it's possible to specify during execution that a client wants to execute method only
+NOTE: It's especially useful when used together with upstream labels. If an archive upstream has label `archive: true` it's
-on an archive node.
+possible to specify that the client wants to execute method `trace_transaction` only on an archive node(s), which has
+complete historical data for tracing.
 
 === Authentication
 
@@ -167,7 +174,9 @@ on an archive node.
 All connection types can use TLS secured connection, with optional client certificate authentication:
 
 - `ca` path to certificate required from remote server
-- optional `certificate` and `key` for client authentication. Please note that `key` is encoded with _PKCS 8_
+- optional `certificate` and `key` for client authentication.
+
+NOTE: Please note that `key` must be encoded with _PKCS 8_
 
 ==== Basic Authentication
 

src/main/kotlin/io/emeraldpay/dshackle/Defaults.kt

@@ -13,21 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.emeraldpay.dshackle.upstream
+package io.emeraldpay.dshackle
 
-import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
 import java.time.Duration
 
-class UpstreamServices {
+class Defaults {
 
     companion object {
-        fun onceOk(up: Upstream, waitTime: Duration = Duration.ofSeconds(15)): Mono<Boolean> {
+        val timeout = Duration.ofSeconds(60)
-            return Flux.concat(Flux.just(up.getStatus()), up.observeStatus())
-                    .timeout(waitTime, Mono.just(UpstreamAvailability.UNAVAILABLE))
-                    .filter { it == UpstreamAvailability.OK }
-                    .next()
-                    .hasElement()
-        }
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -15,7 +15,9 @@
  */
 package io.emeraldpay.dshackle.config
 
+import io.emeraldpay.dshackle.Defaults
 import java.net.URI
+import java.time.Duration
 import java.util.*
 import kotlin.collections.ArrayList
 import kotlin.collections.HashMap
@@ -27,6 +29,7 @@ class UpstreamsConfig {
 
     open class Options {
         var disableValidation: Boolean? = null
+        var timeout = Defaults.timeout
 
         var minPeers: Int? = 1
             set(minPeers) {

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -27,6 +27,7 @@ import java.io.InputStream
 import java.io.InputStreamReader
 import java.lang.IllegalArgumentException
 import java.net.URI
+import java.time.Duration
 
 class UpstreamsConfigReader {
 
@@ -185,6 +186,9 @@ class UpstreamsConfigReader {
         getValueAsInt(values, "min-peers")?.let {
             options.minPeers = it
         }
+        getValueAsInt(values, "timeout")?.let {
+            options.timeout = Duration.ofSeconds(it.toLong())
+        }
         getValueAsBool(values, "disable-validation")?.let {
             options.disableValidation = it
         }

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.rpc
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
@@ -169,7 +170,7 @@ class TrackAddress(
         val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(Exception("Unsupported chain: ${addr.chain}"))
         return up.getApi(Selector.empty)
                 .flatMap { api -> api.executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) }
-                .timeout(Duration.ofSeconds(15))
+                .timeout(Defaults.timeout)
     }
 
     private fun updateBalances(chain: Chain, group: List<TrackedAddress>): Flux<TrackedAddress> {

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

@@ -63,7 +63,8 @@ open class ConfiguredUpstreams(
         config.upstreams.forEach { up ->
 
             if (up.connection is UpstreamsConfig.GrpcConnection) {
-                buildGrpcUpstream(up as UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>)
+                val options = up.options ?: UpstreamsConfig.Options()
+                buildGrpcUpstream(up as UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>, options)
             } else {
                 val chain = chainNames[up.chain]
                 if (chain == null) {
@@ -145,7 +146,10 @@ open class ConfiguredUpstreams(
                     rpcClient,
                     objectMapper,
                     methods
-            )
+            ).apply {
+                timeout = options.timeout
+            }
+
             urls.add(endpoint.url)
         }
         if (rpcApi != null) {
@@ -174,7 +178,7 @@ open class ConfiguredUpstreams(
         }
     }
 
-    private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>) {
+    private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>, options: UpstreamsConfig.Options) {
         val endpoint = config.connection!!
         val ds = GrpcUpstreams(
                 config.id!!,
@@ -183,7 +187,9 @@ open class ConfiguredUpstreams(
                 objectMapper,
                 endpoint.auth,
                 fileResolver
-        )
+        ).apply {
+            timeout = options.timeout
+        }
         log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
         ds.start()
                 .doOnNext {

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

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.upstream
 
+import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
 import io.infinitape.etherjar.rpc.Batch
@@ -36,7 +37,7 @@ class UpstreamValidator(
         return ethereumUpstream.getApi(Selector.empty)
                 .map { api -> api.rpcClient.execute(batch) }
                 .flatMap { Mono.fromCompletionStage(it) }
-                .timeout(Duration.ofSeconds(10))
+                .timeout(Defaults.timeout)
                 .map {
                     if (syncing.get().isSyncing) {
                         UpstreamAvailability.SYNCING

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

@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.upstream.CallMethods
 import io.infinitape.etherjar.rpc.RpcCall
 import io.infinitape.etherjar.rpc.RpcClient
@@ -31,7 +32,7 @@ open class DirectEthereumApi(
         val targets: CallMethods
 ): EthereumApi(objectMapper) {
 
-    private val timeout = Duration.ofSeconds(5)
+    var timeout = Defaults.timeout
     private val log = LoggerFactory.getLogger(EthereumApi::class.java)
 
     override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {

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

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.Defaults
 import io.infinitape.etherjar.rpc.Batch
 import io.infinitape.etherjar.rpc.Commands
 import org.slf4j.LoggerFactory
@@ -39,13 +40,13 @@ class EthereumRpcHead(
                     val batch = Batch()
                     val f = batch.add(Commands.eth().blockNumber)
                     api.rpcClient.execute(batch)
-                    Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty())
+                    Mono.fromCompletionStage(f).timeout(Defaults.timeout, Mono.empty())
                 }
                 .flatMap {
                     val batch = Batch()
                     val f = batch.add(Commands.eth().getBlock(it))
                     api.rpcClient.execute(batch)
-                    Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty())
+                    Mono.fromCompletionStage(f).timeout(Defaults.timeout, Mono.empty())
                 }
                 .onErrorContinue { err, _ ->
                     log.debug("RPC error ${err.message}")

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

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.Commands
@@ -62,7 +63,7 @@ class EthereumWs(
                             .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250))
                             .apply(n)
                 }
-                .timeout(Duration.ofSeconds(5), Mono.empty())
+                .timeout(Defaults.timeout, Mono.empty())
                 .subscribe(topic::onNext)
             } else {
                 topic.onNext(it)
@@ -73,6 +74,5 @@ class EthereumWs(
     fun getFlux(): Flux<BlockJson<TransactionId>> {
         return Flux.from(this.topic)
                 .onBackpressureLatest()
-                .sample(Duration.ofMillis(100))
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt

@@ -20,6 +20,7 @@ import com.salesforce.reactorgrpc.GrpcRetry
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead
@@ -61,9 +62,10 @@ open class GrpcUpstream(
     private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
     private val head = DefaultEthereumHead()
     private var targets: CallMethods? = null
-
     private var headSubscription: Disposable? = null
 
+    var timeout = Defaults.timeout
+
     open fun createApi(matcher: Selector.Matcher): DirectEthereumApi {
         val targets = this.getMethods()
         val transport = Selector.extractLabels(matcher)?.let { selector ->
@@ -123,7 +125,7 @@ open class GrpcUpstream(
         }.flatMap {
             getApi(Selector.EmptyMatcher())
                     .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(it.hash)) }
-                    .timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Timeout from upstream")))
+                    .timeout(timeout, Mono.error(TimeoutException("Timeout from upstream")))
                     .doOnError { t ->
                         setStatus(UpstreamAvailability.UNAVAILABLE)
                         val msg = "Failed to download block data for chain $chain on $parentId"

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.grpc
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
+import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
@@ -50,10 +51,11 @@ class GrpcUpstreams(
 ) {
     private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
 
+    var timeout = Defaults.timeout
+
     private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null
     private val known = HashMap<Chain, GrpcUpstream>()
     private val lock = ReentrantLock()
-
     private var grpcTransport: EmeraldGrpcTransport? = null
 
     fun start(): Flux<UpstreamChange> {
@@ -162,6 +164,7 @@ class GrpcUpstreams(
             val current = known[chain]
             return if (current == null) {
                 val created = GrpcUpstream(id, chain, client!!, objectMapper, grpcTransport!!.copyForChain(chain))
+                created.timeout = this.timeout
                 known[chain] = created
                 created.start()
                 UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED)