Recover stuck multistream head and cap lower bound retries

gradle.properties

@@ -1,6 +1,6 @@
 kotlin.code.style=official
 
-grpcVersion=1.49.2
+grpcVersion=1.81.0
 reactiveGrpcVersion=1.2.4
 
 

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) {

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",
         )
     }
 

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() {}
 }

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

@@ -4,13 +4,15 @@ 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 {
     private val head = AtomicReference<BlockContainer>(null)
     private val seenBlocks = CacheBuilder.newBuilder()
         .maximumSize(10)
         .build<BlockId, Boolean>()
+    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
             }
         }

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<String>): 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
+    }
 }

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]
+    }
 }

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<Head> {
+            on { getCurrentHeight() } doReturn height
+        }
+        val reader = mock<ChainReader> {
+            on { read(any()) } doAnswer { inv ->
+                val block = ((inv.getArgument<ChainRequest>(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<Upstream> {
+            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"