solution: upstream apis as a controller Publisher

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

@@ -32,7 +32,8 @@ class BlockApiReader(
     override fun read(key: BlockHash): Mono<BlockJson<TransactionId>> {
         return Mono.just(key)
                 .flatMap {
-                    upstream.getApi(Selector.empty).executeAndConvert(Commands.eth().getBlock(it))
+                    upstream.getApi(Selector.empty)
+                            .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(it)) }
                 }.repeatWhenEmpty { n ->
                     Repeat.times<Any>(3)
                             .exponentialBackoff(Duration.ofMillis(100), Duration.ofMillis(500))

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

@@ -19,7 +19,6 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.upstream.*
-import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
 import io.emeraldpay.grpc.Chain
@@ -30,9 +29,6 @@ import org.springframework.stereotype.Service
 import reactor.core.publisher.*
 import reactor.util.function.Tuples
 import java.lang.Exception
-import java.time.Duration
-import java.util.concurrent.atomic.AtomicInteger
-import java.util.function.Predicate
 
 @Service
 class NativeCall(
@@ -122,24 +118,27 @@ class NativeCall(
     }
 
     fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
-        val all = ctx.getApis().toFlux().share()
-        //execute on the first API immediately, and then make a delay between each call to not overload upstreams
-        val immediate = Flux.from(all).take(1)
-        val repeatControl = EmitterProcessor.create<Boolean>()
-        val retries = Flux.from(all).skip(1)
-                .zipWith(repeatControl.delayElements(Duration.ofMillis(200))) //manages when need another call, make delay for at least of 200ms between calls
-                .map { it.t1 }
-
-        return Flux.concat(immediate, retries)
+        val apis = ctx.getApis()
+        apis.request(1)
+        var failures = 0
+        return Flux.from(apis)
                 .flatMap { api ->
                     api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) }
                 }
-                .retry(3)
+                .retry {
+                    failures++
+                    if (failures <= 3) {
+                        apis.request(1)
+                        true
+                    } else {
+                        false
+                    }
+                }
                 .reduce(ctx.callQuorum, {res, a ->
                     if (res.record(a.t1, a.t2)) {
-                        repeatControl.onComplete()
+                        apis.resolve()
                     } else {
-                        repeatControl.onNext(true)
+                        apis.request(1)
                     }
                     res
                 })
@@ -181,7 +180,7 @@ class NativeCall(
             return CallContext(id, upstream, matcher, callQuorum, payload)
         }
 
-        fun getApis(): Iterator<DirectEthereumApi> {
+        fun getApis(): ApiSource {
             return upstream.getApis(matcher)
         }
     }

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

@@ -168,7 +168,7 @@ class TrackAddress(
     fun getBalance(addr: SimpleAddress): Mono<Wei> {
         val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(Exception("Unsupported chain: ${addr.chain}"))
         return up.getApi(Selector.empty)
-                .executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST))
+                .flatMap { api -> api.executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) }
                 .timeout(Duration.ofSeconds(15))
     }
 

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

@@ -208,7 +208,7 @@ class TrackTx(
         val upstream = upstreams.getUpstream(tx.chain)
                 ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}"))
         return upstream.getApi(Selector.empty)
