diff --git a/gradle.properties b/gradle.properties index eacee52bf..8af90b978 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,6 +1,6 @@ kotlin.code.style=official -grpcVersion=1.49.2 +grpcVersion=1.81.0 reactiveGrpcVersion=1.2.4 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index bbcb2088f..506e102fd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -181,7 +181,8 @@ abstract class AbstractHead @JvmOverloads constructor( } protected open fun onNoHeadUpdates() { - // NOOP + // the head is stuck, e.g. upstream height went below the last accepted block + forkChoice.reset() } override fun onSyncingNode(isSyncing: Boolean) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt index dad06e075..e367893db 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt @@ -32,6 +32,8 @@ class EthereumLowerBoundProofDetector( "not supported", "evm module does not exist on height", "state is not available", // opbnb / bsc — eth_getProof on pruned state + "World state unavailable", // besu, linea-besu + "Worldstate unavailable", ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt index 2db6006ab..4d1267975 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/ForkChoice.kt @@ -14,4 +14,9 @@ interface ForkChoice { fun filter(block: BlockContainer): Boolean fun choose(block: BlockContainer): ChoiceResult + + /** + * Called when the head is stuck; lets the fork choice accept a block it would otherwise reject + */ + fun reset() {} } 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 2ac5484e0..971fb7c2d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt @@ -4,6 +4,7 @@ import com.google.common.cache.CacheBuilder import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import org.slf4j.LoggerFactory +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicReference class PriorityForkChoice : ForkChoice { @@ -11,6 +12,7 @@ class PriorityForkChoice : ForkChoice { private val seenBlocks = CacheBuilder.newBuilder() .maximumSize(10) .build() + private val acceptNext = AtomicBoolean(false) companion object { private val log = LoggerFactory.getLogger(PriorityForkChoice::class.java) @@ -22,7 +24,12 @@ class PriorityForkChoice : ForkChoice { override fun filter(block: BlockContainer): Boolean { val curr = head.get() - return seenBlocks.getIfPresent(block.hash) == null && block.height > (curr?.height ?: 0) + return seenBlocks.getIfPresent(block.hash) == null && + (acceptNext.get() || block.height > (curr?.height ?: 0)) + } + + override fun reset() { + acceptNext.set(true) } override fun choose(block: BlockContainer): ForkChoice.ChoiceResult { @@ -36,6 +43,7 @@ class PriorityForkChoice : ForkChoice { } else { log.trace("Preparing to accept block ${block.height}") seenBlocks.put(block.hash, true) + acceptNext.set(false) block } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt index dcfcf0f01..49cdab021 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/detector/RecursiveLowerBound.kt @@ -183,7 +183,7 @@ open class RecursiveLowerBound( protected fun retrySpec(block: Long, nonRetryableErrors: Set): RetryBackoffSpec { return Retry.backoff( - Long.MAX_VALUE, + MAX_RETRIES, Duration.ofSeconds(1), ) .maxBackoff(Duration.ofMinutes(3)) @@ -192,13 +192,13 @@ open class RecursiveLowerBound( !nonRetryableErrorPatters.any { err -> it.message?.matches(err) ?: false } } .doAfterRetry { - if (it.totalRetries() > 30) { + if (it.totalRetries() == MAX_RETRIES - 1) { log.warn( "There are too much retries to calculate {} lower bound of upstream {}, block {} " + "probably this error with message `{}` is not retryable, please report it to dshackle devs", - block, type, upstream.getId(), + block, it.failure().message, ) } else { @@ -229,4 +229,9 @@ open class RecursiveLowerBound( constructor(current: Long, found: Boolean) : this(0, 0, current, found) } + + companion object { + // ponytail: ~70 min per block at max backoff, then the block counts as "no data" + private const val MAX_RETRIES = 30L + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy index 37004a631..04bfa4b7a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoiceSpec.groovy @@ -38,4 +38,18 @@ class PriorityForkChoiceSpec extends Specification { then: choice.getHead() == blocks[2] } + + def "accepts a lower block once after reset"() { + def choice = new PriorityForkChoice() + choice.choose(blocks[3]) + when: + choice.reset() + choice.choose(blocks[1]) + then: + choice.getHead() == blocks[1] + when: + choice.choose(blocks[0]) + then: + choice.getHead() == blocks[1] + } } diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/RecursiveLowerBoundServiceTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/RecursiveLowerBoundServiceTest.kt index bbe29ad24..94eb2bdf1 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/RecursiveLowerBoundServiceTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/RecursiveLowerBoundServiceTest.kt @@ -3,10 +3,13 @@ package io.emeraldpay.dshackle.upstream.lowerbound import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.reader.ChainReader +import io.emeraldpay.dshackle.upstream.ChainCallError +import io.emeraldpay.dshackle.upstream.ChainCallUpstreamException import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundProofDetector import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundService import io.emeraldpay.dshackle.upstream.ethereum.EthereumLowerBoundTxDetector.Companion.MAX_OFFSET import io.emeraldpay.dshackle.upstream.ethereum.ZERO_ADDRESS @@ -18,7 +21,9 @@ import org.junit.jupiter.api.Test import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.Arguments import org.junit.jupiter.params.provider.MethodSource +import org.junit.jupiter.params.provider.ValueSource import org.mockito.kotlin.any +import org.mockito.kotlin.doAnswer import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock import reactor.core.publisher.Mono @@ -243,6 +248,43 @@ class RecursiveLowerBoundServiceTest { ) } + @ParameterizedTest + @ValueSource(strings = ["World state unavailable", "Worldstate unavailable", "some unknown node error"]) + fun `proof bound settles on pruned state errors`(error: String) { + val height = 32_048_166L + val available = height - 512 + val head = mock { + on { getCurrentHeight() } doReturn height + } + val reader = mock { + on { read(any()) } doAnswer { inv -> + val block = ((inv.getArgument(0).params as ListParams).list[2] as String) + .removePrefix("0x").toLong(16) + // defer: HttpReader re-sends the request on every retry + Mono.defer { + if (block >= available) { + Mono.just(ChainResponse("{}".toByteArray(), null)) + } else { + Mono.error(ChainCallUpstreamException(ChainResponse.NumberId(1), ChainCallError(-32000, error))) + } + } + } + } + val upstream = mock { + on { getId() } doReturn "id" + on { getHead() } doReturn head + on { getIngressReader() } doReturn reader + on { getChain() } doReturn Chain.UNSPECIFIED + } + + StepVerifier.withVirtualTime { EthereumLowerBoundProofDetector(upstream).detectLowerBound(NoopManualLowerBoundService()) } + .expectSubscription() + .thenAwait(Duration.ofDays(3)) + .expectNextMatches { it.lowerBound == available && it.type == LowerBoundType.PROOF } + .thenCancel() + .verify(Duration.ofSeconds(30)) + } + companion object { private const val STATE_CHECKER_ADDRESS = "0x1111111111111111111111111111111111111111" private const val STATE_CHECKER_CALL_DATA = "0x1eaf190c"