refactor: 'V2' functionalities to aggregator and operator (yetanotherco#875)

uri-99 · JuArce · taturosati · web-flow · commit 81b290dbed7a · 2024-09-05T10:43:02.000-03:00
Co-authored-by: JuArce &lt;52429267+JuArce@users.noreply.github.com&gt;
Co-authored-by: taturosati &lt;“taturosati@users.noreply.github.com”&gt;
diff --git a/Makefile b/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
diff --git a/aggregator/internal/pkg/aggregator.go b/aggregator/internal/pkg/aggregator.go
@@ -42,7 +42,7 @@ type BatchData struct {
 
 type Aggregator struct {
 	AggregatorConfig      *config.AggregatorConfig
-	NewBatchChan          chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch
+	NewBatchChan          chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2
 	avsReader             *chainio.AvsReader
 	avsSubscriber         *chainio.AvsSubscriber
 	avsWriter             *chainio.AvsWriter
@@ -91,7 +91,7 @@ type Aggregator struct {
 }
 
 func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error) {
-	newBatchChan := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch)
+	newBatchChan := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2)
 
 	avsReader, err := chainio.NewAvsReaderFromConfig(aggregatorConfig.BaseConfig, aggregatorConfig.EcdsaConfig)
 	if err != nil {
@@ -333,7 +333,6 @@ func (agg *Aggregator) sendAggregatedResponse(batchMerkleRoot [32]byte, senderAd
 }
 
 func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]byte, taskCreatedBlock uint32) {
-
 	batchIdentifier := append(batchMerkleRoot[:], senderAddress[:]...)
 	var batchIdentifierHash = *(*[32]byte)(crypto.Keccak256(batchIdentifier))
 
diff --git a/aggregator/internal/pkg/server.go b/aggregator/internal/pkg/server.go
@@ -48,7 +48,7 @@ func (agg *Aggregator) ServeOperators() error {
 // Returns:
 //   - 0: Success
 //   - 1: Error
-func (agg *Aggregator) ProcessOperatorSignedTaskResponse(signedTaskResponse *types.SignedTaskResponse, reply *uint8) error {
+func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *types.SignedTaskResponse, reply *uint8) error {
 	agg.AggregatorConfig.BaseConfig.Logger.Info("New task response",
 		"BatchMerkleRoot", "0x"+hex.EncodeToString(signedTaskResponse.BatchMerkleRoot[:]),
 		"SenderAddress", "0x"+hex.EncodeToString(signedTaskResponse.SenderAddress[:]),
diff --git a/aggregator/internal/pkg/subscriber.go b/aggregator/internal/pkg/subscriber.go
@@ -10,16 +10,15 @@ func (agg *Aggregator) SubscribeToNewTasks() error {
 		select {
 		case err := <-agg.taskSubscriber:
 			agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to subscribe to new tasks", "err", err)
-
 			err = agg.subscribeToNewTasks()
 			if err != nil {
 				return err
 			}
 		case newBatch := <-agg.NewBatchChan:
+			agg.AggregatorConfig.BaseConfig.Logger.Info("Adding new task")
 			agg.AddNewTask(newBatch.BatchMerkleRoot, newBatch.SenderAddress, newBatch.TaskCreatedBlock)
 		}
 	}
-
 }
 
 func (agg *Aggregator) subscribeToNewTasks() error {
diff --git a/config-files/config-operator.yaml b/config-files/config-operator.yaml
@@ -22,7 +22,7 @@ bls:
 ## Operator Configurations
 operator:
   aggregator_rpc_server_ip_port_address: aggregator.alignedlayer.com:8090
-  operator_tracker_ip_port_address: holesky.tracker.alignedlayer.com
+  operator_tracker_ip_port_address: https://holesky.tracker.alignedlayer.com
   address: "<operator_address>"
   earnings_receiver_address: "<earnings_receiver_address>" #Can be the same as the operator.
   delegation_approver_address: "0x0000000000000000000000000000000000000000"
diff --git a/contracts/bindings/AlignedLayerServiceManager/binding.go b/contracts/bindings/AlignedLayerServiceManager/binding.go
diff --git a/contracts/scripts/anvil/state/alignedlayer-deployed-anvil-state.json b/contracts/scripts/anvil/state/alignedlayer-deployed-anvil-state.json
diff --git a/contracts/src/core/AlignedLayerServiceManager.sol b/contracts/src/core/AlignedLayerServiceManager.sol
@@ -51,12 +51,12 @@ contract AlignedLayerServiceManager is
         bytes32 batchMerkleRoot,
         string calldata batchDataPointer
     ) external payable {
-        bytes32 batchIdentifierHash = keccak256(
+        bytes32 batchIdentifier = keccak256(
             abi.encodePacked(batchMerkleRoot, msg.sender)
         );
 
-        if (batchesState[batchIdentifierHash].taskCreatedBlock != 0) {
-            revert BatchAlreadySubmitted(batchIdentifierHash);
+        if (batchesState[batchIdentifier].taskCreatedBlock != 0) {
+            revert BatchAlreadySubmitted(batchIdentifier);
         }
 
         if (msg.value > 0) {
@@ -76,17 +76,17 @@ contract AlignedLayerServiceManager is
         batchState.taskCreatedBlock = uint32(block.number);
         batchState.responded = false;
 
-        batchesState[batchIdentifierHash] = batchState;
+        batchesState[batchIdentifier] = batchState;
 
-        emit NewBatch(
+        emit NewBatchV2(
             batchMerkleRoot,
             msg.sender,
             uint32(block.number),
             batchDataPointer
         );
     }
 
-    function respondToTask(
+    function respondToTaskV2(
         // (batchMerkleRoot,senderAddress) is signed as a way to verify the batch was right
         bytes32 batchMerkleRoot,
         address senderAddress,
@@ -174,15 +174,20 @@ contract AlignedLayerServiceManager is
         uint256 verificationDataBatchIndex,
         address senderAddress
     ) external view returns (bool) {
-        bytes32 batchIdentifierHash = keccak256(
-            abi.encodePacked(batchMerkleRoot, senderAddress)
-        );
+        bytes32 batchIdentifier;
+        if (senderAddress == address(0)) {
+            batchIdentifier = batchMerkleRoot;
+        } else {
+            batchIdentifier = keccak256(
+                abi.encodePacked(batchMerkleRoot, senderAddress)
+            );
+        }
 
-        if (batchesState[batchIdentifierHash].taskCreatedBlock == 0) {
+        if (batchesState[batchIdentifier].taskCreatedBlock == 0) {
             return false;
         }
 
-        if (!batchesState[batchIdentifierHash].responded) {
+        if (!batchesState[batchIdentifier].responded) {
             return false;
         }
 
@@ -198,12 +203,34 @@ contract AlignedLayerServiceManager is
         return
             Merkle.verifyInclusionKeccak(
                 merkleProof,
-                batchIdentifierHash,
+                batchMerkleRoot,
                 hashedLeaf,
                 verificationDataBatchIndex
             );
     }
 
+    // Old function signature for backwards compatibility
+    function verifyBatchInclusion(
+        bytes32 proofCommitment,
+        bytes32 pubInputCommitment,
+        bytes32 provingSystemAuxDataCommitment,
+        bytes20 proofGeneratorAddr,
+        bytes32 batchMerkleRoot,
+        bytes memory merkleProof,
+        uint256 verificationDataBatchIndex
+    ) external view returns (bool) {
+        return this.verifyBatchInclusion(
+            proofCommitment,
+            pubInputCommitment,
+            provingSystemAuxDataCommitment,
+            proofGeneratorAddr,
+            batchMerkleRoot,
+            merkleProof,
+            verificationDataBatchIndex,
+            address(0)
+        );
+    }
+
     function withdraw(uint256 amount) external {
         if (batchersBalances[msg.sender] < amount) {
             revert InsufficientFunds(
diff --git a/contracts/src/core/IAlignedLayerServiceManager.sol b/contracts/src/core/IAlignedLayerServiceManager.sol
@@ -5,7 +5,7 @@ import {IBLSSignatureChecker} from "eigenlayer-middleware/interfaces/IBLSSignatu
 
 interface IAlignedLayerServiceManager {
     // EVENTS
-    event NewBatch(
+    event NewBatchV2(
         bytes32 indexed batchMerkleRoot,
         address senderAddress,
         uint32 taskCreatedBlock,
@@ -33,7 +33,7 @@ interface IAlignedLayerServiceManager {
         string calldata batchDataPointer
     ) external payable;
 
-    function respondToTask(
+    function respondToTaskV2(
         bytes32 batchMerkleRoot,
         address senderAddress,
         IBLSSignatureChecker.NonSignerStakesAndSignature
diff --git a/contracts/test/AlignedLayerServiceManager.t.sol b/contracts/test/AlignedLayerServiceManager.t.sol
@@ -16,7 +16,7 @@ contract AlignedLayerServiceManagerTest is BLSMockAVSDeployer {
 
     using stdStorage for StdStorage;
 
-    event NewBatch(
+    event NewBatchV2(
         bytes32 indexed batchMerkleRoot,
         address senderAddress,
         uint32 taskCreatedBlock,
@@ -62,7 +62,7 @@ contract AlignedLayerServiceManagerTest is BLSMockAVSDeployer {
         address(alignedLayerServiceManager).call{value: 0.1 ether}("");
 
         vm.expectEmit(true, true, true, true);
-        emit NewBatch(batchMerkleRoot, batcher, uint32(block.number), batchDataPointer);
+        emit NewBatchV2(batchMerkleRoot, batcher, uint32(block.number), batchDataPointer);
 
         vm.prank(batcher);
         alignedLayerServiceManager.createNewTask(
diff --git a/core/chainio/avs_subscriber.go b/core/chainio/avs_subscriber.go
@@ -64,10 +64,10 @@ func NewAvsSubscriberFromConfig(baseConfig *config.BaseConfig) (*AvsSubscriber,
 		logger:                         baseConfig.Logger,
 	}, nil
 }
-
-func (s *AvsSubscriber) SubscribeToNewTasks(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch) (chan error, error) {
+	
+func (s *AvsSubscriber) SubscribeToNewTasks(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) (chan error, error) {
 	// Create a new channel to receive new tasks
-	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch)
+	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2)
 
 	// Subscribe to new tasks
 	sub, err := subscribeToNewTasks(s.AvsContractBindings.ServiceManager, internalChannel, s.logger)
@@ -135,11 +135,11 @@ func (s *AvsSubscriber) SubscribeToNewTasks(newTaskCreatedChan chan *servicemana
 
 func subscribeToNewTasks(
 	serviceManager *servicemanager.ContractAlignedLayerServiceManager,
-	newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch,
+	newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2,
 	logger sdklogging.Logger,
 ) (event.Subscription, error) {
 	for i := 0; i < MaxRetries; i++ {
-		sub, err := serviceManager.WatchNewBatch(
+		sub, err := serviceManager.WatchNewBatchV2(
 			&bind.WatchOpts{}, newTaskCreatedChan, nil,
 		)
 		if err != nil {
@@ -152,10 +152,10 @@ func subscribeToNewTasks(
 		return sub, nil
 	}
 
-	return nil, fmt.Errorf("Failed to subscribe to new AlignedLayer tasks after %d retries", MaxRetries)
+	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) {
+func (s *AvsSubscriber) processNewBatch(batch *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, batchesSet map[[32]byte]struct{}, newBatchMutex *sync.Mutex, newTaskCreatedChan chan<- *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) {
 	newBatchMutex.Lock()
 	defer newBatchMutex.Unlock()
 
@@ -184,7 +184,7 @@ func (s *AvsSubscriber) processNewBatch(batch *servicemanager.ContractAlignedLay
 // getLatestTaskFromEthereum queries the blockchain for the latest task using the FilterLogs method.
 // The alternative to this is using the FilterNewBatch method from the contract's filterer, but it requires
 // to iterate over all the logs, which is not efficient and not needed since we only need the latest task.
-func (s *AvsSubscriber) getLatestTaskFromEthereum() (*servicemanager.ContractAlignedLayerServiceManagerNewBatch, error) {
+func (s *AvsSubscriber) getLatestTaskFromEthereum() (*servicemanager.ContractAlignedLayerServiceManagerNewBatchV2, error) {
 	latestBlock, err := s.AvsContractBindings.ethClient.BlockNumber(context.Background())
 	if err != nil {
 		latestBlock, err = s.AvsContractBindings.ethClientFallback.BlockNumber(context.Background())
@@ -233,7 +233,7 @@ func (s *AvsSubscriber) getLatestTaskFromEthereum() (*servicemanager.ContractAli
 
 	lastLog := logs[len(logs)-1]
 
-	var latestTask servicemanager.ContractAlignedLayerServiceManagerNewBatch
+	var latestTask servicemanager.ContractAlignedLayerServiceManagerNewBatchV2
 	err = alignedLayerServiceManagerABI.UnpackIntoInterface(&latestTask, "NewBatch", lastLog.Data)
 	if err != nil {
 		return nil, fmt.Errorf("failed to unpack log data: %w", err)
@@ -243,7 +243,6 @@ func (s *AvsSubscriber) getLatestTaskFromEthereum() (*servicemanager.ContractAli
 	latestTask.BatchMerkleRoot = lastLog.Topics[1]
 
 	return &latestTask, nil
-
 }
 
 func (s *AvsSubscriber) WaitForOneBlock(startBlock uint64) error {
diff --git a/core/chainio/avs_writer.go b/core/chainio/avs_writer.go
@@ -89,10 +89,10 @@ func (w *AvsWriter) SendTask(context context.Context, batchMerkleRoot [32]byte,
 func (w *AvsWriter) SendAggregatedResponse(batchMerkleRoot [32]byte, senderAddress [20]byte, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature) (*common.Hash, error) {
 	txOpts := *w.Signer.GetTxOpts()
 	txOpts.NoSend = true // simulate the transaction
-	tx, err := w.AvsContractBindings.ServiceManager.RespondToTask(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+	tx, err := w.AvsContractBindings.ServiceManager.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
 	if err != nil {
 		// Retry with fallback
-		tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTask(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+		tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
 		if err != nil {
 			return nil, err
 		}
@@ -101,10 +101,10 @@ func (w *AvsWriter) SendAggregatedResponse(batchMerkleRoot [32]byte, senderAddre
 	// Send the transaction
 	txOpts.NoSend = false
 	txOpts.GasLimit = tx.Gas() * 110 / 100 // Add 10% to the gas limit
-	tx, err = w.AvsContractBindings.ServiceManager.RespondToTask(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+	tx, err = w.AvsContractBindings.ServiceManager.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
 	if err != nil {
 		// Retry with fallback
-		tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTask(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+		tx, err = w.AvsContractBindings.ServiceManagerFallback.RespondToTaskV2(&txOpts, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
 		if err != nil {
 			return nil, err
 		}
diff --git a/explorer/lib/abi/AlignedLayerServiceManager.json b/explorer/lib/abi/AlignedLayerServiceManager.json
diff --git a/explorer/lib/explorer/contract_managers/aligned_layer_service_manager.ex b/explorer/lib/explorer/contract_managers/aligned_layer_service_manager.ex
@@ -56,7 +56,7 @@ defmodule AlignedLayerServiceManager do
 
   def get_new_batch_events(%{fromBlock: fromBlock, toBlock: toBlock}) do
     events =
-      AlignedLayerServiceManager.EventFilters.new_batch(nil)
+      AlignedLayerServiceManager.EventFilters.new_batch_v2(nil)
       |> Ethers.get_logs(fromBlock: fromBlock, toBlock: toBlock)
 
     case events do
diff --git a/operator/pkg/operator.go b/operator/pkg/operator.go
@@ -45,7 +45,7 @@ type Operator struct {
 	KeyPair            *bls.KeyPair
 	OperatorId         eigentypes.OperatorId
 	avsSubscriber      chainio.AvsSubscriber
-	NewTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch
+	NewTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2
 	Logger             logging.Logger
 	aggRpcClient       AggregatorRpcClient
 	metricsReg         *prometheus.Registry
@@ -91,11 +91,11 @@ func NewOperatorFromConfig(configuration config.OperatorConfig) (*Operator, erro
 	if err != nil {
 		log.Fatalf("Could not create AVS subscriber")
 	}
-	newTaskCreatedChan := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatch)
+	newTaskCreatedChan := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2)
 
 	rpcClient, err := NewAggregatorRpcClient(configuration.Operator.AggregatorServerIpPortAddress, logger)
 	if err != nil {
-		return nil, fmt.Errorf("Could not create RPC client: %s. Is aggregator running?", err)
+		return nil, fmt.Errorf("could not create RPC client: %s. Is aggregator running?", err)
 	}
 
 	operatorId := eigentypes.OperatorIdFromKeyPair(configuration.BlsConfig.KeyPair)
@@ -153,20 +153,21 @@ func (o *Operator) Start(ctx context.Context) error {
 				o.Logger.Fatal("Could not subscribe to new tasks")
 			}
 		case newBatchLog := <-o.NewTaskCreatedChan:
-			go o.HandleNewBatchLog(newBatchLog)
+			go o.handleNewBatchLog(newBatchLog)
 		}
 	}
 }
 
-func (o *Operator) HandleNewBatchLog(newBatchLog *servicemanager.ContractAlignedLayerServiceManagerNewBatch) {
+func (o *Operator) handleNewBatchLog(newBatchLog *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) {
+	o.Logger.Infof("Received new batch log")
 	err := o.ProcessNewBatchLog(newBatchLog)
 	if err != nil {
 		o.Logger.Infof("batch %x did not verify. Err: %v", newBatchLog.BatchMerkleRoot, err)
 		return
 	}
+
 	batchIdentifier := append(newBatchLog.BatchMerkleRoot[:], newBatchLog.SenderAddress[:]...)
 	var batchIdentifierHash = *(*[32]byte)(crypto.Keccak256(batchIdentifier))
-
 	responseSignature := o.SignTaskResponse(batchIdentifierHash)
 	o.Logger.Debugf("responseSignature about to send: %x", responseSignature)
 
@@ -177,19 +178,18 @@ func (o *Operator) HandleNewBatchLog(newBatchLog *servicemanager.ContractAligned
 		BlsSignature:        *responseSignature,
 		OperatorId:          o.OperatorId,
 	}
-
 	o.Logger.Infof("Signed Task Response to send: BatchIdentifierHash=%s, BatchMerkleRoot=%s, SenderAddress=%s",
 		hex.EncodeToString(signedTaskResponse.BatchIdentifierHash[:]),
 		hex.EncodeToString(signedTaskResponse.BatchMerkleRoot[:]),
 		hex.EncodeToString(signedTaskResponse.SenderAddress[:]),
 	)
-
+	
 	o.aggRpcClient.SendSignedTaskResponseToAggregator(&signedTaskResponse)
 }
 
 // Takes a NewTaskCreatedLog struct as input and returns a TaskResponseHeader struct.
 // The TaskResponseHeader struct is the struct that is signed and sent to the contract as a task response.
-func (o *Operator) ProcessNewBatchLog(newBatchLog *servicemanager.ContractAlignedLayerServiceManagerNewBatch) error {
+func (o *Operator) ProcessNewBatchLog(newBatchLog *servicemanager.ContractAlignedLayerServiceManagerNewBatchV2) error {
 
 	o.Logger.Info("Received new batch with proofs to verify",
 		"batch merkle root", "0x"+hex.EncodeToString(newBatchLog.BatchMerkleRoot[:]),
diff --git a/operator/pkg/rpc_client.go b/operator/pkg/rpc_client.go
@@ -39,7 +39,7 @@ func NewAggregatorRpcClient(aggregatorIpPortAddr string, logger logging.Logger)
 func (c *AggregatorRpcClient) SendSignedTaskResponseToAggregator(signedTaskResponse *types.SignedTaskResponse) {
 	var reply uint8
 	for retries := 0; retries < MaxRetries; retries++ {
-		err := c.rpcClient.Call("Aggregator.ProcessOperatorSignedTaskResponse", signedTaskResponse, &reply)
+		err := c.rpcClient.Call("Aggregator.ProcessOperatorSignedTaskResponseV2", signedTaskResponse, &reply)
 		if err != nil {
 			c.logger.Error("Received error from aggregator", "err", err)
 			if errors.Is(err, rpc.ErrShutdown) {
@@ -53,7 +53,7 @@ func (c *AggregatorRpcClient) SendSignedTaskResponseToAggregator(signedTaskRespo
 					c.logger.Info("Reconnected to aggregator")
 				}
 			} else {
-				c.logger.Infof("Received error from aggregator: %s. Retrying ProcessOperatorSignedTaskResponse RPC call...", err)
+				c.logger.Infof("Received error from aggregator: %s. Retrying ProcessOperatorSignedTaskResponseV2 RPC call...", err)
 				time.Sleep(RetryInterval)
 			}
 		} else {

Original file line number	Diff line number	Diff line change
`@@ -10,16 +10,15 @@ func (agg *Aggregator) SubscribeToNewTasks() error {`
`10`	`10`	`select {`
`11`	`11`	`case err := <-agg.taskSubscriber:`
`12`	`12`	`agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to subscribe to new tasks", "err", err)`
`13`		`-`
`14`	`13`	`err = agg.subscribeToNewTasks()`
`15`	`14`	`if err != nil {`
`16`	`15`	`return err`
`17`	`16`	`}`
`18`	`17`	`case newBatch := <-agg.NewBatchChan:`
	`18`	`+ agg.AggregatorConfig.BaseConfig.Logger.Info("Adding new task")`
`19`	`19`	`agg.AddNewTask(newBatch.BatchMerkleRoot, newBatch.SenderAddress, newBatch.TaskCreatedBlock)`
`20`	`20`	`}`
`21`	`21`	`}`
`22`		`-`
`23`	`22`	`}`
`24`	`23`
`25`	`24`	`func (agg *Aggregator) subscribeToNewTasks() error {`