feat: server ProcessOperatorSignedTaskResponseV2

Author: uri-99

aggregator/internal/pkg/server.go

@@ -49,6 +49,97 @@ func (agg *Aggregator) ServeOperators() error {
 //   - 0: Success
 //   - 1: Error
 func (agg *Aggregator) ProcessOperatorSignedTaskResponse(signedTaskResponse *types.SignedTaskResponse, reply *uint8) error {
+	agg.AggregatorConfig.BaseConfig.Logger.Info("New task response",
+		"BatchMerkleRoot", "0x"+hex.EncodeToString(signedTaskResponse.BatchMerkleRoot[:]),
+		"operatorId", hex.EncodeToString(signedTaskResponse.OperatorId[:]))
+
+	taskIndex := uint32(0)
+	ok := false
+
+	for i := 0; i < waitForEventRetries; i++ {
+		agg.taskMutex.Lock()
+		agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Starting processing of Response")
+		taskIndex, ok = agg.batchesIdxByIdentifierHash[signedTaskResponse.BatchMerkleRoot]
+		if !ok {
+			agg.taskMutex.Unlock()
+			agg.logger.Info("- Unlocked Resources: Task not found in the internal map")
+			time.Sleep(waitForEventSleepSeconds)
+		} else {
+			break
+		}
+	}
+
+	if !ok {
+		agg.logger.Warn("Task not found in the internal map, operator signature will be lost. Batch may not reach quorum")
+		*reply = 1
+		return nil
+	}
+
+	// Note: we already have lock here
+	agg.logger.Debug("- Checking if operator already responded")
+	batchResponses, ok := agg.operatorRespondedBatch[taskIndex]
+	if !ok {
+		batchResponses = make(map[eigentypes.Bytes32]struct{})
+		agg.operatorRespondedBatch[taskIndex] = batchResponses
+	}
+
+	if _, ok := batchResponses[signedTaskResponse.OperatorId]; ok {
+		*reply = 0
+		agg.logger.Warn("Operator already responded, ignoring",
+			"operatorId", hex.EncodeToString(signedTaskResponse.OperatorId[:]),
+			"taskIndex", taskIndex, "batchMerkleRoot", hex.EncodeToString(signedTaskResponse.BatchMerkleRoot[:]))
+
+		agg.taskMutex.Unlock()
+		return nil
+	}
+
+	batchResponses[signedTaskResponse.OperatorId] = struct{}{}
+
+	// Don't wait infinitely if it can't answer
+	// Create a context with a timeout of 5 seconds
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel() // Ensure the cancel function is called to release resources
+
+	// Create a channel to signal when the task is done
+	done := make(chan struct{})
+
+	agg.logger.Info("Starting bls signature process")
+	go func() {
+		err := agg.blsAggregationService.ProcessNewSignature(
+			context.Background(), taskIndex, signedTaskResponse.BatchMerkleRoot,
+			&signedTaskResponse.BlsSignature, signedTaskResponse.OperatorId,
+		)
+
+		if err != nil {
+			agg.logger.Warnf("BLS aggregation service error: %s", err)
+			// remove operator from the list of operators that responded
+			// so that it can try again
+			delete(batchResponses, signedTaskResponse.OperatorId)
+		} else {
+			agg.logger.Info("BLS process succeeded")
+		}
+
+		close(done)
+	}()
+
+	*reply = 1
+	// Wait for either the context to be done or the task to complete
+	select {
+	case <-ctx.Done():
+		// The context's deadline was exceeded or it was canceled
+		agg.logger.Info("Bls process timed out, operator signature will be lost. Batch may not reach quorum")
+	case <-done:
+		// The task completed successfully
+		agg.logger.Info("Bls context finished correctly")
+		*reply = 0
+	}
+
+	agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Task response processing finished")
+	agg.taskMutex.Unlock()
+
+	return nil
+}
+func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *types.SignedTaskResponseV2, reply *uint8) error {
 	agg.AggregatorConfig.BaseConfig.Logger.Info("New task response",
 		"BatchMerkleRoot", "0x"+hex.EncodeToString(signedTaskResponse.BatchMerkleRoot[:]),
 		"SenderAddress", "0x"+hex.EncodeToString(signedTaskResponse.SenderAddress[:]),