staterecover: adds blob scan and el client

Author: jpnovais

state-recover/appcore/clients-interfaces/build.gradle

@@ -0,0 +1,11 @@
+plugins {
+  id 'net.consensys.zkevm.kotlin-library-conventions'
+}
+
+group = 'build.linea.staterecover'
+
+dependencies {
+  api(project(':jvm-libs:generic:extensions:kotlin'))
+  api(project(':jvm-libs:linea:core:domain-models'))
+  api(project(':state-recover:appcore:domain-models'))
+}

state-recover/appcore/clients-interfaces/src/main/kotlin/build/linea/staterecover/clients/BlobFetcher.kt

@@ -0,0 +1,7 @@
+package build.linea.staterecover.clients
+
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+
+interface BlobFetcher {
+  fun fetchBlobsByHash(blobVersionedHashes: List<ByteArray>): SafeFuture<List<ByteArray>>
+}

state-recover/appcore/clients-interfaces/src/main/kotlin/build/linea/staterecover/clients/ExecutionLayerClient.kt

@@ -0,0 +1,12 @@
+package build.linea.staterecover.clients
+
+import build.linea.staterecover.BlockL1RecoveredData
+import net.consensys.linea.BlockNumberAndHash
+import net.consensys.linea.BlockParameter
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+
+interface ExecutionLayerClient {
+  fun getBlockNumberAndHash(blockParameter: BlockParameter): SafeFuture<BlockNumberAndHash>
+  fun lineaEngineImportBlocksFromBlob(blocks: List<BlockL1RecoveredData>): SafeFuture<Unit>
+  fun lineaEngineForkChoiceUpdated(headBlockHash: ByteArray, finalizedBlockHash: ByteArray): SafeFuture<Unit>
+}

state-recover/appcore/domain-models/build.gradle

@@ -0,0 +1,9 @@
+plugins {
+  id 'net.consensys.zkevm.kotlin-library-conventions'
+}
+
+group = 'build.linea.staterecover'
+
+dependencies {
+  api(project(':jvm-libs:generic:extensions:kotlin'))
+}

state-recover/appcore/domain-models/src/main/kotlin/build/linea/staterecover/BlockL1RecoveredData.kt

@@ -0,0 +1,78 @@
+package build.linea.staterecover
+
+import kotlinx.datetime.Instant
+import net.consensys.encodeHex
+
+data class BlockExtraData(
+  val beneficiary: ByteArray
+) {
+  override fun equals(other: Any?): Boolean {
+    if (this === other) return true
+    if (javaClass != other?.javaClass) return false
+
+    other as BlockExtraData
+
+    return beneficiary.contentEquals(other.beneficiary)
+  }
+
+  override fun hashCode(): Int {
+    return beneficiary.contentHashCode()
+  }
+
+  override fun toString(): String {
+    return "BlockExtraData(beneficiary=${beneficiary.encodeHex()})"
+  }
+}
+
+data class BlockL1RecoveredData(
+  val blockNumber: ULong,
+  val blockHash: ByteArray,
+  val coinbase: ByteArray,
+  val blockTimestamp: Instant,
+  val gasLimit: ULong,
+  val difficulty: ULong,
+  val extraData: BlockExtraData,
+  val transactions: List<TransactionL1RecoveredData>
+) {
+  override fun equals(other: Any?): Boolean {
+    if (this === other) return true
+    if (javaClass != other?.javaClass) return false
+
+    other as BlockL1RecoveredData
+
+    if (blockNumber != other.blockNumber) return false
+    if (!blockHash.contentEquals(other.blockHash)) return false
+    if (!coinbase.contentEquals(other.coinbase)) return false
+    if (blockTimestamp != other.blockTimestamp) return false
+    if (gasLimit != other.gasLimit) return false
+    if (difficulty != other.difficulty) return false
+    if (extraData != other.extraData) return false
+    if (transactions != other.transactions) return false
+
+    return true
+  }
+
+  override fun hashCode(): Int {
+    var result = blockNumber.hashCode()
+    result = 31 * result + blockHash.contentHashCode()
+    result = 31 * result + coinbase.contentHashCode()
+    result = 31 * result + blockTimestamp.hashCode()
+    result = 31 * result + gasLimit.hashCode()
+    result = 31 * result + difficulty.hashCode()
+    result = 31 * result + extraData.hashCode()
+    result = 31 * result + transactions.hashCode()
+    return result
+  }
+
+  override fun toString(): String {
+    return "BlockL1RecoveredData(" +
+      "blockNumber=$blockNumber, " +
+      "blockHash=${blockHash.encodeHex()}, " +
+      "coinbase=${coinbase.encodeHex()}, " +
+      "blockTimestamp=$blockTimestamp, " +
+      "gasLimit=$gasLimit, " +
+      "difficulty=$difficulty, " +
+      "extraData=$extraData, " +
+      "transactions=$transactions)"
+  }
+}

state-recover/appcore/domain-models/src/main/kotlin/build/linea/staterecover/TransactionL1RecoveredData.kt

