ProofAggregationCoordinatorService: Add poller for proof response

Author: gauravahuja

coordinator/app/src/main/kotlin/net/consensys/zkevm/coordinator/app/conflation/ConflationApp.kt

@@ -383,8 +383,6 @@ class ConflationApp(
         ),
         consecutiveProvenBlobsProvider = maxBlobEndBlockNumberTracker,
         proofAggregationClient = proverClientFactory.proofAggregationProverClient(),
-        l2EthApiClient = l2EthClient,
-        l2MessageService = l2MessageService,
         noL2ActivityTimeout = configs.conflation.conflationDeadlineLastBlockConfirmationDelay,
         waitForNoL2ActivityToTriggerAggregation =
         configs.conflation.proofAggregation.waitForNoL2ActivityToTriggerAggregation,

coordinator/core/src/main/kotlin/net/consensys/zkevm/ethereum/coordination/aggregation/AggregationProofPoller.kt

@@ -0,0 +1,65 @@
+package net.consensys.zkevm.ethereum.coordination.aggregation
+
+import io.vertx.core.Vertx
+import linea.timer.TimerSchedule
+import linea.timer.VertxPeriodicPollingService
+import net.consensys.zkevm.coordinator.clients.ProofAggregationProverClientV2
+import net.consensys.zkevm.domain.Aggregation
+import net.consensys.zkevm.domain.AggregationProofIndex
+import org.apache.logging.log4j.Logger
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+import java.util.concurrent.ConcurrentLinkedDeque
+import kotlin.time.Duration.Companion.milliseconds
+
+class AggregationProofPoller(
+  private val aggregationProofClient: ProofAggregationProverClientV2,
+  private val aggregationProofHandler: AggregationProofHandler,
+  private val log: Logger,
+  vertx: Vertx,
+  pollingIntervalMs: Long = 100.milliseconds.inWholeMilliseconds,
+  name: String = "AggregationProofPoller",
+  timerSchedule: TimerSchedule = TimerSchedule.FIXED_DELAY,
+) : VertxPeriodicPollingService(
+  vertx = vertx,
+  pollingIntervalMs = pollingIntervalMs,
+  log = log,
+  name = name,
+  timerSchedule = timerSchedule,
+) {
+  data class ProofInProgress(
+    val proofIndex: AggregationProofIndex,
+    val unProvenAggregation: Aggregation,
+  )
+
+  private val proofRequestsInProgress = ConcurrentLinkedDeque<ProofInProgress>()
+
+  @Synchronized
+  fun addProofRequestsInProgressForPolling(proofIndex: AggregationProofIndex, unProvenAggregation: Aggregation) {
+    proofRequestsInProgress.add(ProofInProgress(proofIndex, unProvenAggregation))
+  }
+
+  override fun action(): SafeFuture<*> {
+    return if (proofRequestsInProgress.isNotEmpty()) {
+      val proofInProgress = proofRequestsInProgress.peekFirst()
+      aggregationProofClient.findProofResponse(proofInProgress.proofIndex)
+        .thenCompose { proofResponse ->
+          if (proofResponse != null) {
+            log.info(
+              "aggregation proof generated: aggregation={}",
+              proofInProgress.unProvenAggregation.intervalString(),
+            )
+
+            val provenAggregation = proofInProgress.unProvenAggregation.copy(aggregationProof = proofResponse)
+            aggregationProofHandler.acceptNewAggregation(provenAggregation)
+              .thenApply {
+                proofRequestsInProgress.remove(proofInProgress)
+              }
+          } else {
+            SafeFuture.completedFuture(Unit)
+          }
+        }
+    } else {
+      SafeFuture.completedFuture(Unit)
+    }
+  }
+}

coordinator/core/src/main/kotlin/net/consensys/zkevm/ethereum/coordination/aggregation/ProofAggregationCoordinatorService.kt

@@ -2,21 +2,19 @@ package net.consensys.zkevm.ethereum.coordination.aggregation
 
 import io.vertx.core.Vertx
 import linea.LongRunningService
-import linea.contract.l2.L2MessageServiceSmartContractClientReadOnly
 import linea.domain.BlockIntervals
 import linea.domain.toBlockIntervalsString
-import linea.ethapi.EthApiClient
 import linea.timer.TimerSchedule
 import linea.timer.VertxPeriodicPollingService
 import net.consensys.linea.async.AsyncRetryer
 import net.consensys.linea.metrics.LineaMetricsCategory
 import net.consensys.linea.metrics.MetricsFacade
 import net.consensys.zkevm.coordinator.clients.ProofAggregationProverClientV2
 import net.consensys.zkevm.domain.Aggregation
+import net.consensys.zkevm.domain.AggregationProofIndex
 import net.consensys.zkevm.domain.BlobAndBatchCounters
 import net.consensys.zkevm.domain.BlobsToAggregate
 import net.consensys.zkevm.domain.CompressionProofIndex
