fix: operator downtime

config-files/config-operator-1.yaml

@@ -31,6 +31,7 @@ operator:
   enable_metrics: true
   metrics_ip_port_address: localhost:9092
   max_batch_size: 268435456 # 256 MiB
+  last_processed_batch_filepath: "operator-1.json"
 
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"

config-files/config-operator-2.yaml

@@ -29,6 +29,7 @@ operator:
   staker_opt_out_window_blocks: 0
   metadata_url: "https://yetanotherco.github.io/operator_metadata/metadata.json"
   max_batch_size: 268435456 # 256 MiB
+  last_processed_batch_filepath: "operator-2.json"
 
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"

config-files/config-operator-3.yaml

@@ -29,6 +29,7 @@ operator:
   staker_opt_out_window_blocks: 0
   metadata_url: "https://yetanotherco.github.io/operator_metadata/metadata.json"
   max_batch_size: 268435456 # 256 MiB
+  last_processed_batch_filepath: "operator-3.json"
 
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"

config-files/config-operator.yaml

@@ -31,3 +31,4 @@ operator:
   enable_metrics: true
   metrics_ip_port_address: localhost:9092
   max_batch_size: 268435456 # 256 MiB
+  last_processed_batch_filepath: "operator.json"

config-files/config.yaml

@@ -48,6 +48,7 @@ operator:
   enable_metrics: true
   metrics_ip_port_address: localhost:9092
   max_batch_size: 268435456 # 256 MiB
+  last_processed_batch_filepath: "operator.json"
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"
 private_key_store_path: config-files/anvil.ecdsa.key.json

core/chainio/avs_reader.go

