diff --git a/batcher/aligned-batcher/src/lib.rs b/batcher/aligned-batcher/src/lib.rs index 11117b055e..52f62ce115 100644 --- a/batcher/aligned-batcher/src/lib.rs +++ b/batcher/aligned-batcher/src/lib.rs @@ -24,8 +24,8 @@ use aligned_sdk::core::constants::{ BUMP_MAX_RETRIES, BUMP_MAX_RETRY_DELAY, BUMP_MIN_RETRY_DELAY, CONSTANT_GAS_COST, DEFAULT_AGGREGATOR_FEE_PERCENTAGE_MULTIPLIER, DEFAULT_MAX_FEE_PER_PROOF, ETHEREUM_CALL_BACKOFF_FACTOR, ETHEREUM_CALL_MAX_RETRIES, ETHEREUM_CALL_MAX_RETRY_DELAY, - ETHEREUM_CALL_MIN_RETRY_DELAY, GAS_PRICE_PERCENTAGE_MULTIPLIER, MIN_FEE_PER_PROOF, - PERCENTAGE_DIVIDER, RESPOND_TO_TASK_FEE_LIMIT_PERCENTAGE_MULTIPLIER, + ETHEREUM_CALL_MIN_RETRY_DELAY, GAS_PRICE_PERCENTAGE_MULTIPLIER, PERCENTAGE_DIVIDER, + RESPOND_TO_TASK_FEE_LIMIT_PERCENTAGE_MULTIPLIER, }; use aligned_sdk::core::types::{ ClientMessage, GetNonceResponseMessage, NoncedVerificationData, ProofInvalidReason, @@ -519,7 +519,7 @@ impl Batcher { ) .await; self.metrics - .user_error(&["invalid_paument_service_address", ""]); + .user_error(&["invalid_payment_service_address", ""]); return Ok(()); } @@ -630,10 +630,10 @@ impl Batcher { ); send_message( ws_conn_sink.clone(), - SubmitProofResponseMessage::InvalidNonce, + SubmitProofResponseMessage::EthRpcError, ) .await; - self.metrics.user_error(&["invalid_nonce", ""]); + self.metrics.user_error(&["eth_rpc_error", ""]); return Ok(()); } }; @@ -665,19 +665,34 @@ impl Batcher { // finally add the proof to the batch queue. let batch_state_lock = self.batch_state.lock().await; - let Some(proofs_in_batch) = batch_state_lock.get_user_proof_count(&addr).await else { - error!("Failed to get user proof count: User not found in user states, but it should have been already inserted"); + + let msg_max_fee = nonced_verification_data.max_fee; + let Some(user_last_max_fee_limit) = + batch_state_lock.get_user_last_max_fee_limit(&addr).await + else { std::mem::drop(batch_state_lock); send_message( ws_conn_sink.clone(), - SubmitProofResponseMessage::InvalidNonce, + SubmitProofResponseMessage::AddToBatchError, ) .await; - self.metrics.user_error(&["invalid_nonce", ""]); + self.metrics.user_error(&["batcher_state_error", ""]); + 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); + send_message( + ws_conn_sink.clone(), + SubmitProofResponseMessage::AddToBatchError, + ) + .await; + self.metrics.user_error(&["batcher_state_error", ""]); return Ok(()); }; - if !self.check_min_balance(proofs_in_batch + 1, user_balance) { + if !self.verify_user_has_enough_balance(user_balance, user_accumulated_fee, msg_max_fee) { std::mem::drop(batch_state_lock); send_message( ws_conn_sink.clone(), @@ -694,10 +709,10 @@ impl Batcher { std::mem::drop(batch_state_lock); send_message( ws_conn_sink.clone(), - SubmitProofResponseMessage::InvalidNonce, + SubmitProofResponseMessage::AddToBatchError, ) .await; - self.metrics.user_error(&["invalid_nonce", ""]); + self.metrics.user_error(&["batcher_state_error", ""]); return Ok(()); }; @@ -729,21 +744,9 @@ impl Batcher { return Ok(()); } - let msg_max_fee = nonced_verification_data.max_fee; - let Some(user_min_fee) = batch_state_lock.get_user_min_fee(&addr).await else { - std::mem::drop(batch_state_lock); - send_message( - ws_conn_sink.clone(), - SubmitProofResponseMessage::InvalidNonce, - ) - .await; - self.metrics.user_error(&["invalid_nonce", ""]); - return Ok(()); - }; - - if msg_max_fee > user_min_fee { + if msg_max_fee > user_last_max_fee_limit { std::mem::drop(batch_state_lock); - warn!("Invalid max fee for address {addr}, had fee {user_min_fee:?} < {msg_max_fee:?}"); + 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, @@ -782,10 +785,15 @@ impl Batcher { zk_utils::is_verifier_disabled(*disabled_verifiers, verifier) } - // Checks user has sufficient balance for paying all its the proofs in the current batch. - fn check_min_balance(&self, user_proofs_in_batch: usize, user_balance: U256) -> bool { - let min_balance = U256::from(user_proofs_in_batch) * U256::from(MIN_FEE_PER_PROOF); - user_balance >= min_balance + // Verifies user has enough balance for paying all his proofs in the current batch. + fn verify_user_has_enough_balance( + &self, + user_balance: U256, + user_accumulated_fee: U256, + new_msg_max_fee: U256, + ) -> bool { + let required_balance: U256 = user_accumulated_fee + new_msg_max_fee; + user_balance >= required_balance } /// Handles a replacement message @@ -820,7 +828,7 @@ impl Batcher { let original_max_fee = entry.nonced_verification_data.max_fee; if original_max_fee > replacement_max_fee { std::mem::drop(batch_state_lock); - warn!("Invalid replacement message for address {addr}, had fee {original_max_fee:?} < {replacement_max_fee:?}"); + warn!("Invalid replacement message for address {addr}, had max fee: {original_max_fee:?}, received fee: {replacement_max_fee:?}"); send_message( ws_conn_sink.clone(), SubmitProofResponseMessage::InvalidReplacementMessage, @@ -886,13 +894,38 @@ impl Batcher { BatchQueueEntryPriority::new(replacement_max_fee, nonce), ); - let updated_min_fee_in_batch = batch_state_lock.get_user_min_fee_in_batch(&addr); + // update max_fee_limit + let updated_max_fee_limit_in_batch = batch_state_lock.get_user_min_fee_in_batch(&addr); if batch_state_lock - .update_user_min_fee(&addr, updated_min_fee_in_batch) + .update_user_max_fee_limit(&addr, updated_max_fee_limit_in_batch) .is_none() { std::mem::drop(batch_state_lock); warn!("User state for address {addr:?} was not present in batcher user states, but it should be"); + send_message( + ws_conn_sink.clone(), + SubmitProofResponseMessage::AddToBatchError, + ) + .await; + return; + }; + + // update total_fees_in_queue + if batch_state_lock + .update_user_total_fees_in_queue_of_replacement_message( + &addr, + original_max_fee, + replacement_max_fee, + ) + .is_none() + { + std::mem::drop(batch_state_lock); + warn!("User state for address {addr:?} was not present in batcher user states, but it should be"); + send_message( + ws_conn_sink.clone(), + SubmitProofResponseMessage::AddToBatchError, + ) + .await; }; } @@ -984,6 +1017,17 @@ impl Batcher { )); }; + let Some(current_total_fees_in_queue) = batch_state_lock + .get_user_total_fees_in_queue(&proof_submitter_addr) + .await + else { + error!("User state of address {proof_submitter_addr} was not found when trying to update user state. This user state should have been present"); + std::mem::drop(batch_state_lock); + return Err(BatcherError::AddressNotFoundInUserStates( + proof_submitter_addr, + )); + }; + // User state is updated if batch_state_lock .update_user_state( @@ -991,6 +1035,7 @@ impl Batcher { nonce + U256::one(), max_fee, user_proof_count + 1, + current_total_fees_in_queue + max_fee, ) .is_none() { @@ -1029,7 +1074,7 @@ impl Batcher { if current_batch_len < 2 { info!( - "Current batch has {} proof. Waiting for more proofs...", + "Current batch has {} proofs. Waiting for more proofs...", current_batch_len ); return None; @@ -1072,14 +1117,14 @@ impl Batcher { .ok()?; batch_state_lock.batch_queue = resulting_batch_queue; - let updated_user_proof_count_and_min_fee = - batch_state_lock.get_user_proofs_in_batch_and_min_fee(); + let new_user_states = // proofs, max_fee_limit, total_fees_in_queue + batch_state_lock.calculate_new_user_states_data(); let user_addresses: Vec
= batch_state_lock.user_states.keys().cloned().collect(); + let default_value = (0, U256::MAX, U256::zero()); for addr in user_addresses.iter() { - let (proof_count, min_fee) = updated_user_proof_count_and_min_fee - .get(addr) - .unwrap_or(&(0, U256::MAX)); + let (proof_count, max_fee_limit, total_fees_in_queue) = + new_user_states.get(addr).unwrap_or(&default_value); // FIXME: The case where a the update functions return `None` can only happen when the user was not found // in the `user_states` map should not really happen here, but doing this check so that we don't unwrap. @@ -1088,7 +1133,8 @@ impl Batcher { // Now we update the user states related to the batch (proof count in batch and min fee in batch) batch_state_lock.update_user_proof_count(addr, *proof_count)?; - batch_state_lock.update_user_min_fee(addr, *min_fee)?; + batch_state_lock.update_user_max_fee_limit(addr, *max_fee_limit)?; + batch_state_lock.update_user_total_fees_in_queue(addr, *total_fees_in_queue)?; } Some(finalized_batch) diff --git a/batcher/aligned-batcher/src/types/batch_state.rs b/batcher/aligned-batcher/src/types/batch_state.rs index 3e09377ac9..2df6a0cdc6 100644 --- a/batcher/aligned-batcher/src/types/batch_state.rs +++ b/batcher/aligned-batcher/src/types/batch_state.rs @@ -13,6 +13,8 @@ pub(crate) struct BatchState { } impl BatchState { + // CONSTRUCTORS: + pub(crate) fn new() -> Self { Self { batch_queue: BatchQueue::new(), @@ -27,6 +29,8 @@ impl BatchState { } } + // GETTERS: + pub(crate) fn get_entry(&self, sender: Address, nonce: U256) -> Option<&BatchQueueEntry> { self.batch_queue .iter() @@ -43,17 +47,14 @@ impl BatchState { Some(user_state.nonce) } - pub(crate) async fn get_user_min_fee(&self, addr: &Address) -> Option { + pub(crate) async fn get_user_last_max_fee_limit(&self, addr: &Address) -> Option { let user_state = self.get_user_state(addr)?; - Some(user_state.min_fee) + Some(user_state.last_max_fee_limit) } - pub(crate) fn update_user_nonce(&mut self, addr: &Address, new_nonce: U256) -> Option { - if let Entry::Occupied(mut user_state) = self.user_states.entry(*addr) { - user_state.get_mut().nonce = new_nonce; - return Some(new_nonce); - } - None + pub(crate) async fn get_user_total_fees_in_queue(&self, addr: &Address) -> Option { + let user_state = self.get_user_state(addr)?; + Some(user_state.total_fees_in_queue) } pub(crate) async fn get_user_proof_count(&self, addr: &Address) -> Option { @@ -61,29 +62,6 @@ impl BatchState { Some(user_state.proofs_in_batch) } - /// Checks if the entry is valid - /// An entry is valid if there is no entry with the same sender, lower nonce and a lower fee - pub(crate) fn replacement_entry_is_valid( - &mut self, - replacement_entry: &BatchQueueEntry, - ) -> bool { - let replacement_max_fee = replacement_entry.nonced_verification_data.max_fee; - let nonce = replacement_entry.nonced_verification_data.nonce; - let sender = replacement_entry.sender; - - debug!( - "Checking validity of entry with sender: {:?}, nonce: {:?}, max_fee: {:?}", - sender, nonce, replacement_max_fee - ); - - // it is a valid entry only if there is no entry with the same sender, lower nonce and a lower fee - !self.batch_queue.iter().any(|(entry, _)| { - entry.sender == sender - && entry.nonced_verification_data.nonce < nonce - && entry.nonced_verification_data.max_fee < replacement_max_fee - }) - } - pub(crate) fn get_user_min_fee_in_batch(&self, addr: &Address) -> U256 { self.batch_queue .iter() @@ -93,14 +71,17 @@ impl BatchState { .unwrap_or(U256::max_value()) } - pub(crate) fn update_user_min_fee( + // SETTERS: + + pub(crate) fn update_user_max_fee_limit( &mut self, addr: &Address, - new_min_fee: U256, + new_max_fee_limit: U256, ) -> Option { + // TODO refactor to return Result, or something less redundant if let Entry::Occupied(mut user_state) = self.user_states.entry(*addr) { - user_state.get_mut().min_fee = new_min_fee; - return Some(new_min_fee); + user_state.get_mut().last_max_fee_limit = new_max_fee_limit; + return Some(new_max_fee_limit); } None } @@ -110,6 +91,7 @@ impl BatchState { addr: &Address, new_proof_count: usize, ) -> Option { + // TODO refactor to return Result, or something less redundant if let Entry::Occupied(mut user_state) = self.user_states.entry(*addr) { user_state.get_mut().proofs_in_batch = new_proof_count; return Some(new_proof_count); @@ -117,42 +99,119 @@ impl BatchState { None } - /// Updates the user with address `addr` with the provided values of `new_nonce`, `new_min_fee` and - /// `new_proof_count`. + pub(crate) fn update_user_nonce(&mut self, addr: &Address, new_nonce: U256) -> Option { + // TODO refactor to return Result, or something less redundant + if let Entry::Occupied(mut user_state) = self.user_states.entry(*addr) { + user_state.get_mut().nonce = new_nonce; + return Some(new_nonce); + } + None + } + + pub(crate) fn update_user_total_fees_in_queue( + &mut self, + addr: &Address, + new_total_fees_in_queue: U256, + ) -> Option { + // TODO refactor to return Result, or something less redundant + if let Entry::Occupied(mut user_state) = self.user_states.entry(*addr) { + user_state.get_mut().total_fees_in_queue = new_total_fees_in_queue; + return Some(new_total_fees_in_queue); + } + None + } + + pub(crate) fn update_user_total_fees_in_queue_of_replacement_message( + &mut self, + addr: &Address, + original_max_fee: U256, + new_max_fee: U256, + ) -> Option { + // TODO refactor to return Result, or something less redundant + let fee_difference = new_max_fee - original_max_fee; //here we already know new_max_fee > original_max_fee + if let Entry::Occupied(mut user_state) = self.user_states.entry(*addr) { + user_state.get_mut().total_fees_in_queue += fee_difference; + return Some(user_state.get().total_fees_in_queue); + } + None + } + + /// Updates the user with address `addr` with the provided values of + /// `new_nonce`, `new_max_fee_limit`, `new_proof_count` and `new_total_fees_in_queue` /// If state is updated successfully, returns the updated values inside a `Some()` /// If the address was not found in the user states, returns `None` pub(crate) fn update_user_state( &mut self, addr: &Address, new_nonce: U256, - new_min_fee: U256, + new_max_fee_limit: U256, new_proof_count: usize, - ) -> Option<(U256, U256, usize)> { + new_total_fees_in_queue: U256, + ) -> Option<(U256, U256, usize, U256)> { + // TODO refactor to return Result, or something less redundant let updated_nonce = self.update_user_nonce(addr, new_nonce); - let updated_min_fee = self.update_user_min_fee(addr, new_min_fee); + let updated_max_fee_limit = self.update_user_max_fee_limit(addr, new_max_fee_limit); let updated_proof_count = self.update_user_proof_count(addr, new_proof_count); - - if updated_nonce.is_some() && updated_min_fee.is_some() && updated_proof_count.is_some() { - return Some((new_nonce, new_min_fee, new_proof_count)); + let updated_total_fees_in_queue = + self.update_user_total_fees_in_queue(addr, new_total_fees_in_queue); + + if updated_nonce.is_some() + && updated_max_fee_limit.is_some() + && updated_proof_count.is_some() + && updated_total_fees_in_queue.is_some() + { + return Some(( + new_nonce, + new_max_fee_limit, + new_proof_count, + new_total_fees_in_queue, + )); } None } - pub(crate) fn get_user_proofs_in_batch_and_min_fee(&self) -> HashMap { - let mut updated_user_states = HashMap::new(); + // LOGIC: + + pub(crate) fn calculate_new_user_states_data(&self) -> HashMap { + let mut updated_user_states = HashMap::new(); // address -> (proof_count, max_fee_limit, total_fees_in_queue) for (entry, _) in self.batch_queue.iter() { let addr = entry.sender; - let user_min_fee = entry.nonced_verification_data.max_fee; + let max_fee = entry.nonced_verification_data.max_fee; - let (proof_count, min_fee) = - updated_user_states.entry(addr).or_insert((0, user_min_fee)); + let (proof_count, max_fee_limit, total_fees_in_queue) = updated_user_states + .entry(addr) + .or_insert((0, max_fee, U256::zero())); *proof_count += 1; - if entry.nonced_verification_data.max_fee < *min_fee { - *min_fee = entry.nonced_verification_data.max_fee; + *total_fees_in_queue += max_fee; + if max_fee < *max_fee_limit { + *max_fee_limit = max_fee; } } updated_user_states } + + /// Checks if the entry is valid + /// An entry is valid if there is no entry with the same sender, lower nonce and a lower fee + pub(crate) fn replacement_entry_is_valid( + &mut self, + replacement_entry: &BatchQueueEntry, + ) -> bool { + let replacement_max_fee = replacement_entry.nonced_verification_data.max_fee; + let nonce = replacement_entry.nonced_verification_data.nonce; + let sender = replacement_entry.sender; + + debug!( + "Checking validity of entry with sender: {:?}, nonce: {:?}, max_fee: {:?}", + sender, nonce, replacement_max_fee + ); + + // it is a valid entry only if there is no entry with the same sender, lower nonce and a lower fee + !self.batch_queue.iter().any(|(entry, _)| { + entry.sender == sender + && entry.nonced_verification_data.nonce < nonce + && entry.nonced_verification_data.max_fee < replacement_max_fee + }) + } } diff --git a/batcher/aligned-batcher/src/types/user_state.rs b/batcher/aligned-batcher/src/types/user_state.rs index a1627b25f5..203c37bbb0 100644 --- a/batcher/aligned-batcher/src/types/user_state.rs +++ b/batcher/aligned-batcher/src/types/user_state.rs @@ -2,11 +2,8 @@ use ethers::types::U256; pub(crate) struct UserState { pub nonce: U256, - /// The minimum fee of a pending proof for a user. - /// This should always be the fee of the biggest pending nonce by the user. - /// This is used to check if a user is submitting a proof with a higher nonce and higher fee, - /// which is invalid and should be rejected. - pub min_fee: U256, + pub last_max_fee_limit: U256, + pub total_fees_in_queue: U256, pub proofs_in_batch: usize, } @@ -14,7 +11,8 @@ impl UserState { pub(crate) fn new(nonce: U256) -> Self { UserState { nonce, - min_fee: U256::max_value(), + last_max_fee_limit: U256::max_value(), + total_fees_in_queue: U256::zero(), proofs_in_batch: 0, } } diff --git a/batcher/aligned-sdk/src/core/constants.rs b/batcher/aligned-sdk/src/core/constants.rs index c77c5c8bf6..77be77aef7 100644 --- a/batcher/aligned-sdk/src/core/constants.rs +++ b/batcher/aligned-sdk/src/core/constants.rs @@ -8,7 +8,6 @@ pub const CONSTANT_GAS_COST: u128 = + BATCHER_SUBMISSION_BASE_GAS_COST; pub const DEFAULT_MAX_FEE_PER_PROOF: u128 = ADDITIONAL_SUBMISSION_GAS_COST_PER_PROOF * 100_000_000_000; // gas_price = 100 Gwei = 0.0000001 ether (high gas price) -pub const MIN_FEE_PER_PROOF: u128 = ADDITIONAL_SUBMISSION_GAS_COST_PER_PROOF * 100_000_000; // gas_price = 0.1 Gwei = 0.0000000001 ether (low gas price) // % modifiers: (100% is x1, 10% is x0.1, 1000% is x10) pub const RESPOND_TO_TASK_FEE_LIMIT_PERCENTAGE_MULTIPLIER: u128 = 250; // fee_for_aggregator -> respondToTaskFeeLimit modifier