Use DoublePriorityQueue

Author: Mechanix97

Makefile

@@ -393,6 +393,8 @@ batcher_send_sp1_task:
 		--vm_program ../../scripts/test_files/sp1/sp1_fibonacci_4_1_3.elf \
 		--proof_generator_addr 0x66f9664f97F2b50F62D13eA064982f936dE76657 \
 		--rpc_url $(RPC_URL) \
+		--max_fee $(MAX_FEE) \
+		--private_key 0x224b7eb7449992aac96d631d9677f7bf5888245eef6d6eeda31e62d2f29a83e4 \
 		--network $(NETWORK)
 
 batcher_send_sp1_burst:
@@ -418,6 +420,7 @@ batcher_send_risc0_task:
         --vm_program ../../scripts/test_files/risc_zero/fibonacci_proof_generator/fibonacci_id_2_0.bin \
         --public_input ../../scripts/test_files/risc_zero/fibonacci_proof_generator/risc_zero_fibonacci_2_0.pub \
 		--proof_generator_addr 0x66f9664f97F2b50F62D13eA064982f936dE76657 \
+		--private_key 0xc7f9f3b5c5a35632e8a8720e84b39d3987bfb01dbfd305bf359d9bb891ec5596 \
 		--rpc_url $(RPC_URL) \
 		--network $(NETWORK)
 

batcher/aligned-batcher/src/lib.rs

@@ -16,7 +16,7 @@ use tokio::time::{timeout, Instant};
 use types::batch_state::BatchState;
 use types::user_state::UserState;
 
-use batch_queue::{calculate_batch_size, get_lowest_priority_entry};
+use batch_queue::calculate_batch_size;
 use std::collections::HashMap;
 use std::env;
 use std::net::SocketAddr;
@@ -1050,6 +1050,8 @@ impl Batcher {
         let max_fee = verification_data.max_fee;
         let nonce = verification_data.nonce;
 
+        info!("ME llega max_fee {}", max_fee);
+
         let batch_queue_len = batch_state_lock.batch_queue.len();
 
         let new_entry = BatchQueueEntry::new(
@@ -1061,44 +1063,19 @@ impl Batcher {
         );
         let new_entry_priority = BatchQueueEntryPriority::new(max_fee, nonce);
 
-        if batch_queue_len + 1 > self.max_batch_proof_qty {
-            info!("Batch queue is full. Trying to remove an entry...");
-            let (lowest_priority_entry, lowest_priority_entry_priority) =
-                get_lowest_priority_entry(&batch_state_lock.batch_queue).unwrap();
-
-            if lowest_priority_entry_priority < new_entry_priority {
-                if batch_state_lock
-                    .batch_queue
-                    .remove(&lowest_priority_entry)
-                    .is_none()
-                {
-                    // If this happens, we have a bug in our code
-                    error!("Some proofs were not found in the queue. This should not happen.");
-                    std::mem::drop(batch_state_lock);
-                    return Err(BatcherError::QueueRemoveError(
-                        "Some proofs were not found in the queue".into(),
-                    ));
-                }
-                info!("Removed entry from batch queue.");
+        batch_state_lock
+            .batch_queue
+            .push(new_entry, new_entry_priority);
 
+        // if max batch qty exceded, remove least priority proof
+        if batch_queue_len > self.max_batch_proof_qty {
+            if let Some(lowest_priority_entry) = batch_state_lock.batch_queue.pop_min() {
                 send_message(
-                    lowest_priority_entry.messaging_sink.unwrap(),
+                    lowest_priority_entry.0.messaging_sink.unwrap(),
                     SubmitProofResponseMessage::AddToBatchError,
                 )
                 .await;
-
-                batch_state_lock
-                    .batch_queue
-                    .push(new_entry, new_entry_priority);
-            } else {
-                error!("Queue is full and fee isn't enough");
-                std::mem::drop(batch_state_lock);
-                return Err(BatcherError::BatchQueueIsFull);
             }
-        } else {
-            batch_state_lock
-                .batch_queue
-                .push(new_entry, new_entry_priority);
         }
 
         // Update metrics

batcher/aligned-batcher/src/types/batch_queue.rs

@@ -6,7 +6,7 @@ use aligned_sdk::{
     },
 };
 use ethers::types::{Address, Signature, U256};
-use priority_queue::PriorityQueue;
+use priority_queue::DoublePriorityQueue;
 use std::{
     hash::{Hash, Hasher},
     ops::ControlFlow,
@@ -26,8 +26,8 @@ pub(crate) struct BatchQueueEntry {
 
 #[derive(Clone)]
 pub(crate) struct BatchQueueEntryPriority {
-    max_fee: U256,
-    nonce: U256,
+    pub max_fee: U256,
+    pub nonce: U256,
 }
 
 impl BatchQueueEntry {
@@ -119,7 +119,7 @@ impl Ord for BatchQueueEntryPriority {
     }
 }
 
-pub(crate) type BatchQueue = PriorityQueue<BatchQueueEntry, BatchQueueEntryPriority>;
+pub(crate) type BatchQueue = DoublePriorityQueue<BatchQueueEntry, BatchQueueEntryPriority>;
 
 /// Calculates the size of the batch represented by the given batch queue.
 pub(crate) fn calculate_batch_size(batch_queue: &BatchQueue) -> Result<usize, BatcherError> {
@@ -165,7 +165,7 @@ pub(crate) fn try_build_batch(
     let mut finalized_batch = batch_queue;
     let mut batch_size = calculate_batch_size(&finalized_batch)?;
 
-    while let Some((entry, _)) = finalized_batch.peek() {
+    while let Some((entry, _)) = finalized_batch.peek_max() {
         let batch_len = finalized_batch.len();
         let fee_per_proof = calculate_fee_per_proof(batch_len, gas_price, constant_gas_cost);
 
@@ -186,7 +186,7 @@ pub(crate) fn try_build_batch(
                     .len();
             batch_size -= verification_data_size;
 
-            finalized_batch.pop();
+            finalized_batch.pop_max();
 
             continue;
         }
@@ -201,7 +201,7 @@ pub(crate) fn try_build_batch(
         return Err(BatcherError::BatchCostTooHigh);
     }
 
-    Ok(finalized_batch.clone().into_sorted_vec())
+    Ok(finalized_batch.clone().into_descending_sorted_vec())
 }
 
 fn calculate_fee_per_proof(batch_len: usize, gas_price: U256, constant_gas_cost: u128) -> U256 {
@@ -212,27 +212,6 @@ fn calculate_fee_per_proof(batch_len: usize, gas_price: U256, constant_gas_cost:
     U256::from(gas_per_proof) * gas_price
 }
 
-pub(crate) fn get_lowest_priority_entry(
-    batch_queue: &BatchQueue,
-) -> Option<(BatchQueueEntry, BatchQueueEntryPriority)> {
-    let mut lowest_fee_entry: Option<(BatchQueueEntry, BatchQueueEntryPriority)> = None;
-
-    for (entry, priority) in batch_queue {
-        match &lowest_fee_entry {
-            Some((e, p)) => {
-                if *priority < *p {
-                    lowest_fee_entry = Some((e.clone(), p.clone()));
-                }
-            }
-            None => {
-                lowest_fee_entry = Some((entry.clone(), priority.clone()));
-            }
-        }
-    }
-
-    lowest_fee_entry
-}
-
 #[cfg(test)]
 mod test {
     use aligned_sdk::core::constants::DEFAULT_CONSTANT_GAS_COST;