diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 43266c52..1c8c003f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/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 }, - ).delaySubscription(Duration.ofMillis(retryDelay.toLong())) + Flux.fromIterable(standardWithFallback.sortedBy { up -> up.getStatus().grpcId }) }.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 -> - val matchesResponse = internalMatcher.matchesWithCause(up) - processMatchesResponse(up.getId(), matchesResponse) - matchesResponse.matched() - } - .zipWith(control.asFlux()) - .map { + control.asFlux() + .zipWith(result) + .map { it.t2 } + .filter { up -> + val matchesResponse = internalMatcher.matchesWithCause(up) + processMatchesResponse(up.getId(), matchesResponse) + matchesResponse.matched() + .also { + if (!it) { + this.request(1) + } + } + } + .doOnNext { upstreamsMatchesResponse = null counter.incrementAndGet() - it.t1 } .doOnSubscribe { if (!started) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt index 0ef131b1..76cb719c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt +++ b/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> { - 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}") } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt index a0df02c8..ec21d9a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt +++ b/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() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt index 52287574..2883309c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt +++ b/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() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultStarknetMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultStarknetMethods.kt index 9be630c3..7e0c5f72 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultStarknetMethods.kt +++ b/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() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt index 3aaebc68..2ac5484e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt +++ b/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 && - block.height >= (curr?.height?.minus(10) ?: 0) + return seenBlocks.getIfPresent(block.hash) == null && block.height > (curr?.height ?: 0) } override fun choose(block: BlockContainer): ForkChoice.ChoiceResult {