feat: subscribeToNewTasksV2, SubscribeToNewTasksV2, getLatestTaskFromEthereumV2

Author: uri-99

core/chainio/avs_subscriber.go

@@ -132,6 +132,73 @@ func (s *AvsSubscriber) SubscribeToNewTasks(newTaskCreatedChan chan *servicemana
 
 	return errorChannel, nil
 }
+func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) (chan error, error) {
+	// Create a new channel to receive new tasks
+	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2)
+
+	// Subscribe to new tasks
+	sub, err := subscribeToNewTasksV2(s.AvsContractBindings.ServiceManager, internalChannel, s.logger)
+	if err != nil {
+		s.logger.Error("Failed to subscribe to new AlignedLayer tasks", "err", err)
+		return nil, err
+	}
+
+	subFallback, err := subscribeToNewTasksV2(s.AvsContractBindings.ServiceManagerFallback, internalChannel, s.logger)
+	if err != nil {
+		s.logger.Error("Failed to subscribe to new AlignedLayer tasks", "err", err)
+		return nil, err
+	}
+
+	// create a new channel to foward errors
+	errorChannel := make(chan error)
+
+	pollLatestBatchTicker := time.NewTicker(PollLatestBatchInterval)
+
+	// Forward the new tasks to the provided channel
+	go func() {
+		defer pollLatestBatchTicker.Stop()
+		newBatchMutex := &sync.Mutex{}
+		batchesSet := make(map[[32]byte]struct{})
+		for {
+			select {
+			case newBatch := <-internalChannel:
+				s.processNewBatchV2(newBatch, batchesSet, newBatchMutex, newTaskCreatedChan)
+			case <-pollLatestBatchTicker.C:
+				latestBatch, err := s.getLatestTaskFromEthereumV2()
+				if err != nil {
+					s.logger.Debug("Failed to get latest task from blockchain", "err", err)
+					continue
+				}
+				s.processNewBatchV2(latestBatch, batchesSet, newBatchMutex, newTaskCreatedChan)
+			}
+		}
+
+	}()
+
+	// Handle errors and resubscribe
+	go func() {
+		for {
+			select {
+			case err := <-sub.Err():
+				s.logger.Warn("Error in new task subscription", "err", err)
+				sub.Unsubscribe()
+				sub, err = subscribeToNewTasksV2(s.AvsContractBindings.ServiceManager, internalChannel, s.logger)
+				if err != nil {
+					errorChannel <- err
+				}
+			case err := <-subFallback.Err():
+				s.logger.Warn("Error in fallback new task subscription", "err", err)
+				subFallback.Unsubscribe()
+				subFallback, err = subscribeToNewTasksV2(s.AvsContractBindings.ServiceManagerFallback, internalChannel, s.logger)
+				if err != nil {
+					errorChannel <- err
+				}
+			}
+		}
+	}()
+
+	return errorChannel, nil
+}
 
 func subscribeToNewTasks(
 	serviceManager *servicemanager.ContractAlignedLayerServiceManager,
@@ -154,6 +221,27 @@ func subscribeToNewTasks(
 
 	return nil, fmt.Errorf("failed to subscribe to new AlignedLayer tasks after %d retries", MaxRetries)
 }
+func subscribeToNewTasksV2(
+	serviceManager *servicemanager.ContractAlignedLayerServiceManager,
+	newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2,
+	logger sdklogging.Logger,
+) (event.Subscription, error) {
+	for i := 0; i < MaxRetries; i++ {
+		sub, err := serviceManager.WatchNewBatchV2(
+			&bind.WatchOpts{}, newTaskCreatedChan, nil,
+		)
+		if err != nil {
+			logger.Warn("Failed to subscribe to new AlignedLayer tasks", "err", err)
+			time.Sleep(RetryInterval)
+			continue
+		}
+
+		logger.Info("Subscribed to new AlignedLayer tasks")
+		return sub, nil
+	}
+
+	return nil, fmt.Errorf("failed to subscribe to new AlignedLayer tasks after %d retries", MaxRetries)
+}
 
 func (s *AvsSubscriber) processNewBatch(batch *servicemanager.ContractAlignedLayerServiceManagerNewBatch, batchesSet map[[32]byte]struct{}, newBatchMutex *sync.Mutex, newTaskCreatedChan chan<- *servicemanager.ContractAlignedLayerServiceManagerNewBatch) {
 	newBatchMutex.Lock()
@@ -263,7 +351,66 @@ func (s *AvsSubscriber) getLatestTaskFromEthereum() (*servicemanager.ContractAli
 	latestTask.BatchMerkleRoot = lastLog.Topics[1]
 
 	return &latestTask, nil
+}
+func (s *AvsSubscriber) getLatestTaskFromEthereumV2() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, error) {
+	latestBlock, err := s.AvsContractBindings.ethClient.BlockNumber(context.Background())
+	if err != nil {
+		latestBlock, err = s.AvsContractBindings.ethClientFallback.BlockNumber(context.Background())
+		if err != nil {
+			return nil, fmt.Errorf("failed to get latest block number: %w", err)
+		}
+	}
+
+	var fromBlock uint64
+
+	if latestBlock < BlockInterval {
+		fromBlock = 0
+	} else {
+		fromBlock = latestBlock - BlockInterval
+	}
+
+	alignedLayerServiceManagerABI, err := abi.JSON(strings.NewReader(servicemanager.ContractAlignedLayerServiceManagerMetaData.ABI))
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse ABI: %w", err)
+	}
+
+	// We just care about the NewBatch event
+	newBatchEvent := alignedLayerServiceManagerABI.Events["NewBatch"]
+	if newBatchEvent.ID == (ethcommon.Hash{}) {
+		return nil, fmt.Errorf("NewBatch event not found in ABI")
+	}
 
+	query := ethereum.FilterQuery{
+		FromBlock: big.NewInt(int64(fromBlock)),
+		ToBlock:   big.NewInt(int64(latestBlock)),
+		Addresses: []ethcommon.Address{s.AlignedLayerServiceManagerAddr},
+		Topics:    [][]ethcommon.Hash{{newBatchEvent.ID, {}}},
+	}
+
+	logs, err := s.AvsContractBindings.ethClient.FilterLogs(context.Background(), query)
+	if err != nil {
+		logs, err = s.AvsContractBindings.ethClientFallback.FilterLogs(context.Background(), query)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get logs: %w", err)
+		}
+	}
+
+	if len(logs) == 0 {
+		return nil, fmt.Errorf("no logs found")
+	}
+
+	lastLog := logs[len(logs)-1]
+
+	var latestTask servicemanager.ContractAlignedLayerServiceManagerNewBatchV2
+	err = alignedLayerServiceManagerABI.UnpackIntoInterface(&latestTask, "NewBatchV2", lastLog.Data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to unpack log data: %w", err)
+	}
+
+	// The second topic is the batch merkle root, as it is an indexed variable in the contract
+	latestTask.BatchMerkleRoot = lastLog.Topics[1]
+
+	return &latestTask, nil
 }
 
 func (s *AvsSubscriber) WaitForOneBlock(startBlock uint64) error {