@@ -0,0 +1,74 @@
+package build.linea.staterecover
+
+import java.math.BigInteger
+
+data class TransactionL1RecoveredData(
+  val type: UByte,
+  val nonce: ULong,
+  val maxPriorityFeePerGas: BigInteger,
+  val maxFeePerGas: BigInteger,
+  val gasLimit: ULong,
+  val from: ByteArray,
+  val to: ByteArray,
+  val value: BigInteger,
+  val data: ByteArray,
+  val accessList: List<AccessTuple>
+) {
+
+  data class AccessTuple(
+    val address: ByteArray,
+    val storageKeys: List<ByteArray>
+  ) {
+    override fun equals(other: Any?): Boolean {
+      if (this === other) return true
+      if (javaClass != other?.javaClass) return false
+
+      other as AccessTuple
+
+      if (!address.contentEquals(other.address)) return false
+      if (storageKeys != other.storageKeys) return false
+
+      return true
+    }
+
+    override fun hashCode(): Int {
+      var result = address.contentHashCode()
+      result = 31 * result + storageKeys.hashCode()
+      return result
+    }
+  }
+
+  override fun equals(other: Any?): Boolean {
+    if (this === other) return true
+    if (javaClass != other?.javaClass) return false
+
+    other as TransactionL1RecoveredData
+
+    if (type != other.type) return false
+    if (nonce != other.nonce) return false
+    if (maxPriorityFeePerGas != other.maxPriorityFeePerGas) return false
+    if (maxFeePerGas != other.maxFeePerGas) return false
+    if (gasLimit != other.gasLimit) return false
+    if (!from.contentEquals(other.from)) return false
+    if (!to.contentEquals(other.to)) return false
+    if (value != other.value) return false
+    if (!data.contentEquals(other.data)) return false
+    if (accessList != other.accessList) return false
+
+    return true
+  }
+
+  override fun hashCode(): Int {
+    var result = type.hashCode()
+    result = 31 * result + nonce.hashCode()
+    result = 31 * result + maxPriorityFeePerGas.hashCode()
+    result = 31 * result + maxFeePerGas.hashCode()
+    result = 31 * result + gasLimit.hashCode()
+    result = 31 * result + from.contentHashCode()
+    result = 31 * result + to.contentHashCode()
+    result = 31 * result + value.hashCode()
+    result = 31 * result + data.contentHashCode()
+    result = 31 * result + accessList.hashCode()
+    return result
+  }
+}

state-recover/clients/blobscan-client/build.gradle

@@ -0,0 +1,65 @@
+import org.gradle.api.tasks.testing.logging.TestExceptionFormat
+import org.gradle.api.tasks.testing.logging.TestLogEvent
+
+plugins {
+  id 'net.consensys.zkevm.kotlin-library-conventions'
+}
+
+group = 'build.linea.staterecover'
+
+dependencies {
+  implementation(project(':jvm-libs:generic:extensions:futures'))
+  implementation(project(':jvm-libs:generic:extensions:kotlin'))
+  implementation(project(':jvm-libs:generic:extensions:tuweni'))
+  implementation(project(':jvm-libs:generic:http-rest'))
+  implementation(project(':jvm-libs:generic:json-rpc'))
+  implementation(project(':jvm-libs:generic:vertx-helper'))
+  implementation(project(':jvm-libs:linea:clients:linea-state-manager'))
+  implementation(project(':jvm-libs:linea:core:domain-models'))
+  implementation(project(':jvm-libs:linea:core:long-running-service'))
+  implementation(project(':state-recover:appcore:clients-interfaces'))
+  implementation("io.vertx:vertx-web-client:${libs.versions.vertx}")
+
+  testImplementation "com.github.tomakehurst:wiremock-jre8:${libs.versions.wiremock.get()}"
+  testImplementation "org.slf4j:slf4j-api:1.7.30"
+  testImplementation "org.apache.logging.log4j:log4j-slf4j-impl:${libs.versions.log4j}"
+  testImplementation "org.apache.logging.log4j:log4j-core:${libs.versions.log4j}"
+}
+
+sourceSets {
+  integrationTest {
+    kotlin {
+      compileClasspath += sourceSets.main.output
+      runtimeClasspath += sourceSets.main.output
+    }
+    compileClasspath += sourceSets.main.output + sourceSets.main.compileClasspath + sourceSets.test.compileClasspath
+    runtimeClasspath += sourceSets.main.output + sourceSets.main.runtimeClasspath + sourceSets.test.runtimeClasspath
+  }
+}
+
+task integrationTest(type: Test) {
+  test ->
+    description = "Runs integration tests."
+    group = "verification"
+    useJUnitPlatform()
+
+    classpath = sourceSets.integrationTest.runtimeClasspath
+    testClassesDirs = sourceSets.integrationTest.output.classesDirs
+
+    dependsOn(":localStackComposeUp")
+    dependsOn(rootProject.tasks.compileContracts)
+
+    testLogging {
+      events TestLogEvent.FAILED,
+        TestLogEvent.SKIPPED,
+        TestLogEvent.STANDARD_ERROR,
+        TestLogEvent.STARTED,
+        TestLogEvent.PASSED
+      exceptionFormat TestExceptionFormat.FULL
+      showCauses true
+      showExceptions true
+      showStackTraces true
+      // set showStandardStreams if you need to see test logs
+      showStandardStreams false
+    }
+}

