V0.6.0

Makefile

@@ -5,7 +5,7 @@ OS := $(shell uname -s)
 CONFIG_FILE?=config-files/config.yaml
 AGG_CONFIG_FILE?=config-files/config-aggregator.yaml
 
-OPERATOR_VERSION=v0.4.1
+OPERATOR_VERSION=v0.5.2
 
 ifeq ($(OS),Linux)
 	BUILD_ALL_FFI = $(MAKE) build_all_ffi_linux

aggregator/cmd/main.go

@@ -49,7 +49,7 @@ func aggregatorMain(ctx *cli.Context) error {
 		return err
 	}
 
-	// Listen for new task created in the ServiceManager contract in a separate goroutine
+	// Listen for new task created in the ServiceManager contract in a separate goroutine, both V1 and V2 subscriptions:
 	go func() {
 		listenErr := aggregator.SubscribeToNewTasks()
 		if listenErr != nil {

aggregator/internal/pkg/aggregator.go

@@ -43,6 +43,7 @@ type BatchData struct {
 type Aggregator struct {
 	AggregatorConfig      *config.AggregatorConfig
 	NewBatchChan          chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch
+	NewBatchChanV2        chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2
 	avsReader             *chainio.AvsReader
 	avsSubscriber         *chainio.AvsSubscriber
 	avsWriter             *chainio.AvsWriter
@@ -92,6 +93,7 @@ type Aggregator struct {
 
 func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error) {
 	newBatchChan := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch)
+	newBatchChanV2 := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2)
 
 	avsReader, err := chainio.NewAvsReaderFromConfig(aggregatorConfig.BaseConfig, aggregatorConfig.EcdsaConfig)
 	if err != nil {
@@ -162,6 +164,7 @@ func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error
 		avsSubscriber:    avsSubscriber,
 		avsWriter:        avsWriter,
 		NewBatchChan:     newBatchChan,
+		NewBatchChanV2:   newBatchChanV2,
 
 		batchesIdentifierHashByIdx: batchesIdentifierHashByIdx,
 		batchesIdxByIdentifierHash: batchesIdxByIdentifierHash,
@@ -207,14 +210,16 @@ func (agg *Aggregator) Start(ctx context.Context) error {
 		case blsAggServiceResp := <-agg.blsAggregationService.GetResponseChannel():
 			agg.logger.Info("Received response from BLS aggregation service",
 				"taskIndex", blsAggServiceResp.TaskIndex)
-
+			
 			go agg.handleBlsAggServiceResponse(blsAggServiceResp)
 		}
 	}
 }
 
 const MaxSentTxRetries = 5
 
+var switchBlockNumber = uint64(2_268_375) // 2_268_375 is the block at sep 3th 15:00
+
 func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsAggregationServiceResponse) {
 	if blsAggServiceResp.Err != nil {
 		agg.taskMutex.Lock()
@@ -276,19 +281,37 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 
 	agg.logger.Info("Sending aggregated response onchain", "taskIndex", blsAggServiceResp.TaskIndex,
 		"batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
-
+	current_task_block := agg.batchCreatedBlockByIdx[blsAggServiceResp.TaskIndex]
 	for i := 0; i < MaxSentTxRetries; i++ {
-		_, err = agg.sendAggregatedResponse(batchData.BatchMerkleRoot, batchData.SenderAddress, nonSignerStakesAndSignature)
-		if err == nil {
-			agg.logger.Info("Aggregator successfully responded to task",
-				"taskIndex", blsAggServiceResp.TaskIndex,
-				"batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
-
-			return
+		if current_task_block < switchBlockNumber {
+			agg.logger.Info("agg if V1")
+			_, err = agg.sendAggregatedResponse(batchData.BatchMerkleRoot, nonSignerStakesAndSignature)
+			if err == nil {
+				agg.logger.Info("Aggregator successfully responded to task",
+					"taskIndex", blsAggServiceResp.TaskIndex,
+					"batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
+
+				return
+			}
+
+			// Sleep for a bit before retrying
+			time.Sleep(2 * time.Second)
+
+		} else {
+			agg.logger.Info("agg if V2")
+			_, err = agg.sendAggregatedResponseV2(batchData.BatchMerkleRoot, batchData.SenderAddress, nonSignerStakesAndSignature)
+			if err == nil {
+				agg.logger.Info("Aggregator successfully responded to task",
+					"taskIndex", blsAggServiceResp.TaskIndex,
+					"batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
+
+				return
+			}
+
+			// Sleep for a bit before retrying
+			time.Sleep(2 * time.Second)
 		}
 
-		// Sleep for a bit before retrying
-		time.Sleep(2 * time.Second)
 	}
 
 	agg.logger.Error("Aggregator failed to respond to task, this batch will be lost",
@@ -301,7 +324,32 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 
 // / Sends response to contract and waits for transaction receipt
 // / Returns error if it fails to send tx or receipt is not found
-func (agg *Aggregator) sendAggregatedResponse(batchMerkleRoot [32]byte, senderAddress [20]byte, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature) (*gethtypes.Receipt, error) {
+func (agg *Aggregator) sendAggregatedResponse(batchMerkleRoot [32]byte, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature) (*gethtypes.Receipt, error) {
+	agg.walletMutex.Lock()
+	agg.logger.Infof("- Locked Wallet Resources: Sending aggregated response for batch",
+		"merkleRoot", hex.EncodeToString(batchMerkleRoot[:]))
+
+	txHash, err := agg.avsWriter.SendAggregatedResponse(batchMerkleRoot, nonSignerStakesAndSignature)
+	if err != nil {
+		agg.walletMutex.Unlock()
+		agg.logger.Infof("- Unlocked Wallet Resources: Error sending aggregated response for batch %s. Error: %s", hex.EncodeToString(batchMerkleRoot[:]), err)
+		return nil, err
+	}
+
+	agg.walletMutex.Unlock()
+	agg.logger.Infof("- Unlocked Wallet Resources: Sending aggregated response for batch %s", hex.EncodeToString(batchMerkleRoot[:]))
+
+	receipt, err := utils.WaitForTransactionReceipt(
+		agg.AggregatorConfig.BaseConfig.EthRpcClient, context.Background(), *txHash)
+	if err != nil {
+		return nil, err
+	}
+
+	agg.metrics.IncAggregatedResponses()
+
+	return receipt, nil
+}
+func (agg *Aggregator) sendAggregatedResponseV2(batchMerkleRoot [32]byte, senderAddress [20]byte, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature) (*gethtypes.Receipt, error) {
 	batchIdentifier := append(batchMerkleRoot[:], senderAddress[:]...)
 	var batchIdentifierHash = *(*[32]byte)(crypto.Keccak256(batchIdentifier))
 
@@ -311,7 +359,7 @@ func (agg *Aggregator) sendAggregatedResponse(batchMerkleRoot [32]byte, senderAd
 		"senderAddress", hex.EncodeToString(senderAddress[:]),
 		"batchIdentifierHash", hex.EncodeToString(batchIdentifierHash[:]))
 
-	txHash, err := agg.avsWriter.SendAggregatedResponse(batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+	txHash, err := agg.avsWriter.SendAggregatedResponseV2(batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
 	if err != nil {
 		agg.walletMutex.Unlock()
 		agg.logger.Infof("- Unlocked Wallet Resources: Error sending aggregated response for batch %s. Error: %s", hex.EncodeToString(batchIdentifierHash[:]), err)
@@ -332,8 +380,53 @@ func (agg *Aggregator) sendAggregatedResponse(batchMerkleRoot [32]byte, senderAd
 	return receipt, nil
 }
 
-func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]byte, taskCreatedBlock uint32) {
+func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, taskCreatedBlock uint32) {
+	agg.AggregatorConfig.BaseConfig.Logger.Info("Adding new task",
+		"Batch merkle root", "0x"+hex.EncodeToString(batchMerkleRoot[:]))
 
+	agg.taskMutex.Lock()
+	agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Adding new task")
+
+	// --- UPDATE BATCH - INDEX CACHES ---
+	batchIndex := agg.nextBatchIndex
+	if _, ok := agg.batchesIdxByIdentifierHash[batchMerkleRoot]; ok {
+		agg.logger.Warn("Batch already exists", "batchIndex", batchIndex, "batchIdentifierHash (actually batchMerkleRoot)", batchMerkleRoot)
+		agg.taskMutex.Unlock()
+		agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Adding new task")
+		return
+	}
+
+	// This shouldn't happen, since both maps are updated together
+	if _, ok := agg.batchesIdentifierHashByIdx[batchIndex]; ok {
+		agg.logger.Warn("Batch already exists", "batchIndex", batchIndex, "batchIdentifierHash (actually batchMerkleRoot)", batchMerkleRoot)
+		agg.taskMutex.Unlock()
+		agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Adding new task")
+		return
+	}
+
+	agg.batchesIdxByIdentifierHash[batchMerkleRoot] = batchIndex
+	agg.batchCreatedBlockByIdx[batchIndex] = uint64(taskCreatedBlock)
+	agg.batchesIdentifierHashByIdx[batchIndex] = batchMerkleRoot
+	agg.batchDataByIdentifierHash[batchMerkleRoot] = BatchData{
+		BatchMerkleRoot: batchMerkleRoot,
+		SenderAddress:   [20]byte{},
+	}
+	agg.nextBatchIndex += 1
+
+	quorumNums := eigentypes.QuorumNums{eigentypes.QuorumNum(QUORUM_NUMBER)}
+	quorumThresholdPercentages := eigentypes.QuorumThresholdPercentages{eigentypes.QuorumThresholdPercentage(QUORUM_THRESHOLD)}
+
+	err := agg.blsAggregationService.InitializeNewTask(batchIndex, taskCreatedBlock, quorumNums, quorumThresholdPercentages, 100*time.Second)
+	// FIXME(marian): When this errors, should we retry initializing new task? Logging fatal for now.
+	if err != nil {
+		agg.logger.Fatalf("BLS aggregation service error when initializing new task: %s", err)
+	}
+
+	agg.taskMutex.Unlock()
+	agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Adding new task")
+	agg.logger.Info("New task added", "batchIndex", batchIndex, "batchIdentifierHash (actually batchMerkleRoot)", "0x"+hex.EncodeToString(batchMerkleRoot[:]))
+}
+func (agg *Aggregator) AddNewTaskV2(batchMerkleRoot [32]byte, senderAddress [20]byte, taskCreatedBlock uint32) {
 	batchIdentifier := append(batchMerkleRoot[:], senderAddress[:]...)
 	var batchIdentifierHash = *(*[32]byte)(crypto.Keccak256(batchIdentifier))
 

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[:]),

aggregator/internal/pkg/subscriber.go

@@ -5,21 +5,39 @@ func (agg *Aggregator) SubscribeToNewTasks() error {
 	if err != nil {
 		return err
 	}
+	err = agg.subscribeToNewTasksV2()
+	if err != nil {
+		return err
+	}
+
+	var switchBlockNumber = uint32(2_268_375) // 2_268_375 is the block at sep 3th 15:00
 
 	for {
 		select {
 		case err := <-agg.taskSubscriber:
 			agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to subscribe to new tasks", "err", err)
 
+			// TODO not sure if this is the best way, but no way to calculate blocknumber from here
 			err = agg.subscribeToNewTasks()
+			errV2 := agg.subscribeToNewTasksV2()
 			if err != nil {
 				return err
 			}
+			if errV2 != nil {
+				return err
+			}
 		case newBatch := <-agg.NewBatchChan:
-			agg.AddNewTask(newBatch.BatchMerkleRoot, newBatch.SenderAddress, newBatch.TaskCreatedBlock)
+			if newBatch.TaskCreatedBlock < switchBlockNumber {
+				agg.AggregatorConfig.BaseConfig.Logger.Info("Adding new task, V1")
+				agg.AddNewTask(newBatch.BatchMerkleRoot, newBatch.TaskCreatedBlock)
+			}
+		case newBatchV2 := <-agg.NewBatchChanV2:
+			if newBatchV2.TaskCreatedBlock >= switchBlockNumber {
+				agg.AggregatorConfig.BaseConfig.Logger.Info("Adding new task, V2")
+				agg.AddNewTaskV2(newBatchV2.BatchMerkleRoot, newBatchV2.SenderAddress, newBatchV2.TaskCreatedBlock)
+			}
 		}
 	}
-
 }
 
 func (agg *Aggregator) subscribeToNewTasks() error {
@@ -33,3 +51,14 @@ func (agg *Aggregator) subscribeToNewTasks() error {
 
 	return err
 }
+func (agg *Aggregator) subscribeToNewTasksV2() error {
+	var err error
+
+	agg.taskSubscriber, err = agg.avsSubscriber.SubscribeToNewTasksV2(agg.NewBatchChanV2)
+
+	if err != nil {
+		agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to create task subscriber", "err", err)
+	}
+
+	return err
+}

batcher/aligned-sdk/src/communication/batch.rs

@@ -44,11 +44,17 @@ pub async fn await_batch_verification(
     aligned_verification_data: &AlignedVerificationData,
     rpc_url: &str,
     chain: Chain,
+    payment_service_addr: &str,
 ) -> Result<(), errors::SubmitError> {
     for _ in 0..RETRIES {
-        if is_proof_verified(aligned_verification_data, chain.clone(), rpc_url)
-            .await
-            .is_ok_and(|r| r)
+        if is_proof_verified(
+            aligned_verification_data,
+            chain.clone(),
+            rpc_url,
+            payment_service_addr,
+        )
+        .await
+        .is_ok_and(|r| r)
         {
             return Ok(());
         }