Merge branch 'refs/heads/staging' into merge_testnet_v_0_10_3

# Conflicts:
#	batcher/Cargo.lock

Author: JuArce

Makefile

@@ -360,7 +360,7 @@ batcher_send_risc0_task_no_pub_input:
         --vm_program ../../scripts/test_files/risc_zero/no_public_inputs/no_pub_input_id.bin \
 		--proof_generator_addr 0x66f9664f97F2b50F62D13eA064982f936dE76657 \
 		--rpc_url $(RPC_URL) \
-		--payment_service_addr $(BATCHER_PAYMENTS_CONTRACT_ADDRESS)
+		--network $(NETWORK)
 
 batcher_send_risc0_burst:
 	@echo "Sending Risc0 fibonacci task to Batcher..."
@@ -1044,3 +1044,25 @@ telemetry_dump_db:
 telemetry_create_env:
 	@cd telemetry_api && \
 		cp .env.dev .env
+
+setup_local_aligned_all:
+	tmux kill-session -t aligned_layer || true
+	tmux new-session -d -s aligned_layer
+
+	tmux new-window -t aligned_layer -n anvil
+	tmux send-keys -t aligned_layer 'make anvil_start_with_block_time' C-m
+
+	tmux new-window -t aligned_layer -n aggregator
+	tmux send-keys -t aligned_layer:aggregator 'make aggregator_start' C-m
+
+	tmux new-window -t aligned_layer -n operator
+	tmux send-keys -t aligned_layer:operator 'sleep 5 && make operator_register_and_start' C-m
+
+	tmux new-window -t aligned_layer -n batcher
+	tmux send-keys -t aligned_layer:batcher 'sleep 60 && make batcher_start_local' C-m
+
+	tmux new-window -t aligned_layer -n explorer
+	tmux send-keys -t aligned_layer:explorer 'make explorer_create_env && make explorer_build_db && make run_explorer' C-m
+
+	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

batcher/aligned-batcher/src/connection.rs

@@ -7,7 +7,7 @@ use aligned_sdk::{
 };
 use futures_util::{stream::SplitSink, SinkExt};
 use lambdaworks_crypto::merkle_tree::merkle::MerkleTree;