-import net.consensys.zkevm.domain.ProofToFinalize
 import net.consensys.zkevm.domain.ProofsToAggregate
 import net.consensys.zkevm.ethereum.coordination.blockcreation.SafeBlockProvider
 import org.apache.logging.log4j.LogManager
@@ -53,6 +51,13 @@ class ProofAggregationCoordinatorService(
     val proofGenerationRetryBackoffDelay: Duration,
   )
 
+  internal val aggregationProofPoller: AggregationProofPoller = AggregationProofPoller(
+    aggregationProofClient = proofAggregationClient,
+    aggregationProofHandler = aggregationProofHandler,
+    log = log,
+    vertx = vertx,
+  )
+
   private val pendingBlobs = ConcurrentLinkedQueue<BlobAndBatchCounters>()
   private val aggregationSizeInBlocksHistogram =
     metricsFacade.createHistogram(
@@ -186,28 +191,25 @@ class ProofAggregationCoordinatorService(
         )
       },
     ) {
-      log.debug("requesting aggregation proof: aggregation={}", blobsToAggregate.intervalString())
+      log.debug("creating aggregation proof request: aggregation={}", blobsToAggregate.intervalString())
       aggregationProofCreation(blockIntervals, compressionProofIndexes)
     }
-      .thenPeek {
-        log.info("aggregation proof generated: aggregation={}", blobsToAggregate.intervalString())
-      }
-      .thenCompose { aggregationProof ->
-        val aggregation =
+      .thenApply { aggregationProofIndex ->
+        val unProvenAggregation =
           Aggregation(
             startBlockNumber = blobsToAggregate.startBlockNumber,
             endBlockNumber = blobsToAggregate.endBlockNumber,
             batchCount = batchCount.toULong(),
-            aggregationProof = aggregationProof,
+            aggregationProof = null,
           )
-        aggregationProofHandler.acceptNewAggregation(aggregation)
+        aggregationProofPoller.addProofRequestsInProgressForPolling(aggregationProofIndex, unProvenAggregation)
       }
   }
 
   private fun aggregationProofCreation(
     executionProofsIndexes: BlockIntervals,
     compressionProofIndexes: List<CompressionProofIndex>,
-  ): SafeFuture<ProofToFinalize> {
+  ): SafeFuture<AggregationProofIndex> {
     val blobsToAggregate = executionProofsIndexes.toBlockInterval()
     return aggregationL2StateProvider
       .getAggregationL2State(blockNumber = blobsToAggregate.startBlockNumber.toLong() - 1)
@@ -236,10 +238,10 @@ class ProofAggregationCoordinatorService(
             parentAggregationLastFtxRollingHash = rollingInfo.parentAggregationLastFtxRollingHash,
           )
         }