state-recover/clients/blobscan-client/src/main/kotlin/build/linea/staterecover/clients/blobscan/BlobScanClient.kt

@@ -0,0 +1,56 @@
+package build.linea.staterecover.clients.blobscan
+
+import build.linea.staterecover.clients.BlobFetcher
+import io.vertx.core.Vertx
+import io.vertx.core.json.JsonObject
+import io.vertx.ext.web.client.WebClient
+import io.vertx.ext.web.client.WebClientOptions
+import net.consensys.decodeHex
+import net.consensys.encodeHex
+import net.consensys.linea.jsonrpc.client.RequestRetryConfig
+import net.consensys.linea.vertx.setDefaultsFrom
+import org.apache.logging.log4j.LogManager
+import org.apache.logging.log4j.Logger
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+import java.net.URI
+
+class BlobScanClient(
+  private val restClient: RestClient<JsonObject>,
+  private val log: Logger = LogManager.getLogger(BlobScanClient::class.java)
+) : BlobFetcher {
+  fun getBlobById(id: String): SafeFuture<ByteArray> {
+    return restClient
+      .get("/blobs/$id")
+      .thenApply { response ->
+        if (response.statusCode == 200) {
+          response.body!!.getString("data").decodeHex()
+        } else {
+          throw RuntimeException(
+            "error fetching blobId=$id " +
+              "errorMessage=${response.body?.getString("message") ?: ""}"
+          )
+        }
+      }
+  }
+
+  override fun fetchBlobsByHash(blobVersionedHashes: List<ByteArray>): SafeFuture<List<ByteArray>> {
+    return SafeFuture.collectAll(blobVersionedHashes.map { hash -> getBlobById(hash.encodeHex()) }.stream())
+  }
+
+  companion object {
+    fun create(
+      vertx: Vertx,
+      endpoint: URI,
+      requestRetryConfig: RequestRetryConfig
+    ): BlobScanClient {
+      val restClient = VertxRestClient(
+        vertx = vertx,
+        webClient = WebClient.create(vertx, WebClientOptions().setDefaultsFrom(endpoint)),
+        responseParser = { it.toJsonObject() },
+        retryableErrorCodes = setOf(429, 503, 504),
+        requestRetryConfig = requestRetryConfig
+      )
+      return BlobScanClient(restClient)
+    }
+  }
+}

state-recover/clients/blobscan-client/src/main/kotlin/build/linea/staterecover/clients/blobscan/VertxRestClient.kt

@@ -0,0 +1,66 @@
+package build.linea.staterecover.clients.blobscan
+
+import io.vertx.core.Vertx
+import io.vertx.core.buffer.Buffer
+import io.vertx.ext.web.client.HttpRequest
+import io.vertx.ext.web.client.HttpResponse
+import io.vertx.ext.web.client.WebClient
+import net.consensys.linea.async.AsyncRetryer
+import net.consensys.linea.async.toSafeFuture
+import net.consensys.linea.jsonrpc.client.RequestRetryConfig
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+
+// TODO: move to a common module
+data class RestResponse<T>(
+  val statusCode: Int,
+  val body: T?
+)
+
+interface RestClient<Response> {
+  fun get(path: String): SafeFuture<RestResponse<Response>>
+  // add remaining verbs as we need them
+}
+
+class VertxRestClient<Response>(
+  private val vertx: Vertx,
+  private val webClient: WebClient,
+  private val responseParser: (Buffer) -> Response,
+  private val retryableErrorCodes: Set<Int> = DEFAULT_RETRY_HTTP_CODES,
+  private val requestRetryConfig: RequestRetryConfig,
+  private val asyncRetryer: AsyncRetryer<HttpResponse<Buffer>> = AsyncRetryer.retryer(
+    backoffDelay = requestRetryConfig.backoffDelay,
+    maxRetries = requestRetryConfig.maxRetries?.toInt(),
+    timeout = requestRetryConfig.timeout,
+    vertx = vertx
+  ),
+  private val requestHeaders: Map<String, String> = mapOf("Accept" to "application/json")
+) : RestClient<Response> {
+  private fun makeRequestWithRetry(
+    request: HttpRequest<Buffer>
+  ): SafeFuture<HttpResponse<Buffer>> {
+    return asyncRetryer
+      .retry(
+        stopRetriesPredicate = { response: HttpResponse<Buffer> ->
+          response.statusCode() !in retryableErrorCodes
+        }
+      ) {
+        request.send().toSafeFuture()
+      }
+  }
+
+  override fun get(path: String): SafeFuture<RestResponse<Response>> {
+    return makeRequestWithRetry(
+      webClient
+        .get(path)
+        .apply { requestHeaders.forEach(::putHeader) }
+    )
+      .thenApply { response ->
+        val parsedResponse = response.body()?.let(responseParser)
+        RestResponse(response.statusCode(), parsedResponse)
+      }
+  }
+
+  companion object {
+    val DEFAULT_RETRY_HTTP_CODES = setOf(429, 500, 503, 504)
+  }
+}