Merge branch 'main' into chore/498-v5-techdebt-cleanup

Makefile

@@ -228,10 +228,10 @@ fresh-start-staterecovery-for-replay-only:
 	make start-whole-environment-traces-v2 COMPOSE_PROFILES=$(COMPOSE_PROFILES)
 
 staterecovery-replay-from-block: L1_ROLLUP_CONTRACT_ADDRESS:=0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9
-staterecovery-replay-from-block: PLUGIN_STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER:=1
+staterecovery-replay-from-block: STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER:=1
 staterecovery-replay-from-block:
 	docker compose -f docker/compose.yml down zkbesu-shomei-sr shomei-sr
-	L1_ROLLUP_CONTRACT_ADDRESS=$(L1_ROLLUP_CONTRACT_ADDRESS) PLUGIN_STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER=$(PLUGIN_STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER) docker compose -f docker/compose.yml up zkbesu-shomei-sr shomei-sr -d
+	L1_ROLLUP_CONTRACT_ADDRESS=$(L1_ROLLUP_CONTRACT_ADDRESS) STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER=$(STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER) docker compose -f docker/compose.yml up zkbesu-shomei-sr shomei-sr -d
 
 testnet-start-l2:
 		docker compose -f docker/compose.yml -f docker/compose-testnet-sync.overrides.yml --profile l2 up -d

buildSrc/src/main/groovy/net.consensys.zkevm.kotlin-common-minimal-conventions.gradle

