Coordinator forced tx part14

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

@@ -98,15 +98,6 @@ fun createAppClients(
   stateManagerRequestRetry: RetryConfig = RetryConfig(backoffDelay = 1.seconds),
   zkStateManagerVersion: String = "2.3.0",
 ): AppClients {
-  val lineaContractClient =
-    Web3JLineaRollupSmartContractClientReadOnly(
-      contractAddress = smartContractAddress,
-      web3j =
-      createWeb3jHttpClient(
-        rpcUrl = l1RpcEndpoint.toString(),
-        log = LogManager.getLogger("linea.plugin.staterecovery.clients.l1.smart-contract"),
-      ),
-    )
   val ethLogsSearcher =
     run {
       val log = LogManager.getLogger("linea.plugin.staterecovery.clients.l1.logs-searcher")
@@ -127,6 +118,16 @@ fun createAppClients(
         log = log,
       )
     }
+  val lineaContractClient =
+    Web3JLineaRollupSmartContractClientReadOnly(
+      contractAddress = smartContractAddress,
+      web3j =
+      createWeb3jHttpClient(
+        rpcUrl = l1RpcEndpoint.toString(),
+        log = LogManager.getLogger("linea.plugin.staterecovery.clients.l1.smart-contract"),
+      ),
+      ethLogsClient = ethLogsSearcher,
+    )
   val blobScanClient =
     BlobScanClient.create(
       vertx = vertx,

config/coordinator/coordinator-config-v2-override-local-dev.toml

@@ -9,6 +9,9 @@ fs-responses-directory = "tmp/local/prover/v3/execution/responses"
 [prover.blob-compression]
 fs-requests-directory = "tmp/local/prover/v3/compression/requests"
 fs-responses-directory = "tmp/local/prover/v3/compression/responses"
+[prover.invalidity]
+fs-requests-directory = "tmp/local/prover/v3/invalidity/requests"
+fs-responses-directory = "tmp/local/prover/v3/invalidity/responses"
 [prover.proof-aggregation]
 fs-requests-directory = "tmp/local/prover/v3/aggregation/requests"
 fs-responses-directory = "tmp/local/prover/v3/aggregation/responses"

config/coordinator/coordinator-config-v2.toml

@@ -62,6 +62,9 @@ fs-responses-directory = "/data/prover/v3/execution/responses"
 [prover.blob-compression]
 fs-requests-directory = "/data/prover/v3/compression/requests"
 fs-responses-directory = "/data/prover/v3/compression/responses"
+[prover.invalidity]
+fs-requests-directory = "/data/prover/v3/invalidity/requests"
+fs-responses-directory = "/data/prover/v3/invalidity/responses"
 [prover.proof-aggregation]
 fs-requests-directory = "/data/prover/v3/aggregation/requests"
 fs-responses-directory = "/data/prover/v3/aggregation/responses"

coordinator/app/build.gradle

@@ -30,11 +30,13 @@ dependencies {
   implementation project(':coordinator:clients:traces-generator-api-client')
   implementation project(':coordinator:clients:prover-client:file-based-client')
   implementation project(':coordinator:clients:smart-contract-client')
+  implementation project(':coordinator:clients:forced-transactions-client')
   implementation project(':coordinator:ethereum:finalization-monitor')
   implementation project(':coordinator:ethereum:common')
   implementation project(':coordinator:ethereum:models-helper')
   implementation project(':coordinator:ethereum:blob-submitter')
   implementation project(':coordinator:ethereum:message-anchoring')
+  implementation project(':coordinator:ethereum:forced-transactions')
   implementation project(':coordinator:clients:web3signer-client')
   implementation project(':coordinator:persistence:blob')
   implementation project(':coordinator:persistence:aggregation')

coordinator/app/src/main/kotlin/linea/coordinator/config/v2/ForcedTransactionsConfig.kt

@@ -20,6 +20,7 @@ data class ForcedTransactionsConfig(
   val processingDelay: Duration = Duration.ZERO,
   val l1EventScraping: L1EventScraping = L1EventScraping(),
   val processingBatchSize: UInt = 10u,
+  val invalidityProofCheckInterval: Duration = 2.minutes,
 ) : FeatureToggle {
   init {
     require(processingTickInterval >= 1.milliseconds) {
@@ -28,6 +29,12 @@ data class ForcedTransactionsConfig(
     require(processingDelay >= Duration.ZERO) {
       "processingDelay=$processingDelay must be equal or greater than 0ms"
     }
+    require(processingBatchSize >= 1u) {
+      "processingBatchSize=$processingBatchSize must be equal or greater than 1"
+    }
+    require(invalidityProofCheckInterval >= 1.milliseconds) {
+      "invalidityProofCheckInterval=$invalidityProofCheckInterval must be equal or greater than 1ms"
+    }
   }
 
   data class L1EventScraping(

coordinator/app/src/main/kotlin/linea/coordinator/config/v2/toml/ProverToml.kt

@@ -15,6 +15,7 @@ data class ProverToml(
   val fsPollingTimeout: Duration = Duration.INFINITE,
   val execution: ProverDirectoriesToml,
   val blobCompression: ProverDirectoriesToml,
+  val invalidity: ProverDirectoriesToml,
   val proofAggregation: ProverDirectoriesToml,
   val switchBlockNumberInclusive: ULong? = null,
   val new: ProverToml? = null,
@@ -47,6 +48,14 @@ data class ProverToml(
           pollingInterval = this.fsPollingInterval,
           pollingTimeout = this.fsPollingTimeout,
         ),
+        invalidity = FileBasedProverConfig(
+          requestsDirectory = Path.of(this.invalidity.fsRequestsDirectory),
+          responsesDirectory = Path.of(this.invalidity.fsResponsesDirectory),
+          inprogressProvingSuffixPattern = this.fsInprogressProvingSuffixPattern,
+          inprogressRequestWritingSuffix = this.fsInprogressRequestWritingSuffix,
+          pollingInterval = this.fsPollingInterval,
+          pollingTimeout = this.fsPollingTimeout,
+        ),
         proofAggregation =
         FileBasedProverConfig(
           requestsDirectory = Path.of(this.proofAggregation.fsRequestsDirectory),
@@ -79,6 +88,14 @@ data class ProverToml(
             pollingInterval = newProverConfig.fsPollingInterval,
             pollingTimeout = newProverConfig.fsPollingTimeout,
           ),
+          invalidity = FileBasedProverConfig(
+            requestsDirectory = Path.of(newProverConfig.invalidity.fsRequestsDirectory),
+            responsesDirectory = Path.of(newProverConfig.invalidity.fsResponsesDirectory),
+            inprogressProvingSuffixPattern = newProverConfig.fsInprogressProvingSuffixPattern,
+            inprogressRequestWritingSuffix = newProverConfig.fsInprogressRequestWritingSuffix,
+            pollingInterval = newProverConfig.fsPollingInterval,
+            pollingTimeout = newProverConfig.fsPollingTimeout,
+          ),
           proofAggregation =
           FileBasedProverConfig(
             requestsDirectory = Path.of(newProverConfig.proofAggregation.fsRequestsDirectory),

coordinator/app/src/main/kotlin/net/consensys/zkevm/coordinator/app/CoordinatorApp.kt

@@ -7,6 +7,8 @@ import io.vertx.micrometer.backends.NoopBackendRegistry
 import io.vertx.sqlclient.SqlClient
 import linea.coordinator.config.v2.CoordinatorConfig
 import linea.coordinator.config.v2.DatabaseConfig
+import linea.persistence.ftx.PostgresForcedTransactionsDao
+import linea.persistence.ftx.RetryingPostgresForcedTransactionsDao
 import net.consensys.linea.async.toSafeFuture
 import net.consensys.linea.jsonrpc.client.LoadBalancingJsonRpcClient
 import net.consensys.linea.jsonrpc.client.VertxHttpJsonRpcClientFactory
@@ -30,8 +32,12 @@ import org.apache.logging.log4j.Level
 import org.apache.logging.log4j.LogManager
 import org.apache.logging.log4j.Logger
 import tech.pegasys.teku.infrastructure.async.SafeFuture
+import kotlin.time.Clock
 
-class CoordinatorApp(private val configs: CoordinatorConfig) {
+class CoordinatorApp(
+  private val configs: CoordinatorConfig,
+  private val clock: Clock = Clock.System,
+) {
   private val log: Logger = LogManager.getLogger(this::class.java)
   private val vertx: Vertx =
     run {
@@ -96,6 +102,14 @@ class CoordinatorApp(private val configs: CoordinatorConfig) {
       ),
     )
 
+  private val forcedTransactionsDao = RetryingPostgresForcedTransactionsDao(
+    delegate =
+    PostgresForcedTransactionsDao(
+      connection = sqlClient,
+    ),
+    persistenceRetryer = persistenceRetryer,
+  )
+
   private val blobsRepository =
     BlobsRepositoryImpl(
       blobsDao =
@@ -131,10 +145,12 @@ class CoordinatorApp(private val configs: CoordinatorConfig) {
       httpJsonRpcClientFactory = httpJsonRpcClientFactory,
       batchesRepository = batchesRepository,
       blobsRepository = blobsRepository,
+      forcedTransactionsDao = forcedTransactionsDao,
       aggregationsRepository = aggregationsRepository,
       sqlClient = sqlClient,
       smartContractErrors = configs.smartContractErrors,
       metricsFacade = micrometerMetricsFacade,
+      clock = this.clock,
     )
 
   private val requestFileCleanup =

coordinator/app/src/main/kotlin/net/consensys/zkevm/coordinator/app/L1DependentApp.kt

@@ -14,6 +14,7 @@ import linea.domain.BlockNumberAndHash
 import linea.domain.RetryConfig
 import linea.ethapi.EthApiClient
 import linea.kotlin.toKWeiUInt
+import linea.persistence.ftx.RetryingPostgresForcedTransactionsDao
 import linea.web3j.SmartContractErrors
 import linea.web3j.createWeb3jHttpClient
 import linea.web3j.ethapi.createEthApiClient
@@ -77,10 +78,12 @@ class L1DependentApp(
   private val httpJsonRpcClientFactory: VertxHttpJsonRpcClientFactory,
   private val batchesRepository: BatchesRepository,
   private val blobsRepository: BlobsRepository,
+  private val forcedTransactionsDao: RetryingPostgresForcedTransactionsDao,
   private val aggregationsRepository: AggregationsRepository,
   private val sqlClient: SqlClient,
   private val smartContractErrors: SmartContractErrors,
   private val metricsFacade: MetricsFacade,
+  private val clock: Clock,
 ) : LongRunningService {
   private val log = LogManager.getLogger(this::class.java)
 
@@ -163,14 +166,21 @@ class L1DependentApp(
     )
   }
 
-  val lineaRollupClientForFinalizationMonitor: LineaRollupSmartContractClientReadOnly =
+  val lineaRollupClientForFinalizationMonitor: LineaRollupSmartContractClientReadOnly = run {
+    val web3j = createWeb3jHttpClient(
+      rpcUrl = configs.l1FinalizationMonitor.l1Endpoint.toString(),
+      log = LogManager.getLogger("clients.l1.eth.finalization-monitor"),
+    )
     Web3JLineaRollupSmartContractClientReadOnly(
       contractAddress = configs.protocol.l1.contractAddress,
-      web3j = createWeb3jHttpClient(
-        rpcUrl = configs.l1FinalizationMonitor.l1Endpoint.toString(),
-        log = LogManager.getLogger("clients.l1.eth.finalization-monitor"),
+      web3j = web3j,
+      ethLogsClient = createEthApiClient(
+        web3jClient = web3j,
+        requestRetryConfig = configs.l1FinalizationMonitor.l1RequestRetries,
+        vertx = vertx,
       ),
     )
+  }
 
   private val l1FinalizationMonitor = run {
     FinalizationMonitorImpl(
@@ -496,9 +506,11 @@ class L1DependentApp(
     batchesRepository = batchesRepository,
     blobsRepository = blobsRepository,
     aggregationsRepository = aggregationsRepository,
+    forcedTransactionsDao = forcedTransactionsDao,
     lastFinalizedBlock = lastFinalizedBlock,
     metricsFacade = metricsFacade,
     httpJsonRpcClientFactory = httpJsonRpcClientFactory,
+    clock = clock,
   )
 
   private val l2NetworkGasPricingService: L2NetworkGasPricingService? =