@@ -1,8 +1,17 @@
 package chainio
 
 import (
+	"context"
+	"fmt"
+	"math/big"
+	"strings"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
-	gethcommon "github.com/ethereum/go-ethereum/common"
+	ethcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	servicemanager "github.com/yetanotherco/aligned_layer/contracts/bindings/AlignedLayerServiceManager"
 	contractERC20Mock "github.com/yetanotherco/aligned_layer/contracts/bindings/ERC20Mock"
 	"github.com/yetanotherco/aligned_layer/core/config"
 
@@ -13,8 +22,9 @@ import (
 
 type AvsReader struct {
 	*sdkavsregistry.ChainReader
-	AvsContractBindings *AvsServiceBindings
-	logger              logging.Logger
+	AvsContractBindings            *AvsServiceBindings
+	AlignedLayerServiceManagerAddr ethcommon.Address
+	logger                         logging.Logger
 }
 
 func NewAvsReaderFromConfig(baseConfig *config.BaseConfig, ecdsaConfig *config.EcdsaConfig) (*AvsReader, error) {
@@ -41,13 +51,14 @@ func NewAvsReaderFromConfig(baseConfig *config.BaseConfig, ecdsaConfig *config.E
 	}
 
 	return &AvsReader{
-		ChainReader:         chainReader,
-		AvsContractBindings: avsServiceBindings,
-		logger:              baseConfig.Logger,
+		ChainReader:                    chainReader,
+		AvsContractBindings:            avsServiceBindings,
+		AlignedLayerServiceManagerAddr: baseConfig.AlignedLayerDeploymentConfig.AlignedLayerServiceManagerAddr,
+		logger:                         baseConfig.Logger,
 	}, nil
 }
 
-func (r *AvsReader) GetErc20Mock(tokenAddr gethcommon.Address) (*contractERC20Mock.ContractERC20Mock, error) {
+func (r *AvsReader) GetErc20Mock(tokenAddr ethcommon.Address) (*contractERC20Mock.ContractERC20Mock, error) {
 	erc20Mock, err := contractERC20Mock.NewContractERC20Mock(tokenAddr, r.AvsContractBindings.ethClient)
 	if err != nil {
 		// Retry with fallback client
@@ -59,6 +70,73 @@ func (r *AvsReader) GetErc20Mock(tokenAddr gethcommon.Address) (*contractERC20Mo
 	return erc20Mock, nil
 }
 
-func (r *AvsReader) IsOperatorRegistered(address gethcommon.Address) (bool, error) {
+func (r *AvsReader) IsOperatorRegistered(address ethcommon.Address) (bool, error) {
 	return r.ChainReader.IsOperatorRegistered(&bind.CallOpts{}, address)
 }
+
+// Returns all the "NewBatchV3" logs that have not been responded starting from the given block number
+func (r *AvsReader) GetNotRespondedTasksFrom(fromBlock uint64) ([]servicemanager.ContractAlignedLayerServiceManagerNewBatchV3, error) {
+	latestBlock, err := r.AvsContractBindings.ethClient.BlockNumber(context.Background())
+	if err != nil {
+		latestBlock, err = r.AvsContractBindings.ethClientFallback.BlockNumber(context.Background())
+		if err != nil {
+			return nil, fmt.Errorf("failed to get latest block number: %w", err)
+		}
+	}
+
+	alignedLayerServiceManagerABI, err := abi.JSON(strings.NewReader(servicemanager.ContractAlignedLayerServiceManagerMetaData.ABI))
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse ABI: %w", err)
+	}
+
+	newBatchEvent := alignedLayerServiceManagerABI.Events["NewBatchV3"]
+	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{r.AlignedLayerServiceManagerAddr},
+		Topics:    [][]ethcommon.Hash{{newBatchEvent.ID, {}}},
+	}
+
+	logs, err := r.AvsContractBindings.ethClient.FilterLogs(context.Background(), query)
+	if err != nil {
+		logs, err = r.AvsContractBindings.ethClientFallback.FilterLogs(context.Background(), query)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get logs: %w", err)
+		}
+	}
+
+	var tasks []servicemanager.ContractAlignedLayerServiceManagerNewBatchV3
+
+	for _, logEntry := range logs {
+		var task servicemanager.ContractAlignedLayerServiceManagerNewBatchV3
+
+		err := alignedLayerServiceManagerABI.UnpackIntoInterface(&task, "NewBatchV3", logEntry.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
+		task.BatchMerkleRoot = logEntry.Topics[1]
+
+		// now check if its finalized or not before appending
+		batchIdentifier := append(task.BatchMerkleRoot[:], task.SenderAddress[:]...)
+		batchIdentifierHash := *(*[32]byte)(crypto.Keccak256(batchIdentifier))
+		state, err := r.AvsContractBindings.ServiceManager.ContractAlignedLayerServiceManagerCaller.BatchesState(nil, batchIdentifierHash)
+
+		if err != nil {
+			return nil, fmt.Errorf("err while getting batch state: %w", err)
+		}
+
+		// append the task if not responded yet
+		if !state.Responded {
+			tasks = append(tasks, task)
+		}
+
+	}
+
+	return tasks, nil
+}

core/chainio/avs_subscriber.go

@@ -4,14 +4,15 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"github.com/ethereum/go-ethereum"
-	"github.com/ethereum/go-ethereum/accounts/abi"
-	ethcommon "github.com/ethereum/go-ethereum/common"
 	"math/big"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	ethcommon "github.com/ethereum/go-ethereum/common"
+
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/event"
@@ -102,7 +103,9 @@ func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicema
 					s.logger.Debug("Failed to get latest task from blockchain", "err", err)
 					continue
 				}
-				s.processNewBatchV2(latestBatch, batchesSet, newBatchMutex, newTaskCreatedChan)
+				if latestBatch != nil {
+					s.processNewBatchV2(latestBatch, batchesSet, newBatchMutex, newTaskCreatedChan)
+				}
 			}
 		}
 
@@ -132,7 +135,7 @@ func (s *AvsSubscriber) SubscribeToNewTasksV2(newTaskCreatedChan chan *servicema
 
 	return errorChannel, nil
 }
-	
+
 func (s *AvsSubscriber) SubscribeToNewTasksV3(newTaskCreatedChan chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3) (chan error, error) {
 	// Create a new channel to receive new tasks
 	internalChannel := make(chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3)
@@ -170,7 +173,9 @@ func (s *AvsSubscriber) SubscribeToNewTasksV3(newTaskCreatedChan chan *servicema
 					s.logger.Debug("Failed to get latest task from blockchain", "err", err)
 					continue
 				}
-				s.processNewBatchV3(latestBatch, batchesSet, newBatchMutex, newTaskCreatedChan)
+				if latestBatch != nil {
+					s.processNewBatchV3(latestBatch, batchesSet, newBatchMutex, newTaskCreatedChan)
+				}
 			}
 		}
 
