Merge branch 'main' into mina

Author: gabrielbosio

.github/workflows/build.yml .github/workflows/build-go.yml.github/workflows/build.yml renamed to .github/workflows/build-go.yml

@@ -0,0 +1,44 @@
+name: build-rust-projects
+
+on:
+  merge_group:
+  push:
+    branches: [main]
+  pull_request:
+    branches: ["*"]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          components: rustfmt, clippy
+          override: true
+      - name: Cache Rust dependencies
+        uses: actions/cache@v3
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            batcher/target
+          key: ${{ runner.os }}-rust-${{ hashFiles('**/Cargo.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-rust-
+      - name: Check formatting of Rust projects
+        run: |
+          cd batcher
+          cargo fmt --all -- --check
+      - name: Run Clippy
+        run: |
+          cd batcher
+          cargo clippy --all -- -D warnings
+      - name: Build Rust projects
+        run: |
+          cd batcher
+          cargo build --all

.github/workflows/build-rust.yml

@@ -10,5 +10,6 @@ scripts/test_files/gnark_groth16_bn254_infinite_script/infinite_proofs/**
 batcher/aligned/batch_inclusion_responses/*
 **/aligned_verification_data
 **/broadcast
+volume
 
 nonce_*.bin

.gitignore

@@ -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.3.0
+OPERATOR_VERSION=v0.4.0
 
 ifeq ($(OS),Linux)
 	BUILD_ALL_FFI = $(MAKE) build_all_ffi_linux
@@ -173,10 +173,6 @@ operator_deposit_and_register: operator_deposit_into_strategy operator_register_
 
 operator_full_registration: operator_get_eth operator_register_with_eigen_layer operator_mint_mock_tokens operator_deposit_into_mock_strategy operator_whitelist_devnet operator_register_with_aligned_layer
 
-operator_start_docker:
-	@echo "Starting Operator..."
-	@docker-compose -f operator/docker/compose.yaml up
-
 __BATCHER__:
 
 BURST_SIZE=5
@@ -191,6 +187,11 @@ batcher_start: ./batcher/aligned-batcher/.env user_fund_payment_service
 	@echo "Starting Batcher..."
 	@cargo +nightly-2024-04-17 run --manifest-path ./batcher/aligned-batcher/Cargo.toml --release -- --config ./config-files/config-batcher.yaml --env-file ./batcher/aligned-batcher/.env
 
+batcher_start_local: ./batcher/aligned-batcher/.env user_fund_payment_service
+	@echo "Starting Batcher..."
+	@$(MAKE) run_storage &
+	@cargo +nightly-2024-04-17 run --manifest-path ./batcher/aligned-batcher/Cargo.toml --release -- --config ./config-files/config-batcher.yaml --env-file ./batcher/aligned-batcher/.env.dev
+
 install_batcher:
 	@cargo +nightly-2024-04-17 install --path batcher/aligned-batcher
 
@@ -234,7 +235,6 @@ batcher_send_sp1_burst:
 		--rpc $(RPC_URL) \
 		--batcher_addr $(BATCHER_CONTRACT_ADDRESS)
 
-
 batcher_send_infinite_sp1:
 	@echo "Sending infinite SP1 fibonacci task to Batcher..."
 	@./batcher/aligned/send_infinite_sp1_tasks/send_infinite_sp1_tasks.sh
@@ -319,18 +319,6 @@ batcher_send_groth16_bn254_task: batcher/target/release/aligned
 		--rpc $(RPC_URL) \
 		--batcher_addr $(BATCHER_CONTRACT_ADDRESS)
 
-batcher_send_groth16_burst: batcher/target/release/aligned
-	@echo "Sending Groth16Bn254 1!=0 task to Batcher..."
-	@cd batcher/aligned/ && cargo run --release -- submit \
-		--proving_system Groth16Bn254 \
-		--proof ../../scripts/test_files/gnark_groth16_bn254_infinite_script/infinite_proofs/ineq_1_groth16.proof \
-		--public_input ../../scripts/test_files/gnark_groth16_bn254_infinite_script/infinite_proofs/ineq_1_groth16.pub \
-		--vk ../../scripts/test_files/gnark_groth16_bn254_infinite_script/infinite_proofs/ineq_1_groth16.vk \
-		--repetitions 15 \
-		--proof_generator_addr 0x66f9664f97F2b50F62D13eA064982f936dE76657 \
-		--rpc $(RPC_URL) \
-		--batcher_addr $(BATCHER_CONTRACT_ADDRESS)
-
 batcher_send_infinite_groth16: batcher/target/release/aligned ## Send a different Groth16 BN254 proof using the client every 3 seconds
 	@mkdir -p scripts/test_files/gnark_groth16_bn254_infinite_script/infinite_proofs
 	@echo "Sending a different GROTH16 BN254 proof in a loop every n seconds..."
@@ -440,7 +428,12 @@ generate_groth16_ineq_proof: ## Run the gnark_plonk_bn254_script
 __METRICS__:
 run_metrics: ## Run metrics using metrics-docker-compose.yaml
 	@echo "Running metrics..."
-	@docker-compose -f metrics-docker-compose.yaml up
+	@docker compose -f metrics-docker-compose.yaml up
+
+__STORAGE__:
+run_storage: ## Run storage using storage-docker-compose.yaml
+	@echo "Running storage..."
+	@docker compose -f storage-docker-compose.yaml up
 
 __DEPLOYMENT__:
 deploy_aligned_contracts: ## Deploy Aligned Contracts

Makefile

@@ -11,11 +11,10 @@
   - [How to use the testnet](#how-to-use-the-testnet)
   - [Operator Guide](#operator-guide)
   - [Aligned Infrastructure Guide](#aligned-infrastructure-guide)
-  - [FAQ](https://docs.alignedlayer.com/guides/4_faq)
 
 ## The Project
 
-Aligned is a decentralized network of nodes that verifies Zero-Knowledge and Validity proofs, and post the results in Ethereum. 
+Aligned is a decentralized network of nodes that verifies Zero-Knowledge and Validity proofs, and post the results in Ethereum.
 
 These proofs can be generated and used for a tenth of the price, and with extremely low latency, allowing novel types of applications that weren't possible before in Ethereum.
 
@@ -47,7 +46,9 @@ aligned submit \
 --proof ~/.aligned/test_files/sp1_fibonacci.proof \
 --vm_program ~/.aligned/test_files/sp1_fibonacci.elf \
 --aligned_verification_data_path ~/.aligned/aligned_verification_data \
---conn wss://batcher.alignedlayer.com
+--conn wss://batcher.alignedlayer.com \
+--rpc https://ethereum-holesky-rpc.publicnode.com \
+--batcher_addr 0x815aeCA64a974297942D2Bbf034ABEe22a38A003
 ```
 
 5. You should get a response like this:
@@ -60,7 +61,7 @@ aligned submit \
 [2024-07-01T19:19:18Z INFO  aligned] https://explorer.alignedlayer.com/batches/0xe367d76e832edec893d3a9027b3c231b2e3994c47acfac2e67197c13c9be0c4c
 ```
 
-You can use the link to the explorer to check the status of your transaction. 
+You can use the link to the explorer to check the status of your transaction.
 
 6. After three Ethereum blocks, you can check if it has been verified with:
 
@@ -137,10 +138,10 @@ python3 -m pip install -r examples/verify/requirements.txt
 ```bash
 python3 examples/verify/encode_verification_data.py --aligned-verification-data ~/.aligned/aligned_verification_data/*.json
 ```
- 
+
 If your verification data is in another path, just change the `--aligned-verification-data` parameter.
 
-#### Using a caller contract 
+#### Using a caller contract
 
 To verify a proof in your own contract, use a static call to the Aligned contract. You can use the following [Caller Contract](examples/verify/src/VerifyBatchInclusionCaller.sol) as an example. The code will look like this:
 
@@ -166,4 +167,4 @@ If you want to run an operator, check our [Operator Guide](./docs/operator_guide
 
 ## Aligned Infrastructure Guide
 
-If you are developing in Aligned, or want to run your own devnet, check our [setup Aligned guide](docs/guides/3_setup_aligned.md)
+If you are developing in Aligned, or want to run your own devnet, check our [setup Aligned guide](docs/guides/5_setup_aligned.md)

README.md

@@ -3,10 +3,12 @@ package pkg
 import (
 	"context"
 	"encoding/hex"
-	gethtypes "github.com/ethereum/go-ethereum/core/types"
+	"fmt"
 	"sync"
 	"time"
 
+	gethtypes "github.com/ethereum/go-ethereum/core/types"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/yetanotherco/aligned_layer/metrics"
 
@@ -15,7 +17,7 @@ import (
 	"github.com/Layr-Labs/eigensdk-go/logging"
 	"github.com/Layr-Labs/eigensdk-go/services/avsregistry"
 	blsagg "github.com/Layr-Labs/eigensdk-go/services/bls_aggregation"
-	oppubkeysserv "github.com/Layr-Labs/eigensdk-go/services/operatorpubkeys"
+	oppubkeysserv "github.com/Layr-Labs/eigensdk-go/services/operatorsinfo"
 	eigentypes "github.com/Layr-Labs/eigensdk-go/types"
 	"github.com/ethereum/go-ethereum/event"
 	servicemanager "github.com/yetanotherco/aligned_layer/contracts/bindings/AlignedLayerServiceManager"
@@ -56,6 +58,12 @@ type Aggregator struct {
 	// Stores the taskCreatedBlock for each batch bt batch index
 	batchCreatedBlockByIdx map[uint32]uint64
 
+	// Stores if an operator already submitted a response for a batch
+	// This is to avoid double submissions
+	// struct{} is used as a placeholder because it is the smallest type
+	// go does not have a set type
+	operatorRespondedBatch map[uint32]map[eigentypes.Bytes32]struct{}
+
 	// This task index is to communicate with the local BLS
 	// Service.
 	// Note: In case of a reboot it can start from 0 again
@@ -113,9 +121,24 @@ func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error
 		return nil, err
 	}
 
-	operatorPubkeysService := oppubkeysserv.NewOperatorPubkeysServiceInMemory(context.Background(), clients.AvsRegistryChainSubscriber, clients.AvsRegistryChainReader, logger)
-	avsRegistryService := avsregistry.NewAvsRegistryServiceChainCaller(avsReader.AvsRegistryReader, operatorPubkeysService, logger)
-	blsAggregationService := blsagg.NewBlsAggregatorService(avsRegistryService, logger)
+	// This is a dummy "hash function" made to fulfill the BLS aggregator service API requirements.
+	// When operators respond to a task, a call to `ProcessNewSignature` is made. In `v0.1.6` of the eigensdk,
+	// this function required an argument `TaskResponseDigest`, which has changed to just `TaskResponse` in v0.1.9.
+	// The digest we used in v0.1.6 was just the batch merkle root. To continue with the same idea, the hashing
+	// function is set as the following one, which does nothing more than output the input it receives, which in
+	// our case will be the batch merkle root. If wanted, we could define a real hash function here but there should
+	// not be any need to re-hash the batch merkle root.
+	hashFunction := func(taskResponse eigentypes.TaskResponse) (eigentypes.TaskResponseDigest, error) {
+		taskResponseDigest, ok := taskResponse.([32]byte)
+		if !ok {
+			return eigentypes.TaskResponseDigest{}, fmt.Errorf("TaskResponse is not a 32-byte value")
+		}
+		return taskResponseDigest, nil
+	}
+
+	operatorPubkeysService := oppubkeysserv.NewOperatorsInfoServiceInMemory(context.Background(), clients.AvsRegistryChainSubscriber, clients.AvsRegistryChainReader, nil, logger)
+	avsRegistryService := avsregistry.NewAvsRegistryServiceChainCaller(avsReader.ChainReader, operatorPubkeysService, logger)
+	blsAggregationService := blsagg.NewBlsAggregatorService(avsRegistryService, hashFunction, logger)
 
 	// Metrics
 	reg := prometheus.NewRegistry()
@@ -133,6 +156,7 @@ func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error
 		batchesRootByIdx:       batchesRootByIdx,
 		batchesIdxByRoot:       batchesIdxByRoot,
 		batchCreatedBlockByIdx: batchCreatedBlockByIdx,
+		operatorRespondedBatch: make(map[uint32]map[eigentypes.Bytes32]struct{}),
 		nextBatchIndex:         nextBatchIndex,
 		taskMutex:              &sync.Mutex{},
 		walletMutex:            &sync.Mutex{},
@@ -182,7 +206,16 @@ const MaxSentTxRetries = 5
 
 func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsAggregationServiceResponse) {
 	if blsAggServiceResp.Err != nil {
-		agg.logger.Warn("BlsAggregationServiceResponse contains an error", "err", blsAggServiceResp.Err)
+		agg.taskMutex.Lock()
+		batchMerkleRoot := agg.batchesRootByIdx[blsAggServiceResp.TaskIndex]
+        agg.logger.Error("BlsAggregationServiceResponse contains an error", "err", blsAggServiceResp.Err, "merkleRoot", hex.EncodeToString(batchMerkleRoot[:]))
+		agg.logger.Info("- Locking task mutex: Delete task from operator map", "taskIndex", blsAggServiceResp.TaskIndex)
+
+		// Remove task from the list of tasks
+		delete(agg.operatorRespondedBatch, blsAggServiceResp.TaskIndex)
+
+		agg.logger.Info("- Unlocking task mutex: Delete task from operator map", "taskIndex", blsAggServiceResp.TaskIndex)
+		agg.taskMutex.Unlock()
 		return
 	}
 	nonSignerPubkeys := []servicemanager.BN254G1Point{}
@@ -209,13 +242,16 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 	agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Fetching merkle root")
 	batchMerkleRoot := agg.batchesRootByIdx[blsAggServiceResp.TaskIndex]
 	taskCreatedBlock := agg.batchCreatedBlockByIdx[blsAggServiceResp.TaskIndex]
+
+	// Delete the task from the map
+	delete(agg.operatorRespondedBatch, blsAggServiceResp.TaskIndex)
+
 	agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Fetching merkle root")
 	agg.taskMutex.Unlock()
 
 	agg.logger.Info("Threshold reached", "taskIndex", blsAggServiceResp.TaskIndex,
 		"merkleRoot", hex.EncodeToString(batchMerkleRoot[:]))
 
-
 	currentBlock, err := agg.AggregatorConfig.BaseConfig.EthRpcClient.BlockNumber(context.Background())
 	if err != nil {
 		agg.logger.Error("Error getting current block number", "err", err)
@@ -270,10 +306,8 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 		"merkleRoot", hex.EncodeToString(batchMerkleRoot[:]))
 }
 
-
-
-/// Sends response to contract and waits for transaction receipt
-/// Returns error if it fails to send tx or receipt is not found
+// / 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, nonSignerStakesAndSignature servicemanager.IBLSSignatureCheckerNonSignerStakesAndSignature) (*gethtypes.Receipt, error) {
 	agg.walletMutex.Lock()
 	agg.logger.Infof("- Locked Wallet Resources: Sending aggregated response for batch %s", hex.EncodeToString(batchMerkleRoot[:]))
@@ -299,7 +333,6 @@ func (agg *Aggregator) sendAggregatedResponse(batchMerkleRoot [32]byte, nonSigne
 	return receipt, nil
 }
 
-
 func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, taskCreatedBlock uint32) {
 	agg.AggregatorConfig.BaseConfig.Logger.Info("Adding new task",
 		"Batch merkle root", hex.EncodeToString(batchMerkleRoot[:]))

aggregator/internal/pkg/aggregator.go

@@ -7,6 +7,8 @@ import (
 	"net/rpc"
 	"time"
 
+	eigentypes "github.com/Layr-Labs/eigensdk-go/types"
+
 	"github.com/yetanotherco/aligned_layer/core/types"
 )
 
@@ -73,6 +75,26 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponse(signedTaskResponse *typ
 		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)
@@ -90,6 +112,9 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponse(signedTaskResponse *typ
 
 		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")
 		}