Staterecover part5 (#565)

* staterecovery: fix testing tx sender helper * staterecovery: improve FakeExecutionLayerClient * staterecovery: improve LogsSearcher to be resilient to query failures * staterecovery: fix scenario where headblock greater lastFinalizedBlock * staterecovery: simplify RecoveryModeManager * staterecovery: improve test code reuse and reliability * staterecovery: improve logsSearcher tests * Apply suggestions from code review Co-authored-by: jonesho <81145364+jonesho@users.noreply.github.com> Signed-off-by: Pedro Novais <1478752+jpnovais@users.noreply.github.com> * staterecovery: fix boolean check --------- Signed-off-by: Pedro Novais <1478752+jpnovais@users.noreply.github.com> Co-authored-by: jonesho <81145364+jonesho@users.noreply.github.com>
Consensys · Jan 16, 2025 · be18423 · be18423
1 parent 05d3ee2
commit be18423
Show file tree

Hide file tree

Showing 17 changed files with 251 additions and 136 deletions.
diff --git a/coordinator/ethereum/test-utils/build.gradle b/coordinator/ethereum/test-utils/build.gradle
@@ -9,6 +9,7 @@ dependencies {
   implementation(project(":coordinator:core"))
   implementation(project(":coordinator:clients:smart-contract-client"))
   implementation(project(":jvm-libs:linea:web3j-extensions"))
+  implementation(testFixtures(project(":jvm-libs:linea:web3j-extensions")))
   implementation(project(":coordinator:ethereum:common"))
   implementation(project(":jvm-libs:linea:testing:file-system"))
   implementation("org.web3j:core:${libs.versions.web3j.get()}") {

diff --git a/...inator/ethereum/test-utils/src/main/kotlin/net/consensys/zkevm/ethereum/AccountManager.kt b/...inator/ethereum/test-utils/src/main/kotlin/net/consensys/zkevm/ethereum/AccountManager.kt
@@ -3,6 +3,7 @@ package net.consensys.zkevm.ethereum
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.fasterxml.jackson.module.kotlin.readValue
 import kotlinx.datetime.Clock
+import linea.web3j.waitForTxReceipt
 import net.consensys.linea.contract.AsyncFriendlyTransactionManager
 import net.consensys.linea.jsonrpc.JsonRpcErrorResponseException
 import net.consensys.linea.testing.filesystem.getPathTo

diff --git a/jvm-libs/linea/testing/l1-blob-and-proof-submission/build.gradle b/jvm-libs/linea/testing/l1-blob-and-proof-submission/build.gradle
@@ -8,6 +8,8 @@ dependencies {
   api(project(':jvm-libs:linea:core:domain-models'))
   implementation(project(':jvm-libs:generic:extensions:kotlin'))
   implementation(project(':jvm-libs:linea:testing:file-system'))
+  implementation(project(':jvm-libs:linea:web3j-extensions'))
+  implementation(testFixtures(project(':jvm-libs:linea:web3j-extensions')))
   implementation(project(':coordinator:clients:prover-client:serialization'))
   implementation(testFixtures(project(":coordinator:core")))
 }
diff --git a/...submission/src/main/kotlin/net/consensys/linea/testing/submission/SubmissionTestHelper.kt b/...submission/src/main/kotlin/net/consensys/linea/testing/submission/SubmissionTestHelper.kt
@@ -1,8 +1,11 @@
 package net.consensys.linea.testing.submission
 
+import linea.web3j.waitForTxReceipt
 import net.consensys.zkevm.coordinator.clients.smartcontract.LineaRollupSmartContractClient
 import net.consensys.zkevm.domain.Aggregation
-import tech.pegasys.teku.infrastructure.async.SafeFuture
+import org.web3j.protocol.Web3j
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.minutes
 
 /**
  * Submits blobs respecting aggregation boundaries
@@ -18,10 +21,9 @@ fun submitBlobs(
   return aggregationsAndBlobs
     .map { (_, aggBlobs) ->
       val blobChunks = aggBlobs.chunked(blobChunksSize)
-      blobChunks.map { blobs -> contractClient.submitBlobs(blobs, gasPriceCaps = null) }
+      blobChunks.map { blobs -> contractClient.submitBlobs(blobs, gasPriceCaps = null).get() }
     }
     .flatten()
-    .let { SafeFuture.collectAll(it.stream()).get() }
 }
 
 /**
@@ -34,13 +36,13 @@ data class SubmissionTxHashes(
   val blobTxHashes: List<String>,
   val aggregationTxHashes: List<String>
 )
+
 fun submitBlobsAndAggregations(
   contractClient: LineaRollupSmartContractClient,
   aggregationsAndBlobs: List<AggregationAndBlobs>,
   blobChunksSize: Int = 6
 ): SubmissionTxHashes {
   val blobSubmissionTxHashes = submitBlobs(contractClient, aggregationsAndBlobs, blobChunksSize)
-
   return aggregationsAndBlobs
     .filter { it.aggregation != null }
     .mapIndexed { index, (aggregation, aggBlobs) ->
@@ -53,8 +55,33 @@ fun submitBlobsAndAggregations(
         parentL1RollingHash = parentAgg?.aggregationProof?.l1RollingHash ?: ByteArray(32),
         parentL1RollingHashMessageNumber = parentAgg?.aggregationProof?.l1RollingHashMessageNumber ?: 0L,
         gasPriceCaps = null
-      )
+      ).get()
     }
-    .let { SafeFuture.collectAll(it.stream()).get() }
     .let { SubmissionTxHashes(blobSubmissionTxHashes, it) }
 }
+
+fun submitBlobsAndAggregationsAndWaitExecution(
+  contractClient: LineaRollupSmartContractClient,
+  aggregationsAndBlobs: List<AggregationAndBlobs>,
+  blobChunksSize: Int = 6,
+  l1Web3jClient: Web3j,
+  waitTimeout: Duration = 2.minutes
+) {
+  val submissionTxHashes = submitBlobsAndAggregations(
+    contractClient = contractClient,
+    aggregationsAndBlobs = aggregationsAndBlobs,
+    blobChunksSize = blobChunksSize
+  )
+
+  l1Web3jClient.waitForTxReceipt(
+    txHash = submissionTxHashes.aggregationTxHashes.last(),
+    timeout = waitTimeout
+  ).also { txReceipt ->
+    if (txReceipt.status != "0x1") {
+      val lastAggregation = aggregationsAndBlobs.findLast { it.aggregation != null }!!.aggregation!!
+      throw IllegalStateException(
+        "latest finalization=${lastAggregation.intervalString()} failed on L1. receipt=$txReceipt"
+      )
+    }
+  }
+}
diff --git a/jvm-libs/linea/web3j-extensions/build.gradle b/jvm-libs/linea/web3j-extensions/build.gradle
@@ -1,6 +1,7 @@
 plugins {
   id 'net.consensys.zkevm.kotlin-common-conventions'
   id 'java-library'
+  id 'java-test-fixtures'
 }
 
 description="Web3j extensions for Linea"

diff --git a/jvm-libs/linea/web3j-extensions/src/main/kotlin/linea/web3j/Web3JLogsSearcher.kt b/jvm-libs/linea/web3j-extensions/src/main/kotlin/linea/web3j/Web3JLogsSearcher.kt
@@ -115,12 +115,15 @@ class Web3JLogsSearcher(
         if (logs.isEmpty()) {
           SearchResult.NoResultsInInterval
         } else {
-          val item = logs.find { shallContinueToSearchPredicate(it) == null }
+          var nextSearchDirection: SearchDirection? = null
+          val item = logs.find {
+            nextSearchDirection = shallContinueToSearchPredicate(it)
+            nextSearchDirection == null
+          }
           if (item != null) {
             SearchResult.ItemFound(item)
           } else {
-            val nextSearchDirection = shallContinueToSearchPredicate(logs.first())!!
-            SearchResult.KeepSearching(nextSearchDirection)
+            SearchResult.KeepSearching(nextSearchDirection!!)
           }
         }
       }
@@ -195,15 +198,25 @@ class Web3JLogsSearcher(
     return if (fromBlock is BlockParameter.BlockNumber && toBlock is BlockParameter.BlockNumber) {
       return SafeFuture.completedFuture(Pair(fromBlock.getNumber(), toBlock.getNumber()))
     } else {
-      SafeFuture.collectAll(
-        web3jClient.ethGetBlockByNumber(fromBlock.toWeb3j(), false).sendAsync().toSafeFuture(),
-        web3jClient.ethGetBlockByNumber(toBlock.toWeb3j(), false).sendAsync().toSafeFuture()
-      ).thenApply { (fromBlockResponse, toBlockResponse) ->
-        Pair(
-          fromBlockResponse.block.number.toULong(),
-          toBlockResponse.block.number.toULong()
-        )
-      }
+      AsyncRetryer.retry(
+        vertx = vertx,
+        backoffDelay = config.backoffDelay,
+        stopRetriesPredicate = { (fromBlockResponse, toBlockResponse) ->
+          fromBlockResponse?.block?.number != null && toBlockResponse?.block?.number != null
+        },
+        action = {
+          SafeFuture.collectAll(
+            web3jClient.ethGetBlockByNumber(fromBlock.toWeb3j(), false).sendAsync().toSafeFuture(),
+            web3jClient.ethGetBlockByNumber(toBlock.toWeb3j(), false).sendAsync().toSafeFuture()
+          )
+        }
+      )
+        .thenApply { (fromBlockResponse, toBlockResponse) ->
+          Pair(
+            fromBlockResponse.block.number.toULong(),
+            toBlockResponse.block.number.toULong()
+          )
+        }
     }
   }
 }
diff --git a/jvm-libs/linea/web3j-extensions/src/test/kotlin/linea/web3j/Web3JLogsSearcherIntTest.kt b/jvm-libs/linea/web3j-extensions/src/test/kotlin/linea/web3j/Web3JLogsSearcherIntTest.kt
@@ -11,13 +11,15 @@ import io.vertx.core.Vertx
 import linea.SearchDirection
 import linea.domain.RetryConfig
 import linea.jsonrpc.TestingJsonRpcServer
+import linea.log4j.configureLoggers
 import net.consensys.encodeHex
 import net.consensys.fromHexString
 import net.consensys.linea.BlockParameter.Companion.toBlockParameter
 import net.consensys.linea.jsonrpc.JsonRpcError
 import net.consensys.linea.jsonrpc.JsonRpcRequest
 import net.consensys.toHexString
 import net.consensys.toHexStringUInt256
+import org.apache.logging.log4j.Level
 import org.apache.logging.log4j.LogManager
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
@@ -49,6 +51,11 @@ class Web3JLogsSearcherIntTest {
   @BeforeEach
   fun beforeEach() {
     vertx = Vertx.vertx()
+    configureLoggers(
+      rootLevel = Level.INFO,
+      log.name to Level.DEBUG,
+      "test.case.Web3JLogsSearcher" to Level.DEBUG
+    )
   }
 
   @AfterEach
@@ -255,11 +262,19 @@ class Web3JLogsSearcherIntTest {
     targetNumber: ULong
   ): SearchDirection? {
     val number = ULong.fromHexString(ethLog.topics[1].encodeHex())
-    return when {
+    val direction = when {
       number < targetNumber -> SearchDirection.FORWARD
       number > targetNumber -> SearchDirection.BACKWARD
       else -> null
     }
+    log.debug(
+      "EthLog{blockNumber={} number={}} targetNumber={} direction={}",
+      ethLog.blockNumber,
+      number,
+      targetNumber,
+      direction
+    )
+    return direction
   }
 
   @Test
@@ -287,8 +302,9 @@ class Web3JLogsSearcherIntTest {
   }
 
   @Test
-  fun `findLogs searches L1 and returns null when not found - before range`() {
+  fun `findLogs searches L1 and returns null when not found - target before fromBlock`() {
     setupClientWithTestingJsonRpcServer()
+    val logsEvaluated = mutableListOf<ULong>()
 
     logsClient.findLog(
       fromBlock = 100UL.toBlockParameter(),
@@ -297,21 +313,25 @@ class Web3JLogsSearcherIntTest {
       topics = listOf("0xffaabbcc"),
       chunkSize = 10,
       shallContinueToSearch = { ethLog ->
+        logsEvaluated.add(ethLog.blockNumber)
         shallContinueToSearch(ethLog, targetNumber = 89UL)
       }
     )
       .get()
       .also { log ->
         assertThat(log).isNull()
         assertThat(TestingJsonRpcServer.callCountByMethod("eth_getLogs")).isBetween(1, 4)
+        assertThat(logsEvaluated).hasSameElementsAs(logsEvaluated.toSet())
       }
   }
 
+  // REFERENCE:
   @Test
-  fun `findLogs searches L1 and returns null when no logs in blockRange`() {
+  fun `findLogs searches L1 and returns null when not found - target expected in chunk that has no logs`() {
     setupClientWithTestingJsonRpcServer(
       subsetOfBlocksWithLogs = listOf(100UL..109UL, 150UL..159UL)
     )
+    val logsEvaluated = mutableListOf<ULong>()
 
     logsClient.findLog(
       fromBlock = 100UL.toBlockParameter(),
@@ -320,18 +340,23 @@ class Web3JLogsSearcherIntTest {
       topics = listOf("0xffaabbcc"),
       chunkSize = 10,
       shallContinueToSearch = { ethLog ->
-        shallContinueToSearch(ethLog, targetNumber = 89UL)
+        shallContinueToSearch(ethLog, targetNumber = 120UL)
       }
     )
       .get()
       .also { log ->
         assertThat(log).isNull()
+        assertThat(logsEvaluated).hasSameElementsAs(logsEvaluated.toSet())
       }
   }
 
   @Test
-  fun `findLogs searches L1 and returns null when not found - after range`() {
-    setupClientWithTestingJsonRpcServer()
+  fun `findLogs searches L1 and returns null when not found - target is after toBlock`() {
+    setupClientWithTestingJsonRpcServer(
+      subsetOfBlocksWithLogs = listOf(100UL..109UL, 150UL..200UL)
+    )
+    val logsEvaluated = mutableListOf<ULong>()
+
     logsClient.findLog(
       fromBlock = 100UL.toBlockParameter(),
       toBlock = 200UL.toBlockParameter(),
@@ -346,14 +371,17 @@ class Web3JLogsSearcherIntTest {
       .also { log ->
         assertThat(log).isNull()
         assertThat(TestingJsonRpcServer.callCountByMethod("eth_getLogs")).isBetween(1, 4)
+        assertThat(logsEvaluated).hasSameElementsAs(logsEvaluated.toSet())
       }
   }
 
+  // REFERECE 2:
   @Test
-  fun `findLogs searches L1 and returns null when range has no logs`() {
+  fun `findLogs searches L1 and returns item when - target found an chunk in the middle`() {
     setupClientWithTestingJsonRpcServer(
-      subsetOfBlocksWithLogs = listOf(10UL..19UL, 50UL..59UL)
+      subsetOfBlocksWithLogs = listOf(10UL..19UL, 30UL..37UL, 50UL..100UL)
     )
+    val logsEvaluated = mutableListOf<ULong>()
     logsClient.findLog(
       fromBlock = 0UL.toBlockParameter(),
       toBlock = 100UL.toBlockParameter(),
@@ -362,13 +390,14 @@ class Web3JLogsSearcherIntTest {
       chunkSize = 5,
       shallContinueToSearch = { ethLog ->
         shallContinueToSearch(ethLog, targetNumber = 35UL)
-          .also { println("log=${ethLog.blockNumber} direction=$it") }
       }
     )
       .get()
       .also { log ->
-        assertThat(log).isNull()
+        assertThat(log).isNotNull()
+        assertThat(ULong.fromHexString(log!!.topics[1].encodeHex())).isEqualTo(35UL)
         assertThat(TestingJsonRpcServer.callCountByMethod("eth_getLogs")).isBetween(1, 11)
+        assertThat(logsEvaluated).hasSameElementsAs(logsEvaluated.toSet())
       }
   }
 

diff --git a/jvm-libs/linea/web3j-extensions/src/test/resources/log4j2.xml b/jvm-libs/linea/web3j-extensions/src/test/resources/log4j2.xml
diff --git a/...t/consensys/zkevm/ethereum/CommonUtils.kt → ...estFixtures/kotlin/linea/web3j/Pollers.kt b/...t/consensys/zkevm/ethereum/CommonUtils.kt → ...estFixtures/kotlin/linea/web3j/Pollers.kt
@@ -1,4 +1,4 @@
-package net.consensys.zkevm.ethereum
+package linea.web3j
 
 import org.web3j.protocol.Web3j
 import org.web3j.protocol.core.methods.response.TransactionReceipt

diff --git a/state-recover/appcore/logic/src/main/kotlin/linea/staterecover/StateRecoverApp.kt b/state-recover/appcore/logic/src/main/kotlin/linea/staterecover/StateRecoverApp.kt
@@ -142,11 +142,16 @@ class StateRecoverApp(
         } else {
           lineaContractClient.finalizedL2BlockNumber(blockParameter = config.l1LatestSearchBlock)
             .thenCompose { lastFinalizedBlockNumber ->
-              val stateRecoverStartBlockNumber = lastFinalizedBlockNumber + 1UL
+              val stateRecoverStartBlockNumber = when {
+                status.headBlockNumber >= lastFinalizedBlockNumber -> status.headBlockNumber + 1UL
+                else -> lastFinalizedBlockNumber + 1UL
+              }
               log.info(
-                "Starting enabling recovery mode: stateRecoverStartBlockNumber={} headBlockNumber={}",
+                "Starting enabling recovery mode: stateRecoverStartBlockNumber={} headBlockNumber={} " +
+                  "L1 lastFinalizedBlockNumber={}",
                 stateRecoverStartBlockNumber,
-                status.headBlockNumber
+                status.headBlockNumber,
+                lastFinalizedBlockNumber
               )
               elClient.lineaEnableStateRecovery(stateRecoverStartBlockNumber)
             }.thenApply { }