-          .thenCompose(proofAggregationClient::requestProof)
+          .thenCompose(proofAggregationClient::createProofRequest)
           .whenException {
             log.debug(
-              "Error getting aggregation proof: aggregation={} errorMessage={}",
+              "Error creating aggregation proof request: aggregation={} errorMessage={}",
               executionProofsIndexes.toBlockInterval().intervalString(),
               it.message,
               it,
@@ -263,8 +265,6 @@ class ProofAggregationCoordinatorService(
       aggregationL2StateProvider: AggregationL2StateProvider,
       consecutiveProvenBlobsProvider: ConsecutiveProvenBlobsProvider,
       proofAggregationClient: ProofAggregationProverClientV2,
-      l2EthApiClient: EthApiClient,
-      l2MessageService: L2MessageServiceSmartContractClientReadOnly,
       noL2ActivityTimeout: Duration,
       waitForNoL2ActivityToTriggerAggregation: Boolean,
       targetEndBlockNumbers: List<ULong>,
@@ -352,4 +352,16 @@ class ProofAggregationCoordinatorService(
   override fun handleError(error: Throwable) {
     log.error("Error polling blobs for aggregation: errorMessage={}", error.message, error)
   }
+
+  override fun start(): SafeFuture<Unit> {
+    return aggregationProofPoller.start().thenCompose {
+      super.start()
+    }
+  }
+
+  override fun stop(): SafeFuture<Unit> {
+    return super.stop().thenCompose {
+      aggregationProofPoller.stop()
+    }
+  }
 }

coordinator/core/src/test/kotlin/net/consensys/zkevm/ethereum/coordination/aggregation/ProofAggregationCoordinatorServiceTest.kt

@@ -8,6 +8,7 @@ import net.consensys.linea.metrics.MetricsFacade
 import net.consensys.linea.metrics.micrometer.MicrometerMetricsFacade
 import net.consensys.zkevm.coordinator.clients.ProofAggregationProverClientV2
 import net.consensys.zkevm.domain.Aggregation
+import net.consensys.zkevm.domain.AggregationProofIndex
 import net.consensys.zkevm.domain.BlobAndBatchCounters
 import net.consensys.zkevm.domain.BlobCounters
 import net.consensys.zkevm.domain.BlobsToAggregate
@@ -98,6 +99,8 @@ class ProofAggregationCoordinatorServiceTest {
         metricsFacade = metricsFacade,
         invalidityProofProvider = mockInvalidityProofProvider,
       )
+    proofAggregationCoordinatorService.aggregationProofPoller.start()
+
     verify(mockAggregationCalculator).onAggregation(proofAggregationCoordinatorService)
 
     val blob1 = listOf(createBlob(11u, 19u), createBlob(20u, 33u), createBlob(34u, 41u))
@@ -242,6 +245,11 @@ class ProofAggregationCoordinatorServiceTest {
         batchCount = compressionBlobs1.sumOf { it.blobCounters.numberOfBatches }.toULong(),
         aggregationProof = aggregationProof1,
       )
+    val aggregation1ProofIndex = AggregationProofIndex(
+      startBlockNumber = aggregation1.startBlockNumber,
+      endBlockNumber = aggregation1.endBlockNumber,
+      hash = Random.nextBytes(32),
+    )
 
     val aggregation2 =
       Aggregation(
@@ -250,18 +258,33 @@ class ProofAggregationCoordinatorServiceTest {
         batchCount = compressionBlobs2.sumOf { it.blobCounters.numberOfBatches }.toULong(),
         aggregationProof = aggregationProof2,
       )
+    val aggregation2ProofIndex = AggregationProofIndex(
+      startBlockNumber = aggregation2.startBlockNumber,
+      endBlockNumber = aggregation2.endBlockNumber,
+      hash = Random.nextBytes(32),
+    )
 
-    whenever(mockProofAggregationClient.requestProof(any()))
+    whenever(mockProofAggregationClient.createProofRequest(any()))
       .thenAnswer {
         if (it.getArgument<ProofsToAggregate>(0) == proofsToAggregate1) {
-          SafeFuture.completedFuture(aggregationProof1)
+          SafeFuture.completedFuture(aggregation1ProofIndex)
         } else if (it.getArgument<ProofsToAggregate>(0) == proofsToAggregate2) {
-          SafeFuture.completedFuture(aggregationProof2)
+          SafeFuture.completedFuture(aggregation2ProofIndex)
         } else {
           throw IllegalStateException()
         }
       }
 
+    whenever(mockProofAggregationClient.findProofResponse(any<AggregationProofIndex>()))
+      .thenAnswer {
+        val proofIndex = it.getArgument<AggregationProofIndex>(0)
+        when (proofIndex) {
+          aggregation1ProofIndex -> SafeFuture.completedFuture(aggregationProof1)
+          aggregation2ProofIndex -> SafeFuture.completedFuture(aggregationProof2)
+          else -> SafeFuture.completedFuture(null)
+        }
+      }
+
     whenever(
       mockAggregationsRepository.findHighestConsecutiveEndBlockNumber(
         aggregation1.startBlockNumber.toLong(),
@@ -301,7 +324,8 @@ class ProofAggregationCoordinatorServiceTest {
         assertThat(meterRegistry.summary("aggregation.blocks.size").max()).isEqualTo(23.0)
         assertThat(meterRegistry.summary("aggregation.batches.size").max()).isEqualTo(6.0)
         assertThat(meterRegistry.summary("aggregation.blobs.size").max()).isEqualTo(2.0)
-        verify(mockProofAggregationClient).requestProof(proofsToAggregate1)
+        verify(mockProofAggregationClient).createProofRequest(proofsToAggregate1)
+        verify(mockProofAggregationClient).findProofResponse(aggregation1ProofIndex)
         verify(mockAggregationsRepository).saveNewAggregation(aggregation1)
         assertThat(provenAggregation).isEqualTo(aggregation1.endBlockNumber)
       }
@@ -319,7 +343,8 @@ class ProofAggregationCoordinatorServiceTest {
         assertThat(meterRegistry.summary("aggregation.blocks.size").max()).isEqualTo(27.0)
         assertThat(meterRegistry.summary("aggregation.batches.size").max()).isEqualTo(6.0)
         assertThat(meterRegistry.summary("aggregation.blobs.size").max()).isEqualTo(2.0)
-        verify(mockProofAggregationClient).requestProof(proofsToAggregate2)
+        verify(mockProofAggregationClient).createProofRequest(proofsToAggregate2)
+        verify(mockProofAggregationClient).findProofResponse(aggregation2ProofIndex)
         verify(mockAggregationsRepository).saveNewAggregation(aggregation2)
         assertThat(provenAggregation).isEqualTo(aggregation2.endBlockNumber)
       }