-use log::{error, info};
+use log::{debug, error};
 use serde::Serialize;
 use tokio::{net::TcpStream, sync::RwLock};
 use tokio_tungstenite::{
@@ -22,7 +22,11 @@ pub(crate) async fn send_batch_inclusion_data_responses(
     batch_merkle_tree: &MerkleTree<VerificationCommitmentBatch>,
 ) -> Result<(), BatcherError> {
     for (vd_batch_idx, entry) in finalized_batch.iter().enumerate() {
-        let batch_inclusion_data = BatchInclusionData::new(vd_batch_idx, batch_merkle_tree);
+        let batch_inclusion_data = BatchInclusionData::new(
+            vd_batch_idx,
+            batch_merkle_tree,
+            entry.nonced_verification_data.nonce,
+        );
         let response = ResponseMessage::BatchInclusionData(batch_inclusion_data);
 
         let serialized_response = cbor_serialize(&response)
@@ -44,7 +48,7 @@ pub(crate) async fn send_batch_inclusion_data_responses(
             Ok(_) => (),
         }
 
-        info!("Response sent");
+        debug!("Response sent");
     }
 
     Ok(())

batcher/aligned-batcher/src/lib.rs

@@ -28,8 +28,7 @@ use aligned_sdk::core::constants::{
 };
 use aligned_sdk::core::types::{
     ClientMessage, NoncedVerificationData, ProofInvalidReason, ProvingSystemId, ResponseMessage,
-    ValidityResponseMessage, VerificationCommitmentBatch, VerificationData,
-    VerificationDataCommitment,
+    VerificationCommitmentBatch, VerificationData, VerificationDataCommitment,
 };
 
 use aws_sdk_s3::client::Client as S3Client;
@@ -415,11 +414,7 @@ impl Batcher {
         let msg_chain_id = client_msg.verification_data.chain_id;
         if msg_chain_id != self.chain_id {
             warn!("Received message with incorrect chain id: {msg_chain_id}");
-            send_message(
-                ws_conn_sink.clone(),
-                ValidityResponseMessage::InvalidChainId,
-            )
-            .await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidChainId).await;
             self.metrics.user_error(&["invalid_chain_id", ""]);
             return Ok(());
         }
@@ -430,7 +425,7 @@ impl Batcher {
             warn!("Received message with incorrect payment service address: {msg_payment_service_addr}");
             send_message(
                 ws_conn_sink.clone(),
-                ValidityResponseMessage::InvalidPaymentServiceAddress(
+                ResponseMessage::InvalidPaymentServiceAddress(
                     msg_payment_service_addr,
                     self.payment_service.address(),
                 ),
@@ -444,11 +439,7 @@ impl Batcher {
         info!("Verifying message signature...");
         let Ok(addr) = client_msg.verify_signature() else {
             error!("Signature verification error");
-            send_message(
-                ws_conn_sink.clone(),
-                ValidityResponseMessage::InvalidSignature,
-            )
-            .await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidSignature).await;
             self.metrics.user_error(&["invalid_signature", ""]);
             return Ok(());
         };
@@ -457,7 +448,7 @@ impl Batcher {
         let proof_size = client_msg.verification_data.verification_data.proof.len();
         if proof_size > self.max_proof_size {
             error!("Proof size exceeds the maximum allowed size.");
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::ProofTooLarge).await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::ProofTooLarge).await;
             self.metrics.user_error(&["proof_too_large", ""]);
             return Ok(());
         }
@@ -477,7 +468,7 @@ impl Batcher {
                 );
                 send_message(
                     ws_conn_sink.clone(),
-                    ValidityResponseMessage::InvalidProof(ProofInvalidReason::DisabledVerifier(
+                    ResponseMessage::InvalidProof(ProofInvalidReason::DisabledVerifier(
                         verification_data.proving_system,
                     )),
                 )
@@ -493,7 +484,7 @@ impl Batcher {
                 error!("Invalid proof detected. Verification failed");
                 send_message(
                     ws_conn_sink.clone(),
-                    ValidityResponseMessage::InvalidProof(ProofInvalidReason::RejectedProof),
+                    ResponseMessage::InvalidProof(ProofInvalidReason::RejectedProof),
                 )
                 .await;
                 self.metrics.user_error(&[
@@ -518,7 +509,7 @@ impl Batcher {
         if self.user_balance_is_unlocked(&addr).await {
             send_message(
                 ws_conn_sink.clone(),
-                ValidityResponseMessage::InsufficientBalance(addr),
+                ResponseMessage::InsufficientBalance(addr),
             )
             .await;
             self.metrics.user_error(&["insufficient_balance", ""]);
@@ -542,7 +533,7 @@ impl Batcher {
                     error!(
                         "Failed to get user nonce from Ethereum for address {addr:?}. Error: {e:?}"
                     );
-                    send_message(ws_conn_sink.clone(), ValidityResponseMessage::InvalidNonce).await;
+                    send_message(ws_conn_sink.clone(), ResponseMessage::InvalidNonce).await;
                     self.metrics.user_error(&["invalid_nonce", ""]);
                     return Ok(());
                 }
@@ -561,7 +552,7 @@ impl Batcher {
 
         let Some(user_balance) = self.get_user_balance(&addr).await else {
             error!("Could not get balance for address {addr:?}");
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::EthRpcError).await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::EthRpcError).await;
             self.metrics.user_error(&["eth_rpc_error", ""]);
             return Ok(());
         };
@@ -574,7 +565,7 @@ impl Batcher {
         let Some(proofs_in_batch) = batch_state_lock.get_user_proof_count(&addr).await else {
             error!("Failed to get user proof count: User not found in user states, but it should have been already inserted");
             std::mem::drop(batch_state_lock);
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::InvalidNonce).await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidNonce).await;
             self.metrics.user_error(&["invalid_nonce", ""]);
             return Ok(());
         };
@@ -583,7 +574,7 @@ impl Batcher {
             std::mem::drop(batch_state_lock);
             send_message(
                 ws_conn_sink.clone(),
-                ValidityResponseMessage::InsufficientBalance(addr),
+                ResponseMessage::InsufficientBalance(addr),
             )
             .await;
             self.metrics.user_error(&["insufficient_balance", ""]);
@@ -594,15 +585,15 @@ impl Batcher {
         let Some(expected_nonce) = cached_user_nonce else {
             error!("Failed to get cached user nonce: User not found in user states, but it should have been already inserted");
             std::mem::drop(batch_state_lock);
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::InvalidNonce).await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidNonce).await;
             self.metrics.user_error(&["invalid_nonce", ""]);
             return Ok(());
         };
 
         if expected_nonce < msg_nonce {
             std::mem::drop(batch_state_lock);
-            warn!("Invalid nonce for address {addr}, had nonce {expected_nonce:?} < {msg_nonce:?}");
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::InvalidNonce).await;
+            warn!("Invalid nonce for address {addr}, expected nonce: {expected_nonce:?}, received nonce: {msg_nonce:?}");
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidNonce).await;
             self.metrics.user_error(&["invalid_nonce", ""]);
             return Ok(());
         }
@@ -626,15 +617,15 @@ impl Batcher {
         let msg_max_fee = nonced_verification_data.max_fee;
         let Some(user_min_fee) = batch_state_lock.get_user_min_fee(&addr).await else {
             std::mem::drop(batch_state_lock);
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::InvalidNonce).await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidNonce).await;
             self.metrics.user_error(&["invalid_nonce", ""]);
             return Ok(());
         };
 
         if msg_max_fee > user_min_fee {
             std::mem::drop(batch_state_lock);
             warn!("Invalid max fee for address {addr}, had fee {user_min_fee:?} < {msg_max_fee:?}");
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::InvalidMaxFee).await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidMaxFee).await;
             self.metrics.user_error(&["invalid_max_fee", ""]);
             return Ok(());
         }
@@ -654,13 +645,12 @@ impl Batcher {
             .await
         {
             error!("Error while adding entry to batch: {e:?}");
-            send_message(ws_conn_sink, ValidityResponseMessage::AddToBatchError).await;
+            send_message(ws_conn_sink, ResponseMessage::AddToBatchError).await;
             self.metrics.user_error(&["add_to_batch_error", ""]);
             return Ok(());
         };
 
         info!("Verification data message handled");
-        send_message(ws_conn_sink, ValidityResponseMessage::Valid).await;
         Ok(())
     }
 
@@ -695,7 +685,7 @@ impl Batcher {
         let Some(entry) = batch_state_lock.get_entry(addr, nonce) else {
             std::mem::drop(batch_state_lock);
             warn!("Invalid nonce for address {addr}. Queue entry with nonce {nonce} not found");
-            send_message(ws_conn_sink.clone(), ValidityResponseMessage::InvalidNonce).await;
+            send_message(ws_conn_sink.clone(), ResponseMessage::InvalidNonce).await;
             self.metrics.user_error(&["invalid_nonce", ""]);
             return;
         };
@@ -706,7 +696,7 @@ impl Batcher {
             warn!("Invalid replacement message for address {addr}, had fee {original_max_fee:?} < {replacement_max_fee:?}");
             send_message(
                 ws_conn_sink.clone(),
-                ValidityResponseMessage::InvalidReplacementMessage,
+                ResponseMessage::InvalidReplacementMessage,
             )
             .await;
             self.metrics
@@ -730,6 +720,8 @@ impl Batcher {
                 if let Err(e) = old_sink.close().await {
                     // we dont want to exit here, just log the error
                     warn!("Error closing sink: {e:?}");
+                } else {
+                    info!("Old websocket sink closed");
                 }
             } else {
                 warn!(
@@ -744,7 +736,7 @@ impl Batcher {
             warn!("Invalid replacement message");
             send_message(
                 ws_conn_sink.clone(),
-                ValidityResponseMessage::InvalidReplacementMessage,
+                ResponseMessage::InvalidReplacementMessage,
             )
             .await;
             self.metrics
@@ -1055,7 +1047,7 @@ impl Batcher {
                     let merkle_root = hex::encode(batch_merkle_tree.root);
                     send_message(
                         ws_sink.clone(),
-                        ResponseMessage::CreateNewTaskError(merkle_root),
+                        ResponseMessage::CreateNewTaskError(merkle_root, format!("{:?}", e)),
                     )
                     .await
                 } else {
@@ -1281,7 +1273,7 @@ impl Batcher {
                 self.cancel_create_new_task_tx(fee_params.gas_price).await;
                 Err(BatcherError::ReceiptNotFoundError)
             }
-            Err(_) => Err(BatcherError::TransactionSendError),
+            Err(RetryError::Permanent(e)) | Err(RetryError::Transient(e)) => Err(e),
         }
     }
 
@@ -1359,7 +1351,7 @@ impl Batcher {
         info!("Handling nonpaying message");
         let Some(non_paying_config) = self.non_paying_config.as_ref() else {
             warn!("There isn't a non-paying configuration loaded. This message will be ignored");
-            send_message(ws_sink.clone(), ValidityResponseMessage::InvalidNonce).await;
+            send_message(ws_sink.clone(), ResponseMessage::InvalidNonce).await;
             return Ok(());
         };
 
@@ -1368,7 +1360,7 @@ impl Batcher {
             error!("Could not get balance for non-paying address {replacement_addr:?}");
             send_message(
                 ws_sink.clone(),
-                ValidityResponseMessage::InsufficientBalance(replacement_addr),
+                ResponseMessage::InsufficientBalance(replacement_addr),
             )
             .await;
             return Ok(());
@@ -1378,26 +1370,17 @@ impl Batcher {
             error!("Insufficient funds for non-paying address {replacement_addr:?}");
             send_message(
                 ws_sink.clone(),
-                ValidityResponseMessage::InsufficientBalance(replacement_addr),
+                ResponseMessage::InsufficientBalance(replacement_addr),
             )
             .await;
             return Ok(());
         }
 
         let batch_state_lock = self.batch_state.lock().await;
-        let Some(non_paying_nonce) = batch_state_lock.get_user_nonce(&replacement_addr).await
-        else {
-            std::mem::drop(batch_state_lock);
-            error!("Nonce for non-paying address {replacement_addr:?} not found in cache.");
-            send_message(ws_sink.clone(), ValidityResponseMessage::EthRpcError).await;
-            return Ok(());
-        };
-
-        debug!("Non-paying nonce: {:?}", non_paying_nonce);
 
         let nonced_verification_data = NoncedVerificationData::new(
             client_msg.verification_data.verification_data.clone(),
-            non_paying_nonce,
+            client_msg.verification_data.nonce,
             DEFAULT_MAX_FEE_PER_PROOF.into(), // 13_000 gas per proof * 100 gwei gas price (upper bound)
             self.chain_id,
             self.payment_service.address(),
@@ -1421,12 +1404,11 @@ impl Batcher {
             .await
         {
             info!("Error while adding nonpaying address entry to batch: {e:?}");
-            send_message(ws_sink, ValidityResponseMessage::AddToBatchError).await;
+            send_message(ws_sink, ResponseMessage::AddToBatchError).await;
             return Ok(());
         };
 
         info!("Non-paying verification data message handled");
-        send_message(ws_sink, ValidityResponseMessage::Valid).await;
         Ok(())
     }
 

batcher/aligned-batcher/src/retry/batcher_retryables.rs

@@ -129,7 +129,9 @@ pub async fn create_new_task_retryable(
         Err(ContractError::Revert(err)) => {
             // Since transaction was reverted, we don't want to retry with fallback.
             warn!("Transaction reverted {:?}", err);
-            return Err(RetryError::Permanent(BatcherError::TransactionSendError));
+            return Err(RetryError::Permanent(BatcherError::TransactionSendError(
+                err.to_string(),
+            )));
         }
         _ => {
             call_fallback = payment_service_fallback
@@ -146,9 +148,15 @@ pub async fn create_new_task_retryable(
                 Ok(pending_tx) => pending_tx,
                 Err(ContractError::Revert(err)) => {
                     warn!("Transaction reverted {:?}", err);
-                    return Err(RetryError::Permanent(BatcherError::TransactionSendError));
+                    return Err(RetryError::Permanent(BatcherError::TransactionSendError(
+                        err.to_string(),
+                    )));
+                }
+                Err(err) => {
+                    return Err(RetryError::Transient(BatcherError::TransactionSendError(
+                        err.to_string(),
+                    )))
                 }
-                _ => return Err(RetryError::Transient(BatcherError::TransactionSendError)),
             }
         }
     };
@@ -162,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)
+            RetryError::Transient(BatcherError::TransactionSendError(e.to_string()))
         })?
         .ok_or(RetryError::Permanent(BatcherError::ReceiptNotFoundError))
 }