-                .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash))
+                .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) }
                 .map { block ->
                     setBlockDetails(tx, block)
                 }.doOnError { t ->
@@ -249,7 +249,7 @@ class TrackTx(
         val initialStatus = tx.status
         val upstream = upstreams.getUpstream(tx.chain) ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}"))
         val execution = upstream.getApi(Selector.empty)
-                .executeAndConvert(Commands.eth().getTransaction(tx.txid))
+                .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) }
         return execution
                 .flatMap { updateFromBlock(upstream, tx, it) }
                 .doOnError { t ->

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

@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
+import org.reactivestreams.Publisher
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
@@ -51,7 +52,7 @@ abstract class AggregatedUpstream(
 
     abstract fun getAll(): List<Upstream>
     abstract fun addUpstream(upstream: Upstream)
-    abstract fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi>
+    abstract fun getApis(matcher: Selector.Matcher): ApiSource
 
     fun onUpstreamsUpdated() {
         reconfigLock.withLock {

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

@@ -0,0 +1,26 @@
+/**
+ * Copyright (c) 2019 ETCDEV GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
+import org.reactivestreams.Publisher
+
+interface ApiSource: Publisher<DirectEthereumApi> {
+
+    fun resolve()
+    fun request(tries: Int)
+
+}

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

@@ -24,6 +24,7 @@ import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
+import reactor.core.publisher.Mono
 import java.lang.IllegalStateException
 import java.time.Duration
 
@@ -116,16 +117,19 @@ open class ChainUpstreams (
         }
     }
 
-    override fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi> {
+    override fun getApis(matcher: Selector.Matcher): ApiSource {
         val i = seq++
         if (seq >= Int.MAX_VALUE / 2) {
             seq = 0
         }
-        return FilteringApiIterator(upstreams, i, matcher)
+        return FilteredApis(upstreams, matcher, i)
     }
 
-    override fun getApi(matcher: Selector.Matcher): DirectEthereumApi {
-        return getApis(matcher).next()
+    override fun getApi(matcher: Selector.Matcher): Mono<DirectEthereumApi> {
+        val apis = getApis(matcher)
+        apis.request(1)
+        return Mono.from(apis)
+                .switchIfEmpty(Mono.error<DirectEthereumApi>(Exception("No API available")))
     }
 
     override fun getHead(): EthereumHead {

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

@@ -0,0 +1,102 @@
+/**
+ * Copyright (c) 2019 ETCDEV GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
+import org.reactivestreams.Subscriber
+import reactor.core.publisher.EmitterProcessor
+import reactor.core.publisher.Flux
+import java.time.Duration
+import kotlin.math.max
+import kotlin.math.min
+import kotlin.math.pow
+import kotlin.math.roundToLong
+import kotlin.random.Random
+
+class FilteredApis(
+        allUpstreams: List<Upstream>,
+        private val matcher: Selector.Matcher,
+        pos: Int,
+        private val repeatLimit: Long,
+        jitter: Int
+): ApiSource {
+
+    companion object {
+        private const val DEFAULT_DELAY_STEP = 100
+        private const val MAX_WAIT_MILLIS = 5000L
+    }
+
+    constructor(allUpstreams: List<Upstream>,
+                matcher: Selector.Matcher,
+                pos: Int): this(allUpstreams, matcher, pos, 10, 7)
+
+    constructor(allUpstreams: List<Upstream>,
+                matcher: Selector.Matcher): this(allUpstreams, matcher, 0, 10, 10)
+
+    private val delay: Int
+    private val upstreams: List<Upstream>
+
+    private val control = EmitterProcessor.create<Boolean>(32, false)
+
+    init {
+        delay = if (jitter > 0) {
+            Random.nextInt(DEFAULT_DELAY_STEP - jitter, DEFAULT_DELAY_STEP + jitter)
+        } else {
+            DEFAULT_DELAY_STEP
+        }
+
+        upstreams = if (allUpstreams.size == 1 || pos == 0 || allUpstreams.isEmpty()) {
+            allUpstreams
+        } else {
+            val safePosition = pos % allUpstreams.size
+            allUpstreams.subList(safePosition, allUpstreams.size) + allUpstreams.subList(0, safePosition)
+        }
+    }
+
+    fun waitDuration(rawn: Long): Duration {
+        val n = max(rawn, 1)
+        val time = min(
+                (n.toDouble().pow(2.0) * delay).roundToLong(),
+                MAX_WAIT_MILLIS
+        )
+        return Duration.ofMillis(time)
+    }
+
+    override fun subscribe(subscriber: Subscriber<in DirectEthereumApi>) {
+        val first = Flux.fromIterable(upstreams)
+        val retries = (1 until repeatLimit).map { r ->
+            Flux.fromIterable(upstreams).delaySubscription(waitDuration(r))
+        }.let { Flux.concat(it) }
+
+        Flux.concat(first, retries)
+                .filter(Upstream::isAvailable)
+                .filter(matcher::matches)
+                .flatMap { it.getApi(matcher) }
+                .zipWith(control).map { it.t1 }
+                .subscribe(subscriber)
+    }
+
+    override fun resolve() {
+        control.onComplete()
+    }
+
+    override fun request(tries: Int) {
+        //TODO check the buffer size before submitting
+        repeat(tries) {
+            control.onNext(true)
+        }
+    }
+}

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

@@ -1,59 +0,0 @@
-/**
- * Copyright (c) 2019 ETCDEV GmbH
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *        http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.emeraldpay.dshackle.upstream
-
-import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
-
-class FilteringApiIterator(
-        private val upstreams: List<Upstream>,
-        private var pos: Int,
-        private val matcher: Selector.Matcher,
-        private val repeatLimit: Int = 5
-): Iterator<DirectEthereumApi> {
-
-    private var nextUpstream: Upstream? = null
-    private var consumed = 0
-
-    private fun nextInternal(): Boolean {
-        if (nextUpstream != null) {
-            return true
-        }
-        while (nextUpstream == null) {
-            consumed++
-            if (consumed > upstreams.size * repeatLimit) {
-                return false
-            }
-            val upstream = upstreams[pos++ % upstreams.size]
-            if (upstream.isAvailable() && matcher.matches(upstream)) {
-                nextUpstream = upstream
-            }
-        }
-        return nextUpstream != null
-    }
-
-    override fun hasNext(): Boolean {
-        return nextInternal()
-    }
-
-    override fun next(): DirectEthereumApi {
-        if (nextInternal()) {
-            val curr = nextUpstream!!
-            nextUpstream = null
-            return curr.getApi(matcher)
-        }
-        throw IllegalStateException("No upstream API available")
-    }
-}

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

@@ -19,13 +19,14 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 
 interface Upstream {
     fun isAvailable(): Boolean
     fun getStatus(): UpstreamAvailability
     fun observeStatus(): Flux<UpstreamAvailability>
     fun getHead(): EthereumHead
-    fun getApi(matcher: Selector.Matcher): DirectEthereumApi
+    fun getApi(matcher: Selector.Matcher): Mono<DirectEthereumApi>
     fun getOptions(): UpstreamsConfig.Options
     fun setLag(lag: Long)
     fun getLag(): Long

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

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
 import io.infinitape.etherjar.rpc.Batch
 import io.infinitape.etherjar.rpc.Commands
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 import java.time.Duration
 import java.util.concurrent.TimeUnit
 
@@ -28,28 +29,29 @@ class UpstreamValidator(
         private val options: UpstreamsConfig.Options
 ) {
 
-    fun validate(): UpstreamAvailability {
+    fun validate(): Mono<UpstreamAvailability> {
         val batch = Batch()
         val peerCount = batch.add(Commands.net().peerCount())
         val syncing = batch.add(Commands.eth().syncing())
-        try {
-            ethereumUpstream.getApi(Selector.empty).rpcClient.execute(batch).get(5, TimeUnit.SECONDS)
-            if (syncing.get().isSyncing) {
-                return UpstreamAvailability.SYNCING
-            }
-            if (options.minPeers != null && peerCount.get() < options.minPeers!!) {
-                return UpstreamAvailability.IMMATURE
-            }
-            return UpstreamAvailability.OK
-        } catch (e: Throwable) {
-            return UpstreamAvailability.UNAVAILABLE
-        }
+        return ethereumUpstream.getApi(Selector.empty)
+                .map { api -> api.rpcClient.execute(batch) }
+                .flatMap { Mono.fromCompletionStage(it) }
+                .timeout(Duration.ofSeconds(10))
+                .map {
+                    if (syncing.get().isSyncing) {
+                        UpstreamAvailability.SYNCING
+                    } else if (options.minPeers != null && peerCount.get() < options.minPeers!!) {
+                        UpstreamAvailability.IMMATURE
+                    } else {
+                        UpstreamAvailability.OK
+                    }
+                }.onErrorContinue { _, _ -> UpstreamAvailability.UNAVAILABLE }
     }
 
     fun start(): Flux<UpstreamAvailability> {
         return Flux.interval(Duration.ofSeconds(15))
-                .map {
+                .flatMap {
                     validate()
-                }.onErrorContinue { _, _ -> UpstreamAvailability.UNAVAILABLE }
+                }
     }
 }

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

@@ -21,6 +21,7 @@ import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
+import reactor.core.publisher.Mono
 import java.time.Duration
 
 open class EthereumUpstream(
@@ -102,8 +103,8 @@ open class EthereumUpstream(
         return head
     }
 
-    override fun getApi(matcher: Selector.Matcher): DirectEthereumApi {
-        return api
+    override fun getApi(matcher: Selector.Matcher): Mono<DirectEthereumApi> {
+        return Mono.just(api)
     }
 
     override fun getOptions(): UpstreamsConfig.Options {

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

@@ -121,7 +121,7 @@ open class GrpcUpstream(
             curr == null || curr.totalDifficulty < block.totalDifficulty
         }.flatMap {
             getApi(Selector.EmptyMatcher())
-                    .executeAndConvert(Commands.eth().getBlock(it.hash))
+                    .flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(it.hash)) }
                     .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream")))
                     .doOnError { t ->
                         val msg = "Failed to download block data for chain $chain"
@@ -194,8 +194,8 @@ open class GrpcUpstream(
         return head
     }
 
-    override fun getApi(matcher: Selector.Matcher): DirectEthereumApi {
-        return createApi(matcher)
+    override fun getApi(matcher: Selector.Matcher): Mono<DirectEthereumApi> {
+        return Mono.just(createApi(matcher))
     }
 
     override fun getOptions(): UpstreamsConfig.Options {