problem: wastes 200ms even for a single call because takes next api before validation

solution: manage apis flow right after value recorded

src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt

@@ -32,9 +32,10 @@ open class AlwaysQuorum: CallQuorum {
         return resolved
     }
 
-    override fun record(response: ByteArray, upstream: Upstream) {
+    override fun record(response: ByteArray, upstream: Upstream): Boolean {
         result = response
         resolved = true
+        return true
     }
 
     override fun getResult(): ByteArray? {

src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt

@@ -28,7 +28,7 @@ interface CallQuorum {
     fun init(head: Head<BlockJson<TransactionId>>)
 
     fun isResolved(): Boolean
-    fun record(response: ByteArray, upstream: Upstream)
+    fun record(response: ByteArray, upstream: Upstream): Boolean
     fun getResult(): ByteArray?
 
     companion object {

src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt

@@ -32,11 +32,13 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
         return result.get() != null
     }
 
-    override fun record(response: ByteArray, upstream: Upstream) {
+    override fun record(response: ByteArray, upstream: Upstream): Boolean {
         val lagging = upstream.getLag() > maxLag
         if (!lagging) {
             result.set(response)
+            return true
         }
+        return false
     }
 
     override fun getResult(): ByteArray {

src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt

@@ -31,7 +31,7 @@ abstract class ValueAwareQuorum<T>(
         return jacksonRpcConverter.fromJson(response.inputStream(), clazz)
     }
 
-    override fun record(response: ByteArray, upstream: Upstream) {
+    override fun record(response: ByteArray, upstream: Upstream): Boolean {
         try {
             val value = extractValue(response, clazz)
             recordValue(response, value, upstream)
@@ -40,6 +40,7 @@ abstract class ValueAwareQuorum<T>(
         } catch (e: Exception) {
             recordError(response, e.message, upstream)
         }
+        return isResolved();
     }
 
     abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream)

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

@@ -27,13 +27,11 @@ import org.apache.commons.lang3.StringUtils
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
-import reactor.core.publisher.Flux
+import reactor.core.publisher.*
-import reactor.core.publisher.Mono
-import reactor.core.publisher.toFlux
-import reactor.core.publisher.toMono
 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
@@ -124,17 +122,26 @@ class NativeCall(
     }
 
     fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
-        val p: Predicate<Any> = CallQuorum.untilResolved(ctx.callQuorum)
         val all = ctx.getApis().toFlux().share()
-        //execute on the first API immediately, and then make a delay between each call to not dos upstreams
+        //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 retries = Flux.from(all).delayElements(Duration.ofMillis(200))
+        val repeatControl = EmitterProcessor.create<Boolean>()//TopicProcessor.create<Boolean>()
+        val retries = Flux.from(all).skip(1)
+                .zipWith(repeatControl).zipWith(Flux.interval(Duration.ofMillis(200)))
+                .map { it.t1.t1 }
+
         return Flux.concat(immediate, retries)
-                .takeWhile(p)
                 .flatMap { api ->
                     api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) }
                 }
-                .reduce(ctx.callQuorum, CallQuorum.asReducer())
+                .reduce(ctx.callQuorum, {res, a ->
+                    if (res.record(a.t1, a.t2)) {
+                        repeatControl.onComplete()
+                    } else {
+                        repeatControl.onNext(true)
+                    }
+                    res
+                })
                 .filter { it.isResolved() }
                 .map {
                     val result  = it.getResult()

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

@@ -21,7 +21,7 @@ class FilteringApiIterator(
         private val upstreams: List<Upstream>,
         private var pos: Int,
         private val matcher: Selector.Matcher,
-        private val repeatLimit: Int = 3
+        private val repeatLimit: Int = 5
 ): Iterator<DirectEthereumApi> {
 
     private var nextUpstream: Upstream? = null

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

@@ -80,6 +80,7 @@ class QuorumBasedMethods(
     override fun getQuorumFor(method: String): CallQuorum {
         return when {
             hardcodedMethods.contains(method) -> AlwaysQuorum()
+            firstValueMethods.contains(method) -> AlwaysQuorum()
             anyResponseMethods.contains(method) -> NotLaggingQuorum(6)
             headVerifiedMethods.contains(method) -> NotLaggingQuorum(1)
             specialMethods.contains(method) -> {

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy

@@ -57,7 +57,6 @@ class NativeCallSpec extends Specification {
         def act = objectMapper.readValue(resp.payload, Map)
         then:
         act == [jsonrpc:"2.0", id:1, result: "foo"]
-        (2..3) * quorum.isResolved() // 2 times during api call (before and after) + 1 time in a filter after
         1 * quorum.record(_, _)
         1 * quorum.getResult()
     }
@@ -86,11 +85,61 @@ class NativeCallSpec extends Specification {
         def act = objectMapper.readValue(resp.payload, Map)
         then:
         act == [jsonrpc:"2.0", id:1, result: "bar"]
-        (3..4) * quorum.isResolved()
         2 * quorum.record(_, _)
         1 * quorum.getResult()
     }
 
+    def "Have pause between repeats"() {
+        setup:
+        def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3))
+
+        def upstreams = Stub(Upstreams)
+        RpcClient rpcClient = Stub(RpcClient)
+        def apiMock = TestingCommons.api(rpcClient)
+        apiMock.upstream = Stub(Upstream)
+
+        apiMock.answerOnce("eth_test", [], null)
+        apiMock.answerOnce("eth_test", [], "bar")
+
+        def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
+        def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock),
+                Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_test", []))
+
+
+        when:
+        def t1 = System.currentTimeMillis()
+        nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2))
+        def delta = System.currentTimeMillis() - t1
+        then:
+        delta >= 200
+    }
+
+    def "One call has no pause"() {
+        setup:
+        def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3))
+
+        def upstreams = Stub(Upstreams)
+        RpcClient rpcClient = Stub(RpcClient)
+        def apiMock = TestingCommons.api(rpcClient)
+        apiMock.upstream = Stub(Upstream)
+
+        apiMock.answerOnce("eth_test", [], "bar")
+
+        def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper())
+        def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock),
+                Selector.empty, quorum,
+                new NativeCall.ParsedCallDetails("eth_test", []))
+
+
+        when:
+        def t1 = System.currentTimeMillis()
+        nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2))
+        def delta = System.currentTimeMillis() - t1
+        then:
+        delta < 50
+    }
+
     def "Returns error if no quorum"() {
         setup:
         def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3))
@@ -107,7 +156,6 @@ class NativeCallSpec extends Specification {
         def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), Selector.empty, quorum,
                 new NativeCall.ParsedCallDetails("eth_test", []))
 
-        (4..5) * quorum.isResolved()
         3 * quorum.record(_, _)
         1 * quorum.getResult()