Change logic of upstreams filtering before request (#389)

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

@@ -26,7 +26,6 @@ import org.reactivestreams.Subscriber
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Sinks
-import java.time.Duration
 import java.util.EnumMap
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.locks.Lock
@@ -141,10 +140,7 @@ class FilteredApis(
         val second = Flux.fromIterable(secondaryUpstreams.sortedBy { it.getStatus().grpcId })
         // if all failed, try both standard and fallback upstreams, repeating in cycle
         val retries = (0 until this.retries).map {
-            val retryDelay = (it + 1) * 30
+            Flux.fromIterable(standardWithFallback.sortedBy { up -> up.getStatus().grpcId })
-            Flux.fromIterable(
-                standardWithFallback.sortedBy { up -> up.getStatus().grpcId },
-            ).delaySubscription(Duration.ofMillis(retryDelay.toLong()))
         }.let { Flux.concat(it) }
 
         val size = primaryUpstreams.size + secondaryUpstreams.size + standardWithFallback.size * this.retries
@@ -157,16 +153,22 @@ class FilteredApis(
                 .doFinally { metrics[chain]?.tried?.record(count.toDouble()) }
         }
 
-        result.filter { up ->
+        control.asFlux()
-            val matchesResponse = internalMatcher.matchesWithCause(up)
+            .zipWith(result)
-            processMatchesResponse(up.getId(), matchesResponse)
+            .map { it.t2 }
-            matchesResponse.matched()
+            .filter { up ->
-        }
+                val matchesResponse = internalMatcher.matchesWithCause(up)
-            .zipWith(control.asFlux())
+                processMatchesResponse(up.getId(), matchesResponse)
-            .map {
+                matchesResponse.matched()
+                    .also {
+                        if (!it) {
+                            this.request(1)
+                        }
+                    }
+            }
+            .doOnNext {
                 upstreamsMatchesResponse = null
                 counter.incrementAndGet()
-                it.t1
             }
             .doOnSubscribe {
                 if (!started) {

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

@@ -37,7 +37,7 @@ class HeadLagObserver(
     private val distanceExtractor: Extractor,
     private val lagObserverScheduler: Scheduler,
     private val forkDistance: Long,
-    private val throttling: Duration = Duration.ofSeconds(5),
+    private val throttling: Duration = Duration.ofMillis(500),
 ) : Lifecycle {
 
     private val log = LoggerFactory.getLogger(HeadLagObserver::class.java)
@@ -63,13 +63,13 @@ class HeadLagObserver(
             .sample(throttling)
             .flatMap(this::probeFollowers)
             .map { item ->
-                log.debug("Set lag ${item.t1} to upstream ${item.t2.getId()}")
+                log.trace("Set lag ${item.t1} to upstream ${item.t2.getId()}")
                 item.t2.setLag(item.t1)
             }
     }
 
     fun probeFollowers(top: BlockContainer): Flux<Tuple2<Long, Upstream>> {
-        log.debug("Compute lag for ${followers.map { it.getId() }}")
+        log.trace("Compute lag for ${followers.map { it.getId() }}")
 
         return Flux.fromIterable(followers)
             .parallel(followers.size)
@@ -92,7 +92,7 @@ class HeadLagObserver(
                 log.warn("Failed to find distance for $up", t)
             }
             .doOnNext {
-                log.debug("Lag for ${it.t2.getId()} is ${it.t1}")
+                log.trace("Lag for ${it.t2.getId()} is ${it.t1}")
             }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt

@@ -19,7 +19,6 @@ import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.BroadcastQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
-import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
 import io.emeraldpay.dshackle.quorum.NotNullQuorum
 import io.emeraldpay.etherjar.rpc.RpcException
 import java.util.Collections
@@ -68,10 +67,7 @@ class DefaultBitcoinMethods : CallMethods {
 
     override fun createQuorumFor(method: String): CallQuorum {
         return when {
-            Collections.binarySearch(hardcodedMethods, method) >= 0 -> AlwaysQuorum()
             Collections.binarySearch(anyResponseMethods, method) >= 0 -> NotNullQuorum()
-            Collections.binarySearch(freshMethods, method) >= 0 -> NotLaggingQuorum(2)
-            Collections.binarySearch(headVerifiedMethods, method) >= 0 -> NotLaggingQuorum(0)
             Collections.binarySearch(broadcastMethods, method) >= 0 -> BroadcastQuorum()
             else -> AlwaysQuorum()
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt

@@ -22,7 +22,6 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.BroadcastQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
 import io.emeraldpay.dshackle.quorum.MaximumValueQuorum
-import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
 import io.emeraldpay.dshackle.quorum.NotNullQuorum
 import io.emeraldpay.etherjar.rpc.RpcException
 
@@ -145,35 +144,15 @@ class DefaultEthereumMethods(
 
     override fun createQuorumFor(method: String): CallQuorum {
         return when {
-            newFilterMethods.contains(method) -> NotLaggingQuorum(4)
-            withFilterIdMethods.contains(method) -> AlwaysQuorum()
-            hardcodedMethods.contains(method) -> AlwaysQuorum()
-            firstValueMethods.contains(method) -> AlwaysQuorum()
-            anyResponseMethods.contains(method) -> NotLaggingQuorum(4)
-            headVerifiedMethods.contains(method) -> NotLaggingQuorum(1)
-            getDrpcVendorMethods(chain).contains(method) -> AlwaysQuorum()
             possibleNotIndexedMethods.contains(method) -> NotNullQuorum()
             specialMethods.contains(method) -> {
                 when (method) {
                     "eth_getTransactionCount" -> MaximumValueQuorum()
-                    "eth_getBalance" -> AlwaysQuorum()
-                    "eth_blockNumber" -> NotLaggingQuorum(0)
                     "eth_sendRawTransaction" -> BroadcastQuorum()
                     else -> AlwaysQuorum()
                 }
             }
 
-            getChainSpecificMethods(chain).contains(method) -> {
-                when (method) {
-                    "bor_getAuthor" -> NotLaggingQuorum(4)
-                    "bor_getCurrentValidators" -> NotLaggingQuorum(0)
-                    "bor_getCurrentProposer" -> NotLaggingQuorum(0)
-                    "bor_getRootHash" -> NotLaggingQuorum(4)
-                    "eth_getRootHash" -> NotLaggingQuorum(4)
-                    else -> AlwaysQuorum()
-                }
-            }
-
             else -> AlwaysQuorum()
         }
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultStarknetMethods.kt

@@ -21,7 +21,6 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum
 import io.emeraldpay.dshackle.quorum.BroadcastQuorum
 import io.emeraldpay.dshackle.quorum.CallQuorum
 import io.emeraldpay.dshackle.quorum.MaximumValueQuorum
-import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
 import io.emeraldpay.dshackle.quorum.NotNullQuorum
 import io.emeraldpay.etherjar.rpc.RpcException
 
@@ -86,11 +85,7 @@ class DefaultStarknetMethods(
         return when {
             nonce.contains(method) -> MaximumValueQuorum()
             add.contains(method) -> BroadcastQuorum()
-            hardcodedMethods.contains(method) -> AlwaysQuorum()
-            nonLagging.contains(method) -> NotLaggingQuorum(1)
             nonNull.contains(method) -> NotNullQuorum()
-            firstValueMethods.contains(method) -> AlwaysQuorum()
-            anyResponseMethods.contains(method) -> NotLaggingQuorum(4)
 
             else -> AlwaysQuorum()
         }

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

@@ -22,8 +22,7 @@ class PriorityForkChoice : ForkChoice {
 
     override fun filter(block: BlockContainer): Boolean {
         val curr = head.get()
-        return (curr == null || curr.nodeRating <= block.nodeRating) && seenBlocks.getIfPresent(block.hash) == null &&
+        return seenBlocks.getIfPresent(block.hash) == null && block.height > (curr?.height ?: 0)
-            block.height >= (curr?.height?.minus(10) ?: 0)
     }
 
     override fun choose(block: BlockContainer): ForkChoice.ChoiceResult {