feat(batcher): add queue limit

batcher/aligned-batcher/src/lib.rs

@@ -718,6 +718,18 @@ impl Batcher {
             return Ok(());
         };
 
+        if msg_max_fee > user_last_max_fee_limit {
+            std::mem::drop(batch_state_lock);
+            warn!("Invalid max fee for address {addr}, had fee limit of {user_last_max_fee_limit:?}, sent {msg_max_fee:?}");
+            send_message(
+                ws_conn_sink.clone(),
+                SubmitProofResponseMessage::InvalidMaxFee,
+            )
+            .await;
+            self.metrics.user_error(&["invalid_max_fee", ""]);
+            return Ok(());
+        }
+
         let Some(user_accumulated_fee) = batch_state_lock.get_user_total_fees_in_queue(&addr).await
         else {
             std::mem::drop(batch_state_lock);
@@ -782,18 +794,6 @@ impl Batcher {
             return Ok(());
         }
 
-        if msg_max_fee > user_last_max_fee_limit {
-            std::mem::drop(batch_state_lock);
-            warn!("Invalid max fee for address {addr}, had fee limit of {user_last_max_fee_limit:?}, sent {msg_max_fee:?}");
-            send_message(
-                ws_conn_sink.clone(),
-                SubmitProofResponseMessage::InvalidMaxFee,
-            )
-            .await;
-            self.metrics.user_error(&["invalid_max_fee", ""]);
-            return Ok(());
-        }
-
         // * ---------------------------------------------------------------------*
         // *        Add message data into the queue and update user state         *
         // * ---------------------------------------------------------------------*
@@ -1028,6 +1028,19 @@ impl Batcher {
             BatchQueueEntryPriority::new(max_fee, nonce),
         );
 
+        // if max batch qty exceded, remove least priority element
+        if batch_state_lock.batch_queue.len() > self.max_batch_proof_qty {
+            info!("Queue limit exceded, removing least priority element");
+
+            if let Some(lowest_priority_entry) = batch_state_lock.batch_queue.pop() {
+                send_message(
+                    lowest_priority_entry.0.messaging_sink.unwrap(),
+                    SubmitProofResponseMessage::AddToBatchError,
+                )
+                .await;
+            }
+        }
+
         // Update metrics
         let queue_len = batch_state_lock.batch_queue.len();
         let queue_size_bytes = calculate_batch_size(&batch_state_lock.batch_queue)?;