feat: build and return array of results instead of result<vec,err>

Author: uri-99

batcher/aligned-sdk/src/communication/messaging.rs

@@ -39,10 +39,10 @@ pub async fn send_messages(
     max_fees: &[U256],
     wallet: Wallet<SigningKey>,
     mut nonce: U256,
-) -> Result<Vec<NoncedVerificationData>, SubmitError> {
+) -> Vec<Result<NoncedVerificationData, SubmitError>> {
     let chain_id = U256::from(wallet.chain_id());
     let mut ws_write = ws_write.lock().await;
-    let mut sent_verification_data: Vec<NoncedVerificationData> = Vec::new();
+    let mut sent_verification_data: Vec<Result<NoncedVerificationData, SubmitError>> = Vec::new();
 
     for (idx, verification_data) in verification_data.iter().enumerate() {
         // Build each message to send
@@ -56,29 +56,37 @@ pub async fn send_messages(
 
         nonce += U256::one();
         let msg = ClientMessage::new(verification_data.clone(), wallet.clone()).await;
-        let msg_bin = cbor_serialize(&msg).map_err(SubmitError::SerializationError)?;
-
+        let msg_bin = match cbor_serialize(&msg).map_err(SubmitError::SerializationError) {
+            Ok(bin) => bin,
+            Err(e) => {
+                error!("Error while serializing message: {:?}", e);
+                sent_verification_data.push(Err(e));
+                return sent_verification_data;
+            }
+        };
+        
         // Send the message
-        ws_write
-            .send(Message::Binary(msg_bin.clone()))
-            .await
-            .map_err(SubmitError::WebSocketConnectionError)?;
+        if let Err(e) = ws_write.send(Message::Binary(msg_bin.clone())).await {
+            error!("Error while sending message: {:?}", e);
+            sent_verification_data.push(Err(SubmitError::WebSocketConnectionError(e)));
+            return sent_verification_data;
+        }
 
         debug!("{:?} Message sent", idx);
 
         // Save the verification data commitment to read its response later
-        sent_verification_data.push(verification_data); 
+        sent_verification_data.push(Ok(verification_data)); 
     }
 
     info!("All proofs sent");
     // This vector is reversed so that while responses are received, removing from the end is cheaper.
-    let sent_verification_data_rev: Vec<NoncedVerificationData> =
+    let sent_verification_data_rev: Vec<Result<NoncedVerificationData, SubmitError>> =
         sent_verification_data
             .into_iter()
             .map(|vd| vd.into())
             .rev()
             .collect();
-    Ok(sent_verification_data_rev)
+    sent_verification_data_rev
 }
 
 // Receives the array of proofs sent
@@ -87,11 +95,11 @@ pub async fn send_messages(
 // finishes when the last proof sent receives its response
 pub async fn receive(
     response_stream: Arc<Mutex<ResponseStream>>,
-    mut sent_verification_data_rev: Vec<NoncedVerificationData>,
-) -> Result<Vec<AlignedVerificationData>, SubmitError> {
+    mut sent_verification_data_rev: Vec<Result<NoncedVerificationData, SubmitError>>,
+) -> Vec<Result<AlignedVerificationData, SubmitError>> {
     // Responses are filtered to only admit binary or close messages.
     let mut response_stream = response_stream.lock().await;
-    let mut aligned_submitted_data: Vec<AlignedVerificationData> = Vec::new();
+    let mut aligned_submitted_data: Vec<Result<AlignedVerificationData, SubmitError>> = Vec::new();
     let last_proof_nonce = get_biggest_nonce(&sent_verification_data_rev);
 
     // read from WS
@@ -100,35 +108,51 @@ pub async fn receive(
         if let Message::Close(close_frame) = msg {
             warn!("Unexpected WS close");
             if let Some(close_msg) = close_frame {
-                return Err(SubmitError::WebSocketClosedUnexpectedlyError(
-                    close_msg.to_owned(),
-                ));
+                aligned_submitted_data.push(Err(SubmitError::WebSocketClosedUnexpectedlyError(close_msg.to_owned())));
+                break;
             }
-            return Err(SubmitError::GenericError(
-                "Connection was closed before receive() processed all sent messages ".to_string(),
-            ));
+            aligned_submitted_data.push(Err(SubmitError::GenericError("Connection was closed before receive() processed all sent messages ".to_string())));
+            break;
         }
 
-        let batch_inclusion_data_message = handle_batcher_response(msg).await?;
+        // first error msg from batcher will drop the rest of the messages in the burst
 
-        let related_verification_data = match_batcher_response_with_stored_verification_data(
-            &batch_inclusion_data_message,
-            &mut sent_verification_data_rev,
-        )?;
+        let batch_inclusion_data_message = match handle_batcher_response(msg).await {
+            Ok(data) => data,
+            Err(e) => {
+                warn!("Error while handling batcher response: {:?}", e);
+                aligned_submitted_data.push(Err(e));
+                break;
+            }
+        };
 
-        let aligned_verification_data =
-            process_batcher_response(&batch_inclusion_data_message, &related_verification_data)?;
+        let related_verification_data = match match_batcher_response_with_stored_verification_data(&batch_inclusion_data_message, &mut sent_verification_data_rev) {
+            Ok(data) => data,
+            Err(e) => {
+                warn!("Error while matching batcher response with sent data: {:?}", e);
+                aligned_submitted_data.push(Err(e));
+                break;
+            }
+        };
 
-        aligned_submitted_data.push(aligned_verification_data);
+        let aligned_verification_data = match process_batcher_response(&batch_inclusion_data_message, &related_verification_data) {
+            Ok(data) => data,
+            Err(e) => {
+                warn!("Error while processing batcher response: {:?}", e);
+                aligned_submitted_data.push(Err(e));
+                break;
+            }
+        };
+
+        aligned_submitted_data.push(Ok(aligned_verification_data));
         debug!("Message response handled successfully");
 
         if batch_inclusion_data_message.user_nonce == last_proof_nonce {
             break;
         }
     }
 
-    debug!("All proof responses handled successfully");
-    Ok(aligned_submitted_data)
+    aligned_submitted_data
 }
 
 async fn handle_batcher_response(msg: Message) -> Result<BatchInclusionData, SubmitError> {
@@ -226,20 +250,23 @@ async fn handle_batcher_response(msg: Message) -> Result<BatchInclusionData, Sub
 // This is used to verify the proof you sent was indeed included in the batch
 fn match_batcher_response_with_stored_verification_data(
     batch_inclusion_data: &BatchInclusionData,
-    sent_verification_data_rev: &mut Vec<NoncedVerificationData>,
+    sent_verification_data_rev: &mut Vec<Result<NoncedVerificationData, SubmitError>>,
 ) -> Result<VerificationDataCommitment, SubmitError> {
     debug!("Matching verification data with batcher response ...");
     let mut index = None;
     for (i, sent_nonced_verification_data) in sent_verification_data_rev.iter_mut().enumerate().rev() { // iterate in reverse since the last element is the most probable to match
-        if sent_nonced_verification_data.nonce == batch_inclusion_data.user_nonce {
-            debug!("local nonced verification data matched with batcher response");
-            index = Some(i);
-            break;
+        if let Ok(sent_nonced_verification_data) = sent_nonced_verification_data {
+            if sent_nonced_verification_data.nonce == batch_inclusion_data.user_nonce {
+                debug!("local nonced verification data matched with batcher response");
+                index = Some(i);
+                break;
+            }
         }
     }
 
+    // cant remove an element while iterating, so we remove it here
     if let Some(i) = index {
-        let verification_data = sent_verification_data_rev.remove(i);
+        let verification_data = sent_verification_data_rev.remove(i).unwrap();
         return Ok(verification_data.verification_data.clone().into());
     }
 
@@ -249,11 +276,13 @@ fn match_batcher_response_with_stored_verification_data(
 // Returns the biggest nonce from the sent verification data
 // Used to know which is the last proof sent to the Batcher,
 // to know when to stop reading the WS for responses
-fn get_biggest_nonce(sent_verification_data: &[NoncedVerificationData]) -> U256 {
+fn get_biggest_nonce(sent_verification_data: &[Result<NoncedVerificationData, SubmitError>]) -> U256 {
     let mut biggest_nonce = U256::zero();
     for verification_data in sent_verification_data.iter() {
-        if verification_data.nonce > biggest_nonce {
-            biggest_nonce = verification_data.nonce;
+        if let Ok(verification_data) = verification_data {
+            if verification_data.nonce > biggest_nonce {
+                biggest_nonce = verification_data.nonce;
+            }
         }
     }
     biggest_nonce

batcher/aligned-sdk/src/sdk.rs

@@ -85,24 +85,33 @@ pub async fn submit_multiple_and_wait_verification(
     max_fees: &[U256],
     wallet: Wallet<SigningKey>,
     nonce: U256,
-) -> Result<Vec<AlignedVerificationData>, errors::SubmitError> {
-    let aligned_verification_data = submit_multiple(
+) -> Vec<Result<AlignedVerificationData, errors::SubmitError>> {
+    let mut aligned_verification_data = submit_multiple(
         batcher_url,
         network,
         verification_data,
         max_fees,
         wallet,
         nonce,
     )
-    .await?;
+    .await;
 
     // TODO: open issue: use a join to .await all at the same time, avoiding the loop
     // And await only once per batch, no need to await multiple proofs if they are in the same batch.
+    let mut error_awaiting_batch_verification: Option<errors::SubmitError> = None;
     for aligned_verification_data_item in aligned_verification_data.iter() {
-        await_batch_verification(aligned_verification_data_item, eth_rpc_url, network).await?;
+        if let Ok(aligned_verification_data_item) = aligned_verification_data_item {
+            if let Err(e) = await_batch_verification(aligned_verification_data_item, eth_rpc_url, network).await {
+                error_awaiting_batch_verification = Some(e);
+                break;
+            }
+        }
+    }
+    if let Some(error_awaiting_batch_verification) = error_awaiting_batch_verification {
+        aligned_verification_data.push(Err(error_awaiting_batch_verification));
     }
 
-    Ok(aligned_verification_data)
+    aligned_verification_data
 }
 
 /// Returns the estimated `max_fee` depending on the batch inclusion preference of the user, based on the max priority gas price.
@@ -238,10 +247,13 @@ pub async fn submit_multiple(
     max_fees: &[U256],
     wallet: Wallet<SigningKey>,
     nonce: U256,
-) -> Result<Vec<AlignedVerificationData>, errors::SubmitError> {
-    let (ws_stream, _) = connect_async(batcher_url)
-        .await
-        .map_err(errors::SubmitError::WebSocketConnectionError)?;
+) -> Vec<Result<AlignedVerificationData, errors::SubmitError>> {
+    let (ws_stream, _) = match connect_async(batcher_url).await {
+        Ok((ws_stream, response)) => (ws_stream, response),
+        Err(e) => {
+            return vec![Err(errors::SubmitError::WebSocketConnectionError(e))]
+        }
+    };
 
     debug!("WebSocket handshake has been successfully completed");
     let (ws_write, ws_read) = ws_stream.split();
@@ -288,20 +300,22 @@ async fn _submit_multiple(
     max_fees: &[U256],
     wallet: Wallet<SigningKey>,
     nonce: U256,
-) -> Result<Vec<AlignedVerificationData>, errors::SubmitError> {
+) -> Vec<Result<AlignedVerificationData, errors::SubmitError>> {
     // First message from the batcher is the protocol version
-    check_protocol_version(&mut ws_read).await?;
+    if let Err(e) = check_protocol_version(&mut ws_read).await {
+        return vec![Err(e)];
+    }
 
     if verification_data.is_empty() {
-        return Err(errors::SubmitError::MissingRequiredParameter(
+        return vec![Err(errors::SubmitError::MissingRequiredParameter(
             "verification_data".to_string(),
-        ));
+        ))];
     }
     if verification_data.len() > 10000 {
         //TODO Magic number
-        return Err(errors::SubmitError::GenericError(
+        return vec![Err(errors::SubmitError::GenericError(
             "Trying to submit too many proofs at once".to_string(),
-        ));
+        ))];
     }
 
     let ws_write_clone = ws_write.clone();
@@ -322,14 +336,16 @@ async fn _submit_multiple(
             wallet,
             nonce,
         )
-        .await?;
+        .await;
         receive(response_stream, sent_verification_data_rev).await
     }
     .await;
 
     // Close connection
     info!("Closing WS connection");
-    ws_write_clone.lock().await.close().await?;
+    if let Err(e) = ws_write_clone.lock().await.close().await {
+        return vec![Err(errors::SubmitError::GenericError(e.to_string()))];
+    }
     result
 }
 
@@ -386,9 +402,13 @@ pub async fn submit_and_wait_verification(
         wallet,
         nonce,
     )
-    .await?;
+    .await;
 
-    Ok(aligned_verification_data[0].clone())
+    match aligned_verification_data.get(0) {
+        Some(Ok(aligned_verification_data)) => Ok(aligned_verification_data.clone()),
+        Some(Err(e)) => Err(errors::SubmitError::GenericError(e.to_string())),
+        None => Err(errors::SubmitError::GenericError("No response from the batcher".to_string())),
+    }
 }
 
 /// Submits a proof to the batcher to be verified in Aligned.
@@ -435,9 +455,13 @@ pub async fn submit(
         wallet,
         nonce,
     )
-    .await?;
+    .await;
 
-    Ok(aligned_verification_data[0].clone())
+    match aligned_verification_data.get(0) {
+        Some(Ok(aligned_verification_data)) => Ok(aligned_verification_data.clone()),
+        Some(Err(e)) => Err(errors::SubmitError::GenericError(e.to_string())),
+        None => Err(errors::SubmitError::GenericError("No response from the batcher".to_string())),
+    }
 }
 
 /// Checks if the proof has been verified with Aligned and is included in the batch.

batcher/aligned/src/main.rs

@@ -321,33 +321,36 @@ async fn main() -> Result<(), AlignedError> {
 
             let max_fees = vec![max_fee; repetitions];
 
-            let aligned_verification_data_vec = match submit_multiple(
+            let aligned_verification_data_vec = submit_multiple(
                 &connect_addr,
                 submit_args.network.into(),
                 &verification_data_arr,
                 &max_fees,
                 wallet.clone(),
                 nonce,
             )
-            .await
-            {
-                Ok(aligned_verification_data_vec) => aligned_verification_data_vec,
-                Err(e) => {
-                    let nonce_file = format!("nonce_{:?}.bin", wallet.address());
-
-                    handle_submit_err(e, nonce_file.as_str()).await;
-                    return Ok(());
-                }
-            };
+            .await;
 
             let mut unique_batch_merkle_roots = HashSet::new();
 
             for aligned_verification_data in aligned_verification_data_vec {
-                save_response(
-                    batch_inclusion_data_directory_path.clone(),
-                    &aligned_verification_data,
-                )?;
-                unique_batch_merkle_roots.insert(aligned_verification_data.batch_merkle_root);
+                match aligned_verification_data {
+                    Ok(aligned_verification_data) => {
+                        info!("Proof submitted to aligned. Batch merkle root: 0x{}", hex::encode(aligned_verification_data.batch_merkle_root));
+                        save_response(
+                            batch_inclusion_data_directory_path.clone(),
+                            &aligned_verification_data,
+                        )?;
+                        unique_batch_merkle_roots.insert(aligned_verification_data.batch_merkle_root);
+                    },
+                    Err(e) => {
+                        warn!("Error while submitting proof: {:?}", e);
+                        let nonce_file = format!("nonce_{:?}.bin", wallet.address());
+
+                        handle_submit_err(e, nonce_file.as_str()).await;
+                        return Ok(());
+                    }
+                };
             }
 
             match unique_batch_merkle_roots.len() {