feat: add version check to batcher and client (#503)

Author: NicolasRampoldi

batcher/aligned-batcher/src/lib.rs

@@ -43,6 +43,8 @@ mod zk_utils;
 
 const S3_BUCKET_NAME: &str = "storage.alignedlayer.com";
 
+const PROTOCOL_VERSION: u16 = 0;
+
 pub struct Batcher {
     s3_client: S3Client,
     eth_ws_provider: Provider<Ws>,
@@ -54,6 +56,7 @@ pub struct Batcher {
     max_batch_size: usize,
     last_uploaded_batch_block: Mutex<u64>,
     pre_verification_is_enabled: bool,
+    protocol_version: u16,
 }
 
 impl Batcher {
@@ -100,6 +103,7 @@ impl Batcher {
             max_batch_size: config.batcher.max_batch_size,
             last_uploaded_batch_block: Mutex::new(last_uploaded_batch_block),
             pre_verification_is_enabled: config.batcher.pre_verification_is_enabled,
+            protocol_version: PROTOCOL_VERSION,
         }
     }
 
@@ -146,8 +150,18 @@ impl Batcher {
 
         debug!("WebSocket connection established: {}", addr);
         let (outgoing, incoming) = ws_stream.split();
-
         let outgoing = Arc::new(RwLock::new(outgoing));
+
+        // Send the protocol version to the client
+        let protocol_version_msg = Message::binary(self.protocol_version.to_be_bytes().to_vec());
+
+        outgoing
+            .write()
+            .await
+            .send(protocol_version_msg)
+            .await
+            .expect("Failed to send protocol version");
+
         match incoming
             .try_filter(|msg| future::ready(msg.is_text()))
             .try_for_each(|msg| self.clone().handle_message(msg, outgoing.clone()))

batcher/aligned/src/main.rs

@@ -168,6 +168,8 @@ impl From<ProvingSystemArg> for ProvingSystemId {
     }
 }
 
+const PROTOCOL_VERSION: u16 = 0;
+
 #[tokio::main]
 async fn main() -> Result<(), errors::BatcherClientError> {
     env_logger::Builder::from_env(Env::default().default_filter_or("info")).init();
@@ -182,7 +184,29 @@ async fn main() -> Result<(), errors::BatcherClientError> {
             let (ws_stream, _) = connect_async(url).await?;
 
             info!("WebSocket handshake has been successfully completed");
-            let (mut ws_write, ws_read) = ws_stream.split();
+            let (mut ws_write, mut ws_read) = ws_stream.split();
+
+            // First message from the batcher is the protocol version
+            if let Some(Ok(msg)) = ws_read.next().await {
+                match msg.into_data().try_into() {
+                    Ok(data) => {
+                        let current_protocol_version = u16::from_be_bytes(data);
+                        if current_protocol_version > PROTOCOL_VERSION {
+                            info!(
+                                "You are running an old version of the client, update it running:\ncurl -L https://raw.githubusercontent.com/yetanotherco/aligned_layer/main/batcher/aligned/install_aligned.sh | bash\nClient version: {}, Expected version: {}",
+                                PROTOCOL_VERSION, current_protocol_version
+                            );
+                        }
+                    }
+                    Err(_) => {
+                        error!("Error while reading protocol version");
+                        return Ok(());
+                    }
+                }
+            } else {
+                error!("Batcher did not respond with the protocol version");
+                return Ok(());
+            }
 
             let batch_inclusion_data_directory_path =
                 PathBuf::from(&submit_args.batch_inclusion_data_directory_path);
@@ -336,6 +360,7 @@ async fn receive(
             *num_responses_lock += 1;
 
             let data = msg.into_data();
+
             match serde_json::from_slice::<BatchInclusionData>(&data) {
                 Ok(batch_inclusion_data) => {
                     info!("Received response from batcher");