Merge branch 'staging' into feat/batcher-get-last-max-fee-msg

Author: MarcosNicolau

Makefile

@@ -18,7 +18,7 @@ endif
 CONFIG_FILE?=config-files/config.yaml
 AGG_CONFIG_FILE?=config-files/config-aggregator.yaml
 
-OPERATOR_VERSION=v0.19.0
+OPERATOR_VERSION=v0.19.1
 EIGEN_SDK_GO_VERSION_DEVNET=v0.2.0-beta.1
 EIGEN_SDK_GO_VERSION_TESTNET=v0.2.0-beta.1
 EIGEN_SDK_GO_VERSION_MAINNET=v0.2.0-beta.1

crates/Cargo.lock

@@ -1579,13 +1579,26 @@ impl Batcher {
         // Close old sink in old entry and replace it with the new one
         {
             if let Some(messaging_sink) = replacement_entry.messaging_sink {
-                let mut old_sink = messaging_sink.write().await;
-                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");
-                }
+                tokio::spawn(async move {
+                    // Before closing the old sink, send a message to the client notifying that their proof
+                    // has been replaced
+                    send_message(
+                        messaging_sink.clone(),
+                        SubmitProofResponseMessage::ProofReplaced,
+                    )
+                    .await;
+
+                    // Note: This shuts down the sink, but does not wait for it to close, so the other side
+                    // might not receive the message. However, we don't want to wait here since it would
+                    // block the batcher.
+                    let mut old_sink = messaging_sink.write().await;
+                    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!(
                     "Old websocket sink was empty. This should only happen in testing environments"
@@ -2123,6 +2136,24 @@ impl Batcher {
                     warn!("User {:?} has insufficient balance, flushing entire queue as safety measure", address);
 
                     self.flush_queue_and_clear_nonce_cache().await;
+
+                    for entry in finalized_batch {
+                        if let Some(ws_sink) = entry.messaging_sink.as_ref() {
+                            tokio::spawn(send_message(
+                                ws_sink.clone(),
+                                SubmitProofResponseMessage::BatchReset,
+                            ));
+                        } else {
+                            warn!(
+                                "Websocket sink was found empty. This should only happen in tests"
+                            );
+                        }
+                    }
+
+                    return Err(BatcherError::StateCorruptedAndFlushed(format!(
+                        "Queue and user states flushed due to insufficient balance for user {:?}",
+                        address
+                    )));
                 }
                 _ => {
                     // Add more cases here if we want in the future
@@ -2160,7 +2191,10 @@ impl Batcher {
         let mut batch_state_lock = self.batch_state.lock().await;
         for (entry, _) in batch_state_lock.batch_queue.iter() {
             if let Some(ws_sink) = entry.messaging_sink.as_ref() {
-                send_message(ws_sink.clone(), SubmitProofResponseMessage::BatchReset).await;
+                tokio::spawn(send_message(
+                    ws_sink.clone(),
+                    SubmitProofResponseMessage::BatchReset,
+                ));
             } else {
                 warn!("Websocket sink was found empty. This should only happen in tests");
             }
@@ -2249,12 +2283,19 @@ impl Batcher {
 
             // If batch finalization failed, restore the proofs to the queue
             if let Err(e) = batch_finalization_result {
-                error!(
-                    "Batch finalization failed, restoring proofs to queue: {:?}",
-                    e
-                );
-                self.restore_proofs_after_batch_failure(&finalized_batch)
-                    .await;
+                error!("Batch finalization failed: {:?}", e);
+
+                // If the queue was flushed, don't recover
+                match &e {
+                    BatcherError::StateCorruptedAndFlushed(_) => {
+                        info!("State was corrupted and flushed - not restoring proofs");
+                    }
+                    _ => {
+                        info!("Restoring proofs to queue after batch failure");
+                        self.restore_proofs_after_batch_failure(&finalized_batch)
+                            .await;
+                    }
+                }
                 return Err(e);
             }
         }

crates/batcher/src/lib.rs

@@ -66,6 +66,7 @@ pub enum BatcherError {
     WsSinkEmpty,
     AddressNotFoundInUserStates(Address),
     QueueRemoveError(String),
+    StateCorruptedAndFlushed(String),
     EthereumProviderError(String),
 }
 
@@ -148,6 +149,9 @@ impl fmt::Debug for BatcherError {
             BatcherError::QueueRemoveError(e) => {
                 write!(f, "Error while removing entry from queue: {}", e)
             }
+            BatcherError::StateCorruptedAndFlushed(reason) => {
+                write!(f, "Batcher state was corrupted and flushed: {}", reason)
+            }
             BatcherError::EthereumProviderError(e) => {
                 write!(f, "Ethereum provider error: {}", e)
             }

crates/batcher/src/types/errors.rs

@@ -1,6 +1,6 @@
 [package]
 name = "aligned"
-version = "0.19.0"
+version = "0.19.1"
 edition = "2021"
 
 [dependencies]

crates/cli/Cargo.toml

@@ -40,10 +40,10 @@ fi
 
 echo "Downloading SP1 public inputs file..."
 
-if curl -sSf -L "$SP1_PUBLIC_INPUT_NAME" -o "$ALIGNED_TEST_FILES_DIR/$SP1_PUBLIC_INPUT_NAME"; then
+if curl -sSf -L "$SP1_PUBLIC_INPUT_URL" -o "$ALIGNED_TEST_FILES_DIR/$SP1_PUBLIC_INPUT_NAME"; then
     echo "SP1 public inputs downloaded successful"
 else
-    echo "Error: Failed to downloaded $SP1_PUBLIC_INPUT_NAME"
+    echo "Error: Failed to downloaded $SP1_PUBLIC_INPUT_URL"
     exit 1
 fi