Coordinator: Shared State Manager Client (#233)

* add state-manager client (C&P) from coordinator one for now

jvm-libs/linea/blob-decompressor/build.gradle

@@ -51,7 +51,6 @@ task cleanResources(type: Delete) {
     .filter {
       it.name.endsWith(".so") || it.name.endsWith(".dll") || it.name.endsWith(".dylib")
     }.each {
-      println("Deleting: ${it}")
       delete it
     }
 }

jvm-libs/linea/clients/linea-state-manager/build.gradle

@@ -0,0 +1,22 @@
+plugins {
+  id 'net.consensys.zkevm.kotlin-library-conventions'
+}
+
+dependencies {
+  api project(':jvm-libs:linea:core:domain-models')
+  api project(':jvm-libs:linea:core:metrics')
+  api project(':jvm-libs:linea:core:client-interface')
+  api project(':jvm-libs:generic:json-rpc')
+  api project(':jvm-libs:generic:errors')
+  api project(':jvm-libs:generic:extensions:futures')
+  api project(':jvm-libs:generic:extensions:kotlin')
+  api "io.tmio:tuweni-bytes:${libs.versions.tuweni.get()}"
+
+  implementation "com.fasterxml.jackson.core:jackson-annotations:${libs.versions.jackson.get()}"
+  implementation "com.fasterxml.jackson.core:jackson-databind:${libs.versions.jackson.get()}"
+  implementation "com.fasterxml.jackson.module:jackson-module-kotlin:${libs.versions.jackson.get()}"
+
+  testImplementation(project(":jvm-libs:linea:testing:file-system"))
+  testImplementation "io.vertx:vertx-junit5"
+  testImplementation "com.github.tomakehurst:wiremock-jre8:${libs.versions.wiremock.get()}"
+}

jvm-libs/linea/clients/linea-state-manager/src/main/kotlin/build/linea/clients/StateManagerClientV1.kt

@@ -0,0 +1,87 @@
+package build.linea.clients
+
+import build.linea.domain.BlockInterval
+import com.fasterxml.jackson.databind.node.ArrayNode
+import com.github.michaelbull.result.Result
+import net.consensys.encodeHex
+import net.consensys.linea.errors.ErrorResponse
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+
+enum class StateManagerErrorType : ClientError {
+  UNKNOWN,
+  UNSUPPORTED_VERSION,
+  BLOCK_MISSING_IN_CHAIN
+}
+
+sealed interface StateManagerRequest
+sealed class GetChainHeadRequest() : StateManagerRequest
+data class GetStateMerkleProofRequest(
+  val blockInterval: BlockInterval
+) : StateManagerRequest, BlockInterval by blockInterval
+
+sealed interface StateManagerResponse
+
+data class GetZkEVMStateMerkleProofResponse(
+  val zkStateMerkleProof: ArrayNode,
+  val zkParentStateRootHash: ByteArray,
+  val zkEndStateRootHash: ByteArray,
+  val zkStateManagerVersion: String
+) : StateManagerResponse {
+  override fun equals(other: Any?): Boolean {
+    if (this === other) return true
+    if (javaClass != other?.javaClass) return false
+
+    other as GetZkEVMStateMerkleProofResponse
+
+    if (zkStateMerkleProof != other.zkStateMerkleProof) return false
+    if (!zkParentStateRootHash.contentEquals(other.zkParentStateRootHash)) return false
+    if (!zkEndStateRootHash.contentEquals(other.zkEndStateRootHash)) return false
+    if (zkStateManagerVersion != other.zkStateManagerVersion) return false
+
+    return true
+  }
+
+  override fun hashCode(): Int {
+    var result = zkStateMerkleProof.hashCode()
+    result = 31 * result + zkParentStateRootHash.contentHashCode()
+    result = 31 * result + zkEndStateRootHash.contentHashCode()
+    result = 31 * result + zkStateManagerVersion.hashCode()
+    return result
+  }
+
+  override fun toString(): String {
+    return "GetZkEVMStateMerkleProofResponse(" +
+      "zkStateMerkleProof=$zkStateMerkleProof, zkParentStateRootHash=${zkParentStateRootHash.encodeHex()}, " +
+      "zkEndStateRootHash=${zkEndStateRootHash.encodeHex()}, " +
+      "zkStateManagerVersion='$zkStateManagerVersion')"
+  }
+}
+
+// Type alias dedicated for each method
+typealias StateManagerClientToGetStateMerkleProofV0 =
+  AsyncClient<GetStateMerkleProofRequest, GetZkEVMStateMerkleProofResponse>
+
+typealias StateManagerClientToGetChainHeadV1 =
+  AsyncClient<GetChainHeadRequest, ULong>
+
+interface StateManagerClientV1 {
+  /**
+   * Get the head block number of the chain.
+   * @return GetZkEVMStateMerkleProofResponse
+   * @throws ClientException with errorType StateManagerErrorType when know error occurs
+   */
+  fun rollupGetStateMerkleProof(
+    blockInterval: BlockInterval
+  ): SafeFuture<GetZkEVMStateMerkleProofResponse> = rollupGetStateMerkleProofWithTypedError(blockInterval)
+    .unwrapResultMonad()
+
+  /**
+   * This is for backward compatibility with the old version in the coordinator side.
+   * This error typing is not really usefull anymore
+   */
+  fun rollupGetStateMerkleProofWithTypedError(
+    blockInterval: BlockInterval
+  ): SafeFuture<Result<GetZkEVMStateMerkleProofResponse, ErrorResponse<StateManagerErrorType>>>
+
+  fun rollupGetHeadBlockNumber(): SafeFuture<ULong>
+}

jvm-libs/linea/clients/linea-state-manager/src/main/kotlin/build/linea/clients/StateManagerV1JsonRpcClient.kt

@@ -0,0 +1,109 @@
+package build.linea.clients
+
+import build.linea.domain.BlockInterval
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.node.ArrayNode
+import com.github.michaelbull.result.Err
+import com.github.michaelbull.result.Ok
+import com.github.michaelbull.result.Result
+import io.vertx.core.json.JsonObject
+import net.consensys.decodeHex
+import net.consensys.fromHexString
+import net.consensys.linea.errors.ErrorResponse
+import net.consensys.linea.jsonrpc.JsonRpcErrorResponseException
+import net.consensys.linea.jsonrpc.client.JsonRpcV2Client
+import org.apache.logging.log4j.LogManager
+import org.apache.logging.log4j.Logger
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+
+class StateManagerV1JsonRpcClient(
+  private val rpcClient: JsonRpcV2Client,
+  private val zkStateManagerVersion: String
+) : StateManagerClientV1 {
+  private val log: Logger = LogManager.getLogger(this::class.java)
+
+  override fun rollupGetHeadBlockNumber(): SafeFuture<ULong> {
+    return rpcClient
+      .makeRequest(
+        method = "rollup_getZkEVMBlockNumber",
+        params = emptyList<Unit>(),
+        shallRetryRequestPredicate = { it is Err },
+        resultMapper = { ULong.fromHexString(it as String) }
+      )
+  }
+
+  override fun rollupGetStateMerkleProof(blockInterval: BlockInterval): SafeFuture<GetZkEVMStateMerkleProofResponse> {
+    val params = listOf(
+      JsonObject.of(
+        "startBlockNumber",
+        blockInterval.startBlockNumber.toLong(),
+        "endBlockNumber",
+        blockInterval.endBlockNumber.toLong(),
+        "zkStateManagerVersion",
+        zkStateManagerVersion
+      )
+    )
+
+    return rpcClient
+      .makeRequest(
+        method = "rollup_getZkEVMStateMerkleProofV0",
+        params = params,
+        shallRetryRequestPredicate = { it is Err },
+        resultMapper = { it as JsonNode; parseZkEVMStateMerkleProofResponse(it) }
+      )
+  }
+
+  override fun rollupGetStateMerkleProofWithTypedError(
+    blockInterval: BlockInterval
+  ): SafeFuture<Result<GetZkEVMStateMerkleProofResponse, ErrorResponse<StateManagerErrorType>>> {
+    return rollupGetStateMerkleProof(blockInterval)
+      .handleComposed { result, th ->
+        if (th != null) {
+          if (th is JsonRpcErrorResponseException) {
+            SafeFuture.completedFuture(Err(mapErrorResponse(th)))
+          } else {
+            SafeFuture.failedFuture(th)
+          }
+        } else {
+          SafeFuture.completedFuture(Ok(result))
+        }
+      }
+  }
+
+  private fun mapErrorResponse(
+    jsonRpcErrorResponse: JsonRpcErrorResponseException
+  ): ErrorResponse<StateManagerErrorType> {
+    val errorType =
+      try {
+        StateManagerErrorType.valueOf(
+          jsonRpcErrorResponse.rpcErrorMessage.substringBefore('-').trim()
+        )
+      } catch (_: Exception) {
+        log.error(
+          "State manager found unrecognised JSON-RPC response error: {}",
+          jsonRpcErrorResponse.rpcErrorMessage
+        )
+        StateManagerErrorType.UNKNOWN
+      }
+
+    return ErrorResponse(
+      errorType,
+      listOfNotNull(
+        jsonRpcErrorResponse.rpcErrorMessage,
+        jsonRpcErrorResponse.rpcErrorData?.toString()
+      )
+        .joinToString(": ")
+    )
+  }
+
+  private fun parseZkEVMStateMerkleProofResponse(
+    result: JsonNode
+  ): GetZkEVMStateMerkleProofResponse {
+    return GetZkEVMStateMerkleProofResponse(
+      zkStateManagerVersion = result.get("zkStateManagerVersion").asText(),
+      zkStateMerkleProof = result.get("zkStateMerkleProof") as ArrayNode,
+      zkParentStateRootHash = result.get("zkParentStateRootHash").asText().decodeHex(),
+      zkEndStateRootHash = result.get("zkEndStateRootHash").asText().decodeHex()
+    )
+  }
+}

jvm-libs/linea/clients/linea-state-manager/src/test/kotlin/build/linea/clients/StateManagerV1JsonRpcClientTest.kt

@@ -0,0 +1,215 @@
+package build.linea.clients
+
+import build.linea.domain.BlockInterval
+import com.fasterxml.jackson.databind.node.ArrayNode
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.github.michaelbull.result.Err
+import com.github.michaelbull.result.Ok
+import com.github.tomakehurst.wiremock.WireMockServer
+import com.github.tomakehurst.wiremock.client.WireMock.containing
+import com.github.tomakehurst.wiremock.client.WireMock.ok
+import com.github.tomakehurst.wiremock.client.WireMock.post
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration.options
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry
+import io.vertx.core.Vertx
+import io.vertx.junit5.VertxExtension
+import net.consensys.decodeHex
+import net.consensys.fromHexString
+import net.consensys.linea.async.get
+import net.consensys.linea.errors.ErrorResponse
+import net.consensys.linea.jsonrpc.client.RequestRetryConfig
+import net.consensys.linea.jsonrpc.client.VertxHttpJsonRpcClientFactory
+import net.consensys.linea.testing.filesystem.findPathTo
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import java.net.URI
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.toJavaDuration
+
+@ExtendWith(VertxExtension::class)
+class StateManagerV1JsonRpcClientTest {
+  private lateinit var wiremock: WireMockServer
+  private lateinit var stateManagerClient: StateManagerV1JsonRpcClient
+  private lateinit var meterRegistry: SimpleMeterRegistry
+
+  private fun wiremockStubForPost(response: String) {
+    wiremock.stubFor(
+      post("/")
+        .withHeader("Content-Type", containing("application/json"))
+        .willReturn(
+          ok()
+            .withHeader("Content-type", "application/json")
+            .withBody(response.toByteArray())
+        )
+    )
+  }
+
+  @BeforeEach
+  fun setup(vertx: Vertx) {
+    wiremock = WireMockServer(options().dynamicPort())
+    wiremock.start()
+    meterRegistry = SimpleMeterRegistry()
+    val rpcClientFactory = VertxHttpJsonRpcClientFactory(vertx, meterRegistry)
+    val vertxHttpJsonRpcClient = rpcClientFactory.createV2(
+      endpoints = setOf(URI("http://127.0.0.1:" + wiremock.port()).toURL()),
+      retryConfig = RequestRetryConfig(
+        maxRetries = 2u,
+        timeout = 2.seconds,
+        10.milliseconds,
+        1u
+      )
+    )
+
+    stateManagerClient =
+      StateManagerV1JsonRpcClient(
+        rpcClient = vertxHttpJsonRpcClient,
+        zkStateManagerVersion = "0.1.2"
+      )
+  }
+
+  @AfterEach
+  fun tearDown(vertx: Vertx) {
+    val vertxStopFuture = vertx.close()
+    wiremock.stop()
+    vertxStopFuture.get()
+  }
+
+  @Test
+  fun getZkEVMStateMerkleProof_success() {
+    val testFilePath = findPathTo("testdata")!!.resolve("type2state-manager/state-proof.json")
+    val json = jacksonObjectMapper().readTree(testFilePath.toFile())
+    val zkStateManagerVersion = json.get("zkStateManagerVersion").asText()
+    val zkStateMerkleProof = json.get("zkStateMerkleProof") as ArrayNode
+    val zkParentStateRootHash = json.get("zkParentStateRootHash").asText()
+    val zkEndStateRootHash = json.get("zkEndStateRootHash").asText()
+
+    wiremockStubForPost(
+      """
+      {
+        "jsonrpc":"2.0",
+        "id":"1",
+        "result": {
+          "zkParentStateRootHash": "$zkParentStateRootHash",
+          "zkEndStateRootHash": "$zkEndStateRootHash",
+          "zkStateMerkleProof": $zkStateMerkleProof,
+          "zkStateManagerVersion": "$zkStateManagerVersion"
+        }
+      }
+    """
+    )
+
+    assertThat(stateManagerClient.rollupGetStateMerkleProofWithTypedError(BlockInterval(50UL, 100UL)))
+      .succeedsWithin(5.seconds.toJavaDuration())
+      .isEqualTo(
+        Ok(
+          GetZkEVMStateMerkleProofResponse(
+            zkStateManagerVersion = zkStateManagerVersion,
+            zkStateMerkleProof = zkStateMerkleProof,
+            zkParentStateRootHash = zkParentStateRootHash.decodeHex(),
+            zkEndStateRootHash = zkEndStateRootHash.decodeHex()
+          )
+        )
+      )
+  }
+
+  @Test
+  fun getZkEVMStateMerkleProof_error_block_missing() {
+    wiremockStubForPost(
+      """
+      {
+        "jsonrpc":"2.0",
+        "id":"1",
+        "error":{
+          "code":"-32600",
+          "message":"BLOCK_MISSING_IN_CHAIN - block 1 is missing"
+         }
+      }"""
+    )
+
+    assertThat(stateManagerClient.rollupGetStateMerkleProofWithTypedError(BlockInterval(50UL, 100UL)))
+      .succeedsWithin(5.seconds.toJavaDuration())
+      .isEqualTo(
+        Err(
+          ErrorResponse(
+            StateManagerErrorType.BLOCK_MISSING_IN_CHAIN,
+            "BLOCK_MISSING_IN_CHAIN - block 1 is missing"
+          )
+        )
+      )
+  }
+
+  @Test
+  fun getZkEVMStateMerkleProof_error_unsupported_version() {
+    val response = """
+      {
+        "jsonrpc":"2.0",
+        "id":"1",
+        "error":{
+          "code":"-32602",
+          "message":"UNSUPPORTED_VERSION",
+          "data": {
+            "requestedVersion": "0.1.2",
+            "supportedVersion": "0.0.1-dev-3e607237"
+          }
+         }
+      }"""
+
+    wiremockStubForPost(response)
+
+    assertThat(stateManagerClient.rollupGetStateMerkleProofWithTypedError(BlockInterval(50UL, 100UL)))
+      .succeedsWithin(5.seconds.toJavaDuration())
+      .isEqualTo(
+        Err(
+          ErrorResponse(
+            StateManagerErrorType.UNSUPPORTED_VERSION,
+            "UNSUPPORTED_VERSION: {requestedVersion=0.1.2, supportedVersion=0.0.1-dev-3e607237}"
+          )
+        )
+      )
+  }
+
+  @Test
+  fun getZkEVMStateMerkleProof_error_unknown() {
+    wiremockStubForPost(
+      """
+      {
+        "jsonrpc":"2.0",
+        "id":"1",
+        "error":{
+          "code":-999,
+          "message":"BRA_BRA_BRA_SOME_UNKNOWN_ERROR",
+          "data": {"xyz": "1234", "abc": 100}
+         }
+      }"""
+    )
+
+    assertThat(stateManagerClient.rollupGetStateMerkleProofWithTypedError(BlockInterval(50L, 100L)))
+      .succeedsWithin(5.seconds.toJavaDuration())
+      .isEqualTo(
+        Err(ErrorResponse(StateManagerErrorType.UNKNOWN, """BRA_BRA_BRA_SOME_UNKNOWN_ERROR: {xyz=1234, abc=100}"""))
+      )
+  }
+
+  @Test
+  fun rollupGetHeadBlockNumber_success_response() {
+    wiremockStubForPost("""{"jsonrpc":"2.0","id":1,"result":"0xf1"}""")
+
+    assertThat(stateManagerClient.rollupGetHeadBlockNumber().get())
+      .isEqualTo(ULong.fromHexString("0xf1"))
+  }
+
+  @Test
+  fun rollupGetHeadBlockNumber_error_response() {
+    val response = """{"jsonrpc":"2.0","id":1,"error":{"code": -32603, "message": "Internal error"}}"""
+
+    wiremockStubForPost(response)
+
+    assertThatThrownBy { stateManagerClient.rollupGetHeadBlockNumber().get() }
+      .hasMessageContaining("Internal error")
+  }
+}

settings.gradle

@@ -11,6 +11,7 @@ include 'jvm-libs:generic:vertx-helper'
 include 'jvm-libs:generic:errors'
 include 'jvm-libs:generic:persistence:db'
 
+include 'jvm-libs:linea:clients:linea-state-manager'
 include 'jvm-libs:linea:core:client-interface'
 include 'jvm-libs:linea:core:domain-models'
 include 'jvm-libs:linea:core:long-running-service'