refactor(operator): remove newBatchV2 logic

Author: JuArce

core/chainio/avs_subscriber.go

@@ -68,83 +68,6 @@ func NewAvsSubscriberFromConfig(baseConfig *config.BaseConfig) (*AvsSubscriber,
 	}, nil
 }
 
-func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, errorPairChannel chan ErrorPair) *ErrorPair {
-	// Create a new channel to receive new tasks
-	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2)
-
-	// Subscribe to new tasks
-	sub, errMain := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
-	if errMain != nil {
-		s.logger.Error(fmt.Sprintf("Fallback failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries), "errMain", fmt.Sprintf("%v", errMain))
-	}
-
-	subFallback, errFallback := SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
-	if errFallback != nil {
-		s.logger.Error(fmt.Sprintf("Fallback failed to subscribe to new AlignedLayer V3 tasks after %d retries", MaxRetries), "errFallback", fmt.Sprintf("%v", errFallback))
-	}
-
-	if errMain != nil && errFallback != nil {
-		s.logger.Error("Failed to subscribe to new AlignedLayer V2 tasks with both RPCs", "errMain", errMain, "errFallback", errFallback)
-		return &ErrorPair{ErrorMainRPC: errMain, ErrorFallbackRPC: errFallback}
-	}
-
-	s.logger.Info("Subscribed to new AlignedLayer V2 tasks")
-
-	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.getLatestNotRespondedTaskFromEthereumV2()
-				if err != nil {
-					s.logger.Debug("Failed to get latest task from blockchain", "err", err)
-					continue
-				}
-				if latestBatch != nil {
-					s.processNewBatchV2(latestBatch, batchesSet, newBatchMutex, newTaskCreatedChan)
-				}
-			}
-		}
-
-	}()
-
-	// Handle errors and resubscribe
-	go func() {
-		var errMain, errFallback error
-		var auxSub, auxSubFallback event.Subscription
-		for errMain == nil || errFallback == nil { //while one is active
-			select {
-			case err := <-sub.Err():
-				s.logger.Warn("Error in new task subscription of main connection", "err", err)
-
-				auxSub, errMain = SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManager, internalChannel, nil, retry.NetworkRetryParams())
-				if errMain == nil {
-					sub = auxSub // update the subscription only if it was successful
-					s.logger.Info("Main connection resubscribed to new task subscription")
-				}
-			case err := <-subFallback.Err():
-				s.logger.Warn("Error in new task subscription of fallback connection", "err", err)
-
-				auxSubFallback, errFallback = SubscribeToNewTasksV2Retryable(&bind.WatchOpts{}, s.AvsContractBindings.ServiceManagerFallback, internalChannel, nil, retry.NetworkRetryParams())
-				if errFallback == nil {
-					subFallback = auxSubFallback // update the subscription only if it was successful
-					s.logger.Info("Resubscribed to fallback new task subscription")
-				}
-			}
-		}
-		errorPairChannel <- ErrorPair{ErrorMainRPC: errMain, ErrorFallbackRPC: errFallback}
-	}()
-
-	return nil
-}
-
 func (s *AvsSubscriber) SubscribeToNewTasksV3(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3, errorPairChannel chan ErrorPair) *ErrorPair {
 	// Create a new channel to receive new tasks
 	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3)
@@ -224,32 +147,6 @@ func (s *AvsSubscriber) SubscribeToNewTasksV3(newTaskCreatedChan chan *servicema
 	return nil
 }
 
-func (s *AvsSubscriber) processNewBatchV2(batch *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, batchesSet map[[32]byte]struct{}, newBatchMutex *sync.Mutex, newTaskCreatedChan chan<- *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) {
-	newBatchMutex.Lock()
-	defer newBatchMutex.Unlock()
-
-	batchIdentifier := append(batch.BatchMerkleRoot[:], batch.SenderAddress[:]...)
-	var batchIdentifierHash = *(*[32]byte)(crypto.Keccak256(batchIdentifier))
-
-	if _, ok := batchesSet[batchIdentifierHash]; !ok {
-		s.logger.Info("Received new task",
-			"batchMerkleRoot", hex.EncodeToString(batch.BatchMerkleRoot[:]),
-			"senderAddress", hex.EncodeToString(batch.SenderAddress[:]),
-			"batchIdentifierHash", hex.EncodeToString(batchIdentifierHash[:]))
-
-		batchesSet[batchIdentifierHash] = struct{}{}
-		newTaskCreatedChan <- batch
-
-		// Remove the batch from the set after RemoveBatchFromSetInterval time
-		go func() {
-			time.Sleep(RemoveBatchFromSetInterval)
-			newBatchMutex.Lock()
-			delete(batchesSet, batchIdentifierHash)
-			newBatchMutex.Unlock()
-		}()
-	}
-}
-
 func (s *AvsSubscriber) processNewBatchV3(batch *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3, batchesSet map[[32]byte]struct{}, newBatchMutex *sync.Mutex, newTaskCreatedChan chan<- *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3) {
 	newBatchMutex.Lock()
 	defer newBatchMutex.Unlock()
@@ -276,56 +173,6 @@ func (s *AvsSubscriber) processNewBatchV3(batch *servicemanager.ContractAlignedL
 	}
 }
 