@@ -254,9 +259,9 @@ func (s *AvsSubscriber) processNewBatchV2(batch *servicemanager.ContractAlignedL
 
 	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[:]),)
+			"batchMerkleRoot", hex.EncodeToString(batch.BatchMerkleRoot[:]),
+			"senderAddress", hex.EncodeToString(batch.SenderAddress[:]),
+			"batchIdentifierHash", hex.EncodeToString(batchIdentifierHash[:]))
 
 		batchesSet[batchIdentifierHash] = struct{}{}
 		newTaskCreatedChan <- batch
@@ -280,9 +285,9 @@ func (s *AvsSubscriber) processNewBatchV3(batch *servicemanager.ContractAlignedL
 
 	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[:]),)
+			"batchMerkleRoot", hex.EncodeToString(batch.BatchMerkleRoot[:]),
+			"senderAddress", hex.EncodeToString(batch.SenderAddress[:]),
+			"batchIdentifierHash", hex.EncodeToString(batchIdentifierHash[:]))
 
 		batchesSet[batchIdentifierHash] = struct{}{}
 		newTaskCreatedChan <- batch
@@ -358,6 +363,19 @@ func (s *AvsSubscriber) getLatestTaskFromEthereumV2() (*servicemanager.ContractA
 	// The second topic is the batch merkle root, as it is an indexed variable in the contract
 	latestTask.BatchMerkleRoot = lastLog.Topics[1]
 
+	// return the task if has not been responded only
+	batchIdentifier := append(latestTask.BatchMerkleRoot[:], latestTask.SenderAddress[:]...)
+	batchIdentifierHash := *(*[32]byte)(crypto.Keccak256(batchIdentifier))
+	state, err := s.AvsContractBindings.ServiceManager.ContractAlignedLayerServiceManagerCaller.BatchesState(nil, batchIdentifierHash)
+
+	if err != nil {
+		return nil, fmt.Errorf("err while getting batch state: %w", err)
+	}
+
+	if state.Responded {
+		return nil, nil
+	}
+
 	return &latestTask, nil
 }
 
@@ -422,6 +440,19 @@ func (s *AvsSubscriber) getLatestTaskFromEthereumV3() (*servicemanager.ContractA
 	// The second topic is the batch merkle root, as it is an indexed variable in the contract
 	latestTask.BatchMerkleRoot = lastLog.Topics[1]
 
+	// return the task if has not been responded only
+	batchIdentifier := append(latestTask.BatchMerkleRoot[:], latestTask.SenderAddress[:]...)
+	batchIdentifierHash := *(*[32]byte)(crypto.Keccak256(batchIdentifier))
+	state, err := s.AvsContractBindings.ServiceManager.ContractAlignedLayerServiceManagerCaller.BatchesState(nil, batchIdentifierHash)
+
+	if err != nil {
+		return nil, fmt.Errorf("err while getting batch state: %w", err)
+	}
+
+	if state.Responded {
+		return nil, nil
+	}
+
 	return &latestTask, nil
 }
 

core/config/operator.go

@@ -27,6 +27,7 @@ type OperatorConfig struct {
 		EnableMetrics                 bool
 		MetricsIpPortAddress          string
 		MaxBatchSize                  int64
+		LastProcessedBatchFilePath    string
 	}
 }
 
@@ -43,6 +44,7 @@ type OperatorConfigFromYaml struct {
 		EnableMetrics                 bool           `yaml:"enable_metrics"`
 		MetricsIpPortAddress          string         `yaml:"metrics_ip_port_address"`
 		MaxBatchSize                  int64          `yaml:"max_batch_size"`
+		LastProcessedBatchFilePath    string         `yaml:"last_processed_batch_filepath"`
 	} `yaml:"operator"`
 	EcdsaConfigFromYaml EcdsaConfigFromYaml `yaml:"ecdsa"`
 	BlsConfigFromYaml   BlsConfigFromYaml   `yaml:"bls"`
@@ -92,6 +94,7 @@ func NewOperatorConfig(configFilePath string) *OperatorConfig {
 			EnableMetrics                 bool
 			MetricsIpPortAddress          string
 			MaxBatchSize                  int64
+			LastProcessedBatchFilePath    string
 		}(operatorConfigFromYaml.Operator),
 	}
 }