Merge branch 'staging' into fix/batcher_user_balance_checks

Author: PatStiles

.github/workflows/build-and-test-go.yml

@@ -24,6 +24,8 @@ jobs:
         with:
           go-version: "1.23"
           cache: false
+      - name: foundry-toolchain
+        uses: foundry-rs/[email protected]
       - name: Build SP1 bindings
         run: make build_sp1_linux
       - name: Build Old SP1 bindings

.gitignore

@@ -14,3 +14,5 @@ volume
 config-files/*.last_processed_batch.json
 
 nonce_*.bin
+
+infra/ansible/playbooks/ini/**.ini

Makefile

@@ -1069,3 +1069,21 @@ setup_local_aligned_all:
 
 	tmux new-window -t aligned_layer -n telemetry
 	tmux send-keys -t aligned_layer:telemetry 'docker compose -f telemetry-docker-compose.yaml down && make telemetry_create_env && make telemetry_run_db && make open_telemetry_start && make telemetry_start' C-m
+
+__ANSIBLE__: ## ____
+
+ansible_batcher_create_env: ## Create empty variables files for the Batcher deploy
+	@cp -n infra/ansible/playbooks/ini/caddy-batcher.ini.example infra/ansible/playbooks/ini/caddy-batcher.ini
+	@cp -n infra/ansible/playbooks/ini/config-batcher.ini.example infra/ansible/playbooks/ini/config-batcher.ini
+	@cp -n infra/ansible/playbooks/ini/env-batcher.ini.example infra/ansible/playbooks/ini/env-batcher.ini
+	@echo "Config files for the Batcher created in infra/ansible/playbooks/ini"
+	@echo "Please complete the values and run make ansible_batcher_deploy"
+
+ansible_batcher_deploy: ## Deploy the Batcher. Parameters: INVENTORY, KEYSTORE
+	@if [ -z "$(INVENTORY)" ] || [ -z "$(KEYSTORE)" ]; then \
+		echo "Error: Both INVENTORY and KEYSTORE must be set."; \
+		exit 1; \
+	fi
+	@ansible-playbook infra/ansible/playbooks/batcher.yaml \
+		-i $(INVENTORY) \
+		-e "keystore_path=$(KEYSTORE)"

aggregator/pkg/aggregator.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
+	"math/big"
 	"strings"
 	"sync"
 	"time"
@@ -300,7 +301,21 @@ func (agg *Aggregator) sendAggregatedResponse(batchIdentifierHash [32]byte, batc
 		"senderAddress", hex.EncodeToString(senderAddress[:]),
 		"batchIdentifierHash", hex.EncodeToString(batchIdentifierHash[:]))
 
-	txHash, err := agg.avsWriter.SendAggregatedResponse(batchIdentifierHash, batchMerkleRoot, senderAddress, nonSignerStakesAndSignature)
+	// This function is a callback that is called when the gas price is bumped on the avsWriter.SendAggregatedResponse
+	onGasPriceBumped := func(bumpedGasPrice *big.Int) {
+		agg.metrics.IncBumpedGasPriceForAggregatedResponse()
+		agg.telemetry.BumpedTaskGasPrice(batchMerkleRoot, bumpedGasPrice.String())
+	}
+	receipt, err := agg.avsWriter.SendAggregatedResponse(
+		batchIdentifierHash,
+		batchMerkleRoot,
+		senderAddress,
+		nonSignerStakesAndSignature,
+		agg.AggregatorConfig.Aggregator.GasBaseBumpPercentage,
+		agg.AggregatorConfig.Aggregator.GasBumpIncrementalPercentage,
+		agg.AggregatorConfig.Aggregator.TimeToWaitBeforeBump,
+		onGasPriceBumped,
+	)
 	if err != nil {
 		agg.walletMutex.Unlock()
 		agg.logger.Infof("- Unlocked Wallet Resources: Error sending aggregated response for batch %s. Error: %s", hex.EncodeToString(batchIdentifierHash[:]), err)
@@ -311,13 +326,6 @@ func (agg *Aggregator) sendAggregatedResponse(batchIdentifierHash [32]byte, batc
 	agg.walletMutex.Unlock()
 	agg.logger.Infof("- Unlocked Wallet Resources: Sending aggregated response for batch %s", hex.EncodeToString(batchIdentifierHash[:]))
 
-	receipt, err := utils.WaitForTransactionReceipt(
-		agg.AggregatorConfig.BaseConfig.EthRpcClient, context.Background(), *txHash)
-	if err != nil {
-		agg.telemetry.LogTaskError(batchMerkleRoot, err)
-		return nil, err
-	}
-
 	agg.metrics.IncAggregatedResponses()
 
 	return receipt, nil

aggregator/pkg/telemetry.go

@@ -30,6 +30,11 @@ type TaskErrorMessage struct {
 	TaskError  string `json:"error"`
 }
 
+type TaskGasPriceBumpMessage struct {
+	MerkleRoot     string `json:"merkle_root"`
+	BumpedGasPrice string `json:"bumped_gas_price"`
+}
+
 type TaskSentToEthereumMessage struct {
 	MerkleRoot string `json:"merkle_root"`
 	TxHash     string `json:"tx_hash"`
@@ -96,6 +101,16 @@ func (t *Telemetry) LogTaskError(batchMerkleRoot [32]byte, taskError error) {
 	}
 }
 
+func (t *Telemetry) BumpedTaskGasPrice(batchMerkleRoot [32]byte, bumpedGasPrice string) {
+	body := TaskGasPriceBumpMessage{
+		MerkleRoot:     fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+		BumpedGasPrice: bumpedGasPrice,
+	}
+	if err := t.sendTelemetryMessage("/api/aggregatorTaskGasPriceBump", body); err != nil {
+		t.logger.Error("[Telemetry] Error in LogOperatorResponse", "error", err)
+	}
+}
+
 func (t *Telemetry) TaskSentToEthereum(batchMerkleRoot [32]byte, txHash string) {
 	body := TaskSentToEthereumMessage{
 		MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),

audits/aligned_v0.4/due_diligence_consensys.pdf

@@ -9,7 +9,7 @@ use crate::{
     },
 };
 use aligned_sdk::core::constants::{
-    DEFAULT_BACKOFF_FACTOR, DEFAULT_MAX_RETRIES, DEFAULT_MIN_RETRY_DELAY,
+    DEFAULT_BACKOFF_FACTOR, DEFAULT_MAX_RETRIES, DEFAULT_MAX_RETRY_DELAY, DEFAULT_MIN_RETRY_DELAY,
     GAS_PRICE_INCREMENT_PERCENTAGE_PER_ITERATION, OVERRIDE_GAS_PRICE_PERCENTAGE_MULTIPLIER,
     PERCENTAGE_DIVIDER,
 };
@@ -61,6 +61,9 @@ pub fn calculate_bumped_gas_price(
     bumped_current_gas_price.max(bumped_previous_gas_price)
 }
 
+/// Gets the current nonce from Ethereum.
+/// Retries on recoverable errors using exponential backoff up to `DEFAULT_MAX_RETRIES` times:
+/// (0,5 secs - 1 secs - 2 secs - 4 secs - 8 secs).
 pub async fn get_current_nonce(
     eth_http_provider: &Provider<Http>,
     eth_http_provider_fallback: &Provider<Http>,
@@ -71,6 +74,7 @@ pub async fn get_current_nonce(
         DEFAULT_MIN_RETRY_DELAY,
         DEFAULT_BACKOFF_FACTOR,
         DEFAULT_MAX_RETRIES,
+        DEFAULT_MAX_RETRY_DELAY,
     )
     .await
     .map_err(|e| {
@@ -79,7 +83,9 @@ pub async fn get_current_nonce(
     })
 }
 
-/// Gets the current gas price from Ethereum using exponential backoff.
+/// Gets the current gas price from Ethereum.
+/// Retries on recoverable errors using exponential backoff up to `DEFAULT_MAX_RETRIES` times:
+/// (0,5 secs - 1 secs - 2 secs - 4 secs - 8 secs).
 pub async fn get_gas_price(
     eth_http_provider: &Provider<Http>,
     eth_http_provider_fallback: &Provider<Http>,
@@ -89,6 +95,7 @@ pub async fn get_gas_price(
         DEFAULT_MIN_RETRY_DELAY,
         DEFAULT_BACKOFF_FACTOR,
         DEFAULT_MAX_RETRIES,
+        DEFAULT_MAX_RETRY_DELAY,
     )
     .await
     .map_err(|e| {

audits/aligned_v0.4/fuzzing_labs.pdf

@@ -271,6 +271,9 @@ impl Batcher {
         Ok(())
     }
 
+    /// Listen for Ethereum new blocks.
+    /// Retries on recoverable errors using exponential backoff
+    /// with the maximum number of retries and a `MAX_DELAY` of 1 hour.
     pub async fn listen_new_blocks(self: Arc<Self>) -> Result<(), BatcherError> {
         retry_function(
             || {
@@ -280,6 +283,7 @@ impl Batcher {
             DEFAULT_MIN_RETRY_DELAY,
             DEFAULT_BACKOFF_FACTOR,
             LISTEN_NEW_BLOCKS_MAX_TIMES,
+            DEFAULT_MAX_RETRY_DELAY,
         )
         .await
         .map_err(|e| e.inner())
@@ -916,7 +920,9 @@ impl Batcher {
         }
     }
 
-    /// Gets the user nonce from Ethereum using exponential backoff.
+    /// Gets the user nonce from Ethereum.
+    /// Retries on recoverable errors using exponential backoff up to `DEFAULT_MAX_RETRIES` times:
+    /// (0,5 secs - 1 secs - 2 secs - 4 secs - 8 secs).
     async fn get_user_nonce_from_ethereum(
         &self,
         addr: Address,
@@ -932,6 +938,7 @@ impl Batcher {
             DEFAULT_MIN_RETRY_DELAY,
             DEFAULT_BACKOFF_FACTOR,
             DEFAULT_MAX_RETRIES,
+            DEFAULT_MAX_RETRY_DELAY,
         )
         .await
     }
@@ -1371,6 +1378,10 @@ impl Batcher {
         }
     }
 
+    /// Sends a `create_new_task` transaction to Ethereum and waits for a maximum of 3 blocks for the receipt.
+    /// Retries up to `DEFAULT_MAX_RETRIES` times using exponential backoff on recoverable errors while trying to send the transaction:
+    /// (0,5 secs - 1 secs - 2 secs - 4 secs - 8 secs).
+    /// `ReceiptNotFoundError` is treated as non-recoverable, and the transaction will be canceled using `cancel_create_new_task_tx` in that case.
     async fn create_new_task(
         &self,
         batch_merkle_root: [u8; 32],
@@ -1393,6 +1404,7 @@ impl Batcher {
             DEFAULT_MIN_RETRY_DELAY,
             DEFAULT_BACKOFF_FACTOR,
             DEFAULT_MAX_RETRIES,
+            DEFAULT_MAX_RETRY_DELAY,
         )
         .await;
         match result {
@@ -1416,11 +1428,10 @@ impl Batcher {
     }
 
     /// Sends a transaction to Ethereum with the same nonce as the previous one to override it.
-    /// In case of a recoverable error, it will retry with an exponential backoff up to CANCEL_TRANSACTION_MAX_RETRIES times.
-    /// A tx not included in 3 blocks will be considered an error, and will trigger a bump of the fee, with the rules on ```calculate_bumped_gas_price```
-    /// This will do 5 bumps every 3 blocks, and then the exponential backoff will dominate, doing bumps at 8,13,24,45,89 and so on.
-    /// Errors on ```get_gas_price``` calls inside this function are considered transient,
-    /// so they won't stop the retries.
+    /// Retries on recoverable errors with exponential backoff.
+    /// Bumps the fee if not included in 3 blocks, using `calculate_bumped_gas_price`.
+    /// In the first 5 attemps, bumps the fee every 3 blocks. Then exponential backoff takes over.
+    /// After 2 hours (attempt 13), retries occur hourly for 1 day (33 retries).
     pub async fn cancel_create_new_task_tx(&self, old_tx_gas_price: U256) {
         info!("Cancelling createNewTask transaction...");
         let iteration = Arc::new(Mutex::new(0));
@@ -1458,6 +1469,7 @@ impl Batcher {
             DEFAULT_MIN_RETRY_DELAY,
             DEFAULT_BACKOFF_FACTOR,
             CANCEL_TRANSACTION_MAX_RETRIES,
+            DEFAULT_MAX_RETRY_DELAY,
         )
         .await
         {
@@ -1550,7 +1562,9 @@ impl Batcher {
         Ok(())
     }
 
-    /// Gets the balance of user with address `addr` from Ethereum using exponential backoff.
+    /// Gets the balance of user with address `addr` from Ethereum.
+    /// Retries on recoverable errors using exponential backoff up to `DEFAULT_MAX_RETRIES` times:
+    /// (0,5 secs - 1 secs - 2 secs - 4 secs - 8 secs)
     /// Returns `None` if the balance couldn't be returned
     /// FIXME: This should return a `Result` instead.
     async fn get_user_balance(&self, addr: &Address) -> Option<U256> {
@@ -1565,12 +1579,15 @@ impl Batcher {
             DEFAULT_MIN_RETRY_DELAY,
             DEFAULT_BACKOFF_FACTOR,
             DEFAULT_MAX_RETRIES,
+            DEFAULT_MAX_RETRY_DELAY,
         )
         .await
         .ok()
     }
 
-    /// Checks if the user's balance is unlocked for a given address using exponential backoff.
+    /// Checks if the user's balance is unlocked for a given address.
+    /// Retries on recoverable errors using exponential backoff up to `DEFAULT_MAX_RETRIES` times:
+    /// (0,5 secs - 1 secs - 2 secs - 4 secs - 8 secs).
     /// Returns `false` if an error occurs during the retries.
     async fn user_balance_is_unlocked(&self, addr: &Address) -> bool {
         let Ok(unlocked) = retry_function(
@@ -1584,6 +1601,7 @@ impl Batcher {
             DEFAULT_MIN_RETRY_DELAY,
             DEFAULT_BACKOFF_FACTOR,
             DEFAULT_MAX_RETRIES,
+            DEFAULT_MAX_RETRY_DELAY,
         )
         .await
         else {
@@ -1593,7 +1611,9 @@ impl Batcher {
         unlocked
     }
 
-    /// Uploads the batch to s3 using exponential backoff.
+    /// Uploads the batch to s3.
+    /// Retries on recoverable errors using exponential backoff up to `DEFAULT_MAX_RETRIES` times:
+    /// (0,5 secs - 1 secs - 2 secs - 4 secs - 8 secs).
     async fn upload_batch_to_s3(
         &self,
         batch_bytes: &[u8],
@@ -1611,6 +1631,7 @@ impl Batcher {
             DEFAULT_MIN_RETRY_DELAY,
             DEFAULT_BACKOFF_FACTOR,
             DEFAULT_MAX_RETRIES,
+            DEFAULT_MAX_RETRY_DELAY,
         )
         .await
         .map_err(|e| BatcherError::BatchUploadError(e.to_string()))

batcher/aligned-batcher/src/eth/utils.rs

@@ -170,7 +170,7 @@ pub async fn create_new_task_retryable(
         })?
         .map_err(|e| {
             warn!("Error while waiting for batch inclusion: {e}");
-            RetryError::Transient(BatcherError::TransactionSendError(e.to_string()))
+            RetryError::Permanent(BatcherError::ReceiptNotFoundError)
         })?
         .ok_or(RetryError::Permanent(BatcherError::ReceiptNotFoundError))
 }