-// getLatestNotRespondedTaskFromEthereum queries the blockchain for the latest not responded task using the FilterNewBatch method.
-func (s *AvsSubscriber) getLatestNotRespondedTaskFromEthereumV2() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, error) {
-
-	latestBlock, err := s.BlockNumberRetryable(context.Background(), retry.NetworkRetryParams())
-	if err != nil {
-		return nil, err
-	}
-
-	var fromBlock uint64
-
-	if latestBlock < BlockInterval {
-		fromBlock = 0
-	} else {
-		fromBlock = latestBlock - BlockInterval
-	}
-
-	logs, err := s.FilterBatchV2Retryable(&bind.FilterOpts{Start: fromBlock, End: nil, Context: context.Background()}, nil, retry.NetworkRetryParams())
-	if err != nil {
-		return nil, err
-	}
-
-	var lastLog *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2
-
-	// Iterate over the logs until the end
-	for logs.Next() {
-		lastLog = logs.Event
-	}
-
-	if err := logs.Error(); err != nil {
-		return nil, err
-	}
-
-	if lastLog == nil {
-		return nil, nil
-	}
-
-	batchIdentifier := append(lastLog.BatchMerkleRoot[:], lastLog.SenderAddress[:]...)
-	batchIdentifierHash := *(*[32]byte)(crypto.Keccak256(batchIdentifier))
-	state, err := s.BatchesStateRetryable(nil, batchIdentifierHash, retry.NetworkRetryParams())
-	if err != nil {
-		return nil, err
-	}
-
-	if state.Responded {
-		return nil, nil
-	}
-
-	return lastLog, nil
-}
-
 // getLatestNotRespondedTaskFromEthereum queries the blockchain for the latest not responded task using the FilterNewBatch method.
 func (s *AvsSubscriber) getLatestNotRespondedTaskFromEthereumV3() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3, error) {
 	latestBlock, err := s.BlockNumberRetryable(context.Background(), retry.NetworkRetryParams())

core/chainio/retryable.go

@@ -126,30 +126,17 @@ func (s *AvsSubscriber) BlockNumberRetryable(ctx context.Context, config *retry.
 	return retry.RetryWithData(latestBlock_func, config)
 }
 
-/*
-FilterBatchV2Retryable
-Get NewBatchV2 logs from the AVS contract.
-- All errors are considered Transient Errors
-- Retry times (3 retries): 1 sec, 2 sec, 4 sec.
-*/
-func (s *AvsSubscriber) FilterBatchV2Retryable(opts *bind.FilterOpts, batchMerkleRoot [][32]byte, config *retry.RetryParams) (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV2Iterator, error) {
-	filterNewBatchV2_func := func() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV2Iterator, error) {
-		return s.AvsContractBindings.ServiceManager.FilterNewBatchV2(opts, batchMerkleRoot)
-	}
-	return retry.RetryWithData(filterNewBatchV2_func, config)
-}
-
 /*
 FilterBatchV3Retryable
 Get NewBatchV3 logs from the AVS contract.
 - All errors are considered Transient Errors
 - Retry times (3 retries): 1 sec, 2 sec, 4 sec.
 */
 func (s *AvsSubscriber) FilterBatchV3Retryable(opts *bind.FilterOpts, batchMerkleRoot [][32]byte, config *retry.RetryParams) (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3Iterator, error) {
-	filterNewBatchV2_func := func() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3Iterator, error) {
+	filterNewBatchV3_func := func() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV3Iterator, error) {
 		return s.AvsContractBindings.ServiceManager.FilterNewBatchV3(opts, batchMerkleRoot)
 	}
-	return retry.RetryWithData(filterNewBatchV2_func, config)
+	return retry.RetryWithData(filterNewBatchV3_func, config)
 }
 
 /*
@@ -193,26 +180,6 @@ func (s *AvsSubscriber) SubscribeNewHeadRetryable(ctx context.Context, c chan<-
 	return retry.RetryWithData(subscribeNewHead_func, config)
 }
 
-/*
-SubscribeToNewTasksV2Retryable
-Subscribe to NewBatchV2 logs from the AVS contract.
-- All errors are considered Transient Errors
-- Retry times (3 retries): 1 sec, 2 sec, 4 sec.
-*/
-func SubscribeToNewTasksV2Retryable(
-	opts *bind.WatchOpts,
-	serviceManager *servicemanager.ContractAlignedLayerServiceManager,
-	newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2,
-	batchMerkleRoot [][32]byte,
-	config *retry.RetryParams,
-) (event.Subscription, error) {
-	subscribe_func := func() (event.Subscription, error) {
-		log.Info().Msg("Subscribing to NewBatchV2")
-		return serviceManager.WatchNewBatchV2(opts, newTaskCreatedChan, batchMerkleRoot)
-	}
-	return retry.RetryWithData(subscribe_func, config)
-}
-
 /*
 SubscribeToNewTasksV3Retryable
 Subscribe to NewBatchV3 logs from the AVS contract.