@@ -42,15 +42,13 @@ repositories {
 
 dependencies {
   //<testing>
-  testImplementation platform("org.junit:junit-bom:5.10.1")
-  testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.10.1'
-  testImplementation 'org.junit.jupiter:junit-jupiter:5.10.1'
-  testImplementation 'org.assertj:assertj-core:3.24.2'
-  testImplementation 'org.mockito:mockito-core:5.7.0'
-  // TODO: org.mockito.kotlin:mockito-kotlin uses org.mockito:mockito-core:4.5.1,
-  // check later on if this may raise incompatibilities
-  testImplementation 'org.mockito.kotlin:mockito-kotlin:5.4.0'
-  testImplementation 'org.awaitility:awaitility:4.2.0'
+  testImplementation platform("org.junit:junit-bom:${libs.versions.junit.get()}")
+  testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine:${libs.versions.junit.get()}"
+  testImplementation "org.junit.jupiter:junit-jupiter:${libs.versions.junit.get()}"
+  testImplementation "org.assertj:assertj-core:${libs.versions.assertj.get()}"
+  testImplementation "org.mockito:mockito-core:${libs.versions.mockito.get()}"
+  testImplementation "org.mockito.kotlin:mockito-kotlin:${libs.versions.mockitoKotlin.get()}"
+  testImplementation "org.awaitility:awaitility:${libs.versions.awaitility.get()}"
   //</testing>
 }
 

docker/compose.yml

@@ -726,7 +726,7 @@ services:
     environment:
       LOG4J_CONFIGURATION_FILE: /var/lib/besu/log4j.xml
       L1_ROLLUP_CONTRACT_ADDRESS: ${L1_ROLLUP_CONTRACT_ADDRESS:-0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9}
-      PLUGIN_STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER: ${PLUGIN_STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER:-1}
+      STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER: ${STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER:-1}
     entrypoint:
       - /bin/bash
       - -c
@@ -746,7 +746,7 @@ services:
         --plugin-staterecovery-shomei-endpoint=http://shomei-sr:8888 \
         --plugin-staterecovery-blobscan-endpoint=http://blobscan-api:4001 \
         --plugin-staterecovery-linea-sequencer-beneficiary-address=0x6d976c9b8ceee705d4fe8699b44e5eb58242f484 \
-        --plugin-staterecovery-l1-polling-interval=PT0.1S \
+        --plugin-staterecovery-l1-polling-interval=PT0.01S \
         --bootnodes=enode://14408801a444dafc44afbccce2eb755f902aed3b5743fed787b3c790e021fef28b8c827ed896aa4e8fb46e22bd67c39f994a73768b4b382f8597b0d44370e15d@11.11.11.101:30303
     volumes:
       - ./config/zkbesu-shomei/zkbesu-config.toml:/var/lib/besu/zkbesu-config.toml:ro

docker/config/l1-node/cl/log4j.xml

@@ -6,7 +6,7 @@
     </Console>
   </Appenders>
   <Loggers>
-    <Logger name="tech.pegasys.teku.ethereum.executionclient" level="TRACE" additivity="false">
+    <Logger name="tech.pegasys.teku.ethereum.executionclient" level="INFO" additivity="false">
       <AppenderRef ref="Console"/>
     </Logger>
     <Logger name="io.opentelemetry" level="WARN" additivity="false">

gradle/libs.versions.toml

@@ -8,27 +8,34 @@ jreleaser = {id = "org.jreleaser", version = "1.15.0"}
 jreleaser = { group = "org.jreleaser", name = "jreleaser-gradle-plugin", version = "1.15.0" }
 
 [versions]
+# Testing
+assertj = "3.27.3"
+awaitility = "4.2.0"
+jsonUnit = "3.4.1"
+junit = "5.10.1"
+mockito = "5.7.0"
+mockitoKotlin = "5.4.0"
+restassured = "5.3.0"
+wiremock = "3.0.1"
+
+# Runtime
 besu = "24.12.2"
+blobCompressor = "0.0.4"
+blobShnarfCalculator = "0.0.4"
 bouncycastle = "1.79"
 caffeine = "3.1.6"
 hoplite = "2.7.5"
 jackson = "2.18.0"
 jna = "5.14.0"
-junit = "5.10.1"
+kotlinResult = "1.1.16"
 kotlinxDatetime = "0.6.1"
 ktlint = "0.47.0"
 log4j = "2.20.0"
 micrometer = "1.8.4"
 netty = "4.1.92.Final"
 picoli = "4.7.6"
-restassured = "5.3.0"
+slf4j="1.7.36"
 teku = "23.1.1"
 tuweni = "2.4.2"
-kotlinResult = "1.1.16"
 vertx = "4.5.11"
 web3j = "4.12.2"
-wiremock = "3.0.1"
-jsonUnit = "3.4.1"
-slf4j="1.7.36"
-blobCompressor = "0.0.4"
-blobShnarfCalculator = "0.0.4"

jvm-libs/generic/extensions/futures/src/test/kotlin/net/consensys/linea/async/AsyncRetryerTest.kt

@@ -240,7 +240,7 @@ class AsyncRetryerTest {
 
   @Test
   fun `retry should stop after timeout is elapsed - promise rejected`(vertx: Vertx) {
-    var callCount = AtomicInteger(0)
+    val callCount = AtomicInteger(0)
     val future =
       AsyncRetryer.retry(vertx, 20.milliseconds, timeout = 40.milliseconds) {
         SafeFuture.failedFuture<String>(RuntimeException("Failure number ${callCount.incrementAndGet()}"))
@@ -255,7 +255,7 @@ class AsyncRetryerTest {
       .withMessageContaining("Failure number")
 
     assertThat(callCount.get())
-      .isGreaterThan(1)
+      .isGreaterThanOrEqualTo(1)
       .isLessThanOrEqualTo(3)
   }
 
@@ -296,7 +296,7 @@ class AsyncRetryerTest {
         SafeFuture.failedFuture<Unit>(Exception("Upss"))
       }
     assertThrows<ExecutionException> { result.get() }
-    assertThat(delays.first()).isGreaterThan(100).isLessThan(300)
+    assertThat(delays.first()).isGreaterThan(100)
     assertThat(delays.subList(1, delays.size)).allMatch { it >= 300 }
     assertThat(callCount).isEqualTo(4)
   }

jvm-libs/linea/testing/l1-blob-and-proof-submission/build.gradle

@@ -12,4 +12,5 @@ dependencies {
   implementation(testFixtures(project(':jvm-libs:linea:web3j-extensions')))
   implementation(project(':coordinator:clients:prover-client:serialization'))
   implementation(testFixtures(project(":coordinator:core")))
+  implementation("org.assertj:assertj-core:${libs.versions.assertj.get()}")
 }

jvm-libs/linea/testing/l1-blob-and-proof-submission/src/main/kotlin/net/consensys/linea/testing/submission/SubmissionTestHelper.kt

@@ -1,50 +1,108 @@
 package net.consensys.linea.testing.submission
 
+import build.linea.domain.BlockInterval
 import linea.web3j.waitForTxReceipt
 import net.consensys.zkevm.coordinator.clients.smartcontract.LineaRollupSmartContractClient
 import net.consensys.zkevm.domain.Aggregation
+import net.consensys.zkevm.domain.BlobRecord
+import org.apache.logging.log4j.LogManager
+import org.apache.logging.log4j.Logger
+import org.assertj.core.api.Assertions.assertThat
 import org.web3j.protocol.Web3j
+import tech.pegasys.teku.infrastructure.async.SafeFuture
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.minutes
 
+fun assertTxSuccess(
+  txHash: String,
+  interval: BlockInterval,
+  submissionType: String,
+  l1Web3jClient: Web3j,
+  timeout: Duration = 1.minutes
+) {
+  l1Web3jClient.waitForTxReceipt(
+    txHash = txHash,
+    timeout = timeout
+  ).also { txReceipt ->
+    assertThat(txReceipt.status)
+      .isEqualTo("0x1")
+      .withFailMessage(
+        "submission of $submissionType=${interval.intervalString()}" +
+          " failed on L1. receipt=$txReceipt"
+      )
+  }
+}
+
+fun assertTxsSuccess(
+  txsAndInterval: List<Pair<String, BlockInterval>>,
+  submissionType: String,
+  l1Web3jClient: Web3j,
+  timeout: Duration = 1.minutes
+) {
+  SafeFuture.supplyAsync {
+    txsAndInterval.forEach { (txHash, interval) ->
+      assertTxSuccess(txHash, interval, submissionType, l1Web3jClient, timeout)
+    }
+  }
+    .get(timeout.inWholeMilliseconds, java.util.concurrent.TimeUnit.MILLISECONDS)
+}
+
 /**
  * Submits blobs respecting aggregation boundaries
  * returns list of tx hashes, does not wait for txs to be mined
  */
 fun submitBlobs(
   contractClient: LineaRollupSmartContractClient,
   aggregationsAndBlobs: List<AggregationAndBlobs>,
-  blobChunksSize: Int = 6
-): List<String> {
+  blobChunksSize: Int = 6,
+  log: Logger
+): List<Pair<String, List<BlobRecord>>> {
   require(blobChunksSize in 1..6) { "blobChunksSize must be between 1..6" }
 
   return aggregationsAndBlobs
-    .map { (_, aggBlobs) ->
+    .map { (agg, aggBlobs) ->
       val blobChunks = aggBlobs.chunked(blobChunksSize)
-      blobChunks.map { blobs -> contractClient.submitBlobs(blobs, gasPriceCaps = null).get() }
+      blobChunks.map { blobs ->
+        val txHash = contractClient.submitBlobs(blobs, gasPriceCaps = null).get()
+        log.info(
+          "submitting blobs: aggregation={} blobsChunk={} txHash={}",
+          agg?.intervalString(),
+          blobs.map { it.intervalString() },
+          txHash
+        )
+
+        txHash to blobs
+      }
     }
     .flatten()
 }
 
-/**
- * Submits blobs respecting aggregation boundaries,
- * then submits aggregations
- *
- * returns list of tx hashes of aggregations submissions only, does not wait for txs to be mined
- */
-data class SubmissionTxHashes(
-  val blobTxHashes: List<String>,
-  val aggregationTxHashes: List<String>
-)
-
-fun submitBlobsAndAggregations(
+fun submitBlobsAndAggregationsAndWaitExecution(
   contractClientForBlobSubmission: LineaRollupSmartContractClient,
   contractClientForAggregationSubmission: LineaRollupSmartContractClient = contractClientForBlobSubmission,
   aggregationsAndBlobs: List<AggregationAndBlobs>,
-  blobChunksSize: Int = 6
-): SubmissionTxHashes {
-  val blobSubmissionTxHashes = submitBlobs(contractClientForBlobSubmission, aggregationsAndBlobs, blobChunksSize)
-  return aggregationsAndBlobs
+  blobChunksMaxSize: Int = 6,
+  l1Web3jClient: Web3j,
+  waitTimeout: Duration = 2.minutes,
+  log: Logger = LogManager.getLogger("linea.testing.submission")
+) {
+  val blobSubmissionTxHashes = submitBlobs(
+    contractClientForBlobSubmission,
+    aggregationsAndBlobs,
+    blobChunksMaxSize,
+    log
+  )
+
+  assertTxsSuccess(
+    txsAndInterval = blobSubmissionTxHashes.map { (txHash, blobs) ->
+      txHash to BlockInterval(blobs.first().startBlockNumber, blobs.last().endBlockNumber)
+    },
+    submissionType = "blobs",
+    l1Web3jClient = l1Web3jClient,
+    timeout = waitTimeout
+  )
+
+  val submissions = aggregationsAndBlobs
     .filter { it.aggregation != null }
     .mapIndexed { index, (aggregation, aggBlobs) ->
       aggregation as Aggregation
@@ -56,35 +114,13 @@ fun submitBlobsAndAggregations(
         parentL1RollingHash = parentAgg?.aggregationProof?.l1RollingHash ?: ByteArray(32),
         parentL1RollingHashMessageNumber = parentAgg?.aggregationProof?.l1RollingHashMessageNumber ?: 0L,
         gasPriceCaps = null
-      ).get()
+      ).get() to aggregation
     }
-    .let { SubmissionTxHashes(blobSubmissionTxHashes, it) }
-}
-
-fun submitBlobsAndAggregationsAndWaitExecution(
-  contractClientForBlobSubmission: LineaRollupSmartContractClient,
-  contractClientForAggregationSubmission: LineaRollupSmartContractClient = contractClientForBlobSubmission,
-  aggregationsAndBlobs: List<AggregationAndBlobs>,
-  blobChunksSize: Int = 6,
-  l1Web3jClient: Web3j,
-  waitTimeout: Duration = 2.minutes
-) {
-  val submissionTxHashes = submitBlobsAndAggregations(
-    contractClientForBlobSubmission = contractClientForAggregationSubmission,
-    contractClientForAggregationSubmission = contractClientForAggregationSubmission,
-    aggregationsAndBlobs = aggregationsAndBlobs,
-    blobChunksSize = blobChunksSize
-  )
 
-  l1Web3jClient.waitForTxReceipt(
-    txHash = submissionTxHashes.aggregationTxHashes.last(),
+  assertTxsSuccess(
+    txsAndInterval = submissions,
+    submissionType = "aggregation",
+    l1Web3jClient = l1Web3jClient,
     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"
-      )
-    }
-  }
+  )
 }

state-recovery/besu-plugin/src/main/kotlin/linea/staterecovery/plugin/LineaStateRecoveryPlugin.kt

@@ -113,6 +113,9 @@ open class LineaStateRecoveryPlugin : BesuPlugin {
   }
 
   override fun afterExternalServicePostMainLoop() {
+    // we need to recall this again because Sync and Mining services
+    // may have been started after the plugin start
+    this.recoveryModeManager.enableRecoveryModeIfNecessary()
   }
 
   override fun stop() {

state-recovery/besu-plugin/src/main/kotlin/linea/staterecovery/plugin/PluginOptions.kt

@@ -79,7 +79,7 @@ class PluginCliOptions {
       "Tries to force the recovery start block number to the given value. " +
         "This is mean for testing purposes, not production. Must be greater than or equal to 1."
     ],
-    defaultValue = "\${env:PLUGIN_STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER}",
+    defaultValue = "\${env:STATERECOVERY_OVERRIDE_START_BLOCK_NUMBER}",
     required = false
   )
   var overridingRecoveryStartBlockNumber: Long? = null

state-recovery/besu-plugin/src/main/kotlin/linea/staterecovery/plugin/RecoveryModeManager.kt

@@ -27,6 +27,11 @@ class RecoveryModeManager(
 
   init {
     log.info("RecoveryModeManager initializing: headBlockNumber={}", headBlockNumber)
+    enableRecoveryModeIfNecessary()
+  }
+
+  @Synchronized
+  fun enableRecoveryModeIfNecessary() {
     if (hasReachedTargetBlock()) {
       log.info(
         "enabling recovery mode immediately at blockNumber={} recoveryTargetBlockNumber={}",
@@ -92,9 +97,6 @@ class RecoveryModeManager(
 
   /** Switches the node to recovery mode.  */
   private fun switchToRecoveryMode() {
-    check(!recoveryModeTriggered.get()) {
-      "cannot enable already enabled recovery mode"
-    }
     log.warn("Stopping synchronization service")
     synchronizationService.stop()
 
@@ -105,7 +107,7 @@ class RecoveryModeManager(
     miningService.stop()
 
     log.info(
-      "Switched to state recovery mode at block={}",
+      "switched to state recovery mode at block={}",
       headBlockNumber
     )
     recoveryModeTriggered.set(true)

state-recovery/test-cases/build.gradle

@@ -64,6 +64,8 @@ test {
 }
 
 task integrationTest(type: Test) { test ->
+  outputs.cacheIf { false }
+  outputs.upToDateWhen { false }
   systemProperty "vertx.parameter.filename", project.projectDir.toPath()
       .resolve("src/test/resources/vertx-options.json")
       .toAbsolutePath().toString()
@@ -94,6 +96,8 @@ task integrationTest(type: Test) { test ->
 
 task e2eTest(type: Test) { test ->
   outputs.cacheIf { false }
+  outputs.upToDateWhen { false }
+
   systemProperty "vertx.parameter.filename", project.projectDir.toPath()
       .resolve("src/test/resources/vertx-options.json")
       .toAbsolutePath().toString()