diff --git a/rs/ethereum/cketh/minter/src/lifecycle/init.rs b/rs/ethereum/cketh/minter/src/lifecycle/init.rs index 56f4b7679d73..14b61c6e946c 100644 --- a/rs/ethereum/cketh/minter/src/lifecycle/init.rs +++ b/rs/ethereum/cketh/minter/src/lifecycle/init.rs @@ -3,7 +3,7 @@ use crate::lifecycle::EthereumNetwork; use crate::numeric::{BlockNumber, TransactionNonce, Wei}; use crate::state::automatic_deposits::AutomaticDeposits; use crate::state::eth_logs_scraping::{LogScrapingId, LogScrapings}; -use crate::state::transactions::{SweepId, TransactionPipeline, WithdrawalTransactions}; +use crate::state::transactions::{SweepId, WithdrawalTransactions}; use crate::state::{InvalidStateError, State}; use crate::{EVM_RPC_ID_PRODUCTION, EVM_RPC_ID_STAGING}; use candid::types::number::Nat; @@ -106,7 +106,6 @@ impl TryFrom for State { pending_withdrawal_principals: Default::default(), pending_deposit_principals: Default::default(), withdrawal_transactions: WithdrawalTransactions::new(initial_nonce), - sweeper_transactions: TransactionPipeline::new(initial_sweeper_nonce), next_sweep_id: SweepId(0), cketh_ledger_id: ledger_id, cketh_minimum_withdrawal_amount: minimum_withdrawal_amount, @@ -128,7 +127,7 @@ impl TryFrom for State { ckerc20_tokens: Default::default(), erc20_balances: Default::default(), log_scrapings, - automatic_deposits: AutomaticDeposits::default(), + automatic_deposits: AutomaticDeposits::new(initial_sweeper_nonce), sweeper_contract_address, sweeper_funding: Default::default(), }; diff --git a/rs/ethereum/cketh/minter/src/lifecycle/tests.rs b/rs/ethereum/cketh/minter/src/lifecycle/tests.rs index ce40cb83cecf..ac5d51ac9ecf 100644 --- a/rs/ethereum/cketh/minter/src/lifecycle/tests.rs +++ b/rs/ethereum/cketh/minter/src/lifecycle/tests.rs @@ -159,7 +159,7 @@ mod init { .expect("valid init args"); assert_eq!( - state.sweeper_transactions.next_transaction_nonce(), + state.automatic_deposits.next_sweeper_transaction_nonce(), TransactionNonce::ZERO ); } @@ -178,7 +178,7 @@ mod init { TransactionNonce::from(7_u8) ); assert_eq!( - state.sweeper_transactions.next_transaction_nonce(), + state.automatic_deposits.next_sweeper_transaction_nonce(), TransactionNonce::from(42_u8) ); } diff --git a/rs/ethereum/cketh/minter/src/state.rs b/rs/ethereum/cketh/minter/src/state.rs index 4efee27124d1..6fd7eb06bdaa 100644 --- a/rs/ethereum/cketh/minter/src/state.rs +++ b/rs/ethereum/cketh/minter/src/state.rs @@ -29,7 +29,7 @@ use std::cell::RefCell; use std::collections::{BTreeMap, BTreeSet, HashSet, btree_map}; use std::fmt::{Display, Formatter}; use strum_macros::EnumIter; -use transactions::{SweepId, SweeperTransactionPipeline, WithdrawalTransactions}; +use transactions::{SweepId, WithdrawalTransactions}; pub mod audit; pub mod automatic_deposits; @@ -78,9 +78,6 @@ pub struct State { pub minted_events: BTreeMap, pub invalid_events: BTreeMap, pub withdrawal_transactions: WithdrawalTransactions, - /// The dedicated sweeper address' transaction pipeline: sweeps sent from the sweeper address on - /// its own nonce sequence, independent of the main-address withdrawal pipeline. - pub sweeper_transactions: SweeperTransactionPipeline, /// Monotonic counter minting the next [`SweepId`] for the sweeper pipeline. pub next_sweep_id: SweepId, pub skipped_blocks: BTreeMap>, @@ -598,8 +595,8 @@ impl State { if let Some(nonce) = next_sweeper_transaction_nonce { let nonce = TransactionNonce::try_from(nonce) .map_err(|e| InvalidStateError::InvalidTransactionNonce(format!("ERROR: {e}")))?; - self.sweeper_transactions - .update_next_transaction_nonce(nonce); + self.automatic_deposits + .update_next_sweeper_transaction_nonce(nonce); } if let Some(amount) = minimum_withdrawal_amount { let minimum_withdrawal_amount = Wei::try_from(amount).map_err(|e| { @@ -711,7 +708,8 @@ impl State { other.ledger_suite_orchestrator_id ); ensure_eq!(self.ckerc20_tokens, other.ckerc20_tokens); - ensure_eq!(self.automatic_deposits, other.automatic_deposits); + self.automatic_deposits + .is_equivalent_to(&other.automatic_deposits)?; ensure_eq!( self.sweeper_contract_address, other.sweeper_contract_address @@ -719,8 +717,6 @@ impl State { ensure_eq!(self.sweeper_funding, other.sweeper_funding); ensure_eq!(self.next_sweep_id, other.next_sweep_id); - self.sweeper_transactions - .is_equivalent_to(&other.sweeper_transactions)?; self.withdrawal_transactions .is_equivalent_to(&other.withdrawal_transactions) } diff --git a/rs/ethereum/cketh/minter/src/state/audit.rs b/rs/ethereum/cketh/minter/src/state/audit.rs index 67c2b90479b8..b4f11e2dae21 100644 --- a/rs/ethereum/cketh/minter/src/state/audit.rs +++ b/rs/ethereum/cketh/minter/src/state/audit.rs @@ -116,31 +116,33 @@ pub fn apply_state_transition(state: &mut State, payload: &EventType) { } EventType::AcceptedSweepRequest(request) => { state.next_sweep_id = request.id.next(); - state.sweeper_transactions.record_request(request.clone()); + state + .automatic_deposits + .record_sweep_request(request.clone()); } EventType::CreatedSweeperTransaction { sweep_id, transaction, } => { state - .sweeper_transactions - .record_created_transaction(*sweep_id, transaction.clone()); + .automatic_deposits + .record_created_sweep_transaction(*sweep_id, transaction.clone()); } EventType::SignedSweeperTransaction { sweep_id: _, transaction, } => { state - .sweeper_transactions - .record_signed_transaction(transaction.clone()); + .automatic_deposits + .record_signed_sweep_transaction(transaction.clone()); } EventType::ReplacedSweeperTransaction { sweep_id: _, transaction, } => { state - .sweeper_transactions - .record_resubmit_transaction(transaction.clone()); + .automatic_deposits + .record_resubmit_sweep_transaction(transaction.clone()); } EventType::FinalizedSweeperTransaction { sweep_id, @@ -149,8 +151,8 @@ pub fn apply_state_transition(state: &mut State, payload: &EventType) { // The sweeper pipeline is never reimbursed and holds no ckETH balance, so unlike the main // pipeline there is no reimbursement tail or balance update — just the finalize mechanics. let _ = state - .sweeper_transactions - .record_finalized_transaction(*sweep_id, transaction_receipt); + .automatic_deposits + .record_finalized_sweep_transaction(*sweep_id, transaction_receipt); } EventType::ReimbursedEthWithdrawal(Reimbursed { burn_in_block: withdrawal_id, diff --git a/rs/ethereum/cketh/minter/src/state/automatic_deposits/mod.rs b/rs/ethereum/cketh/minter/src/state/automatic_deposits/mod.rs index 9f374281be3e..25687ca7caf4 100644 --- a/rs/ethereum/cketh/minter/src/state/automatic_deposits/mod.rs +++ b/rs/ethereum/cketh/minter/src/state/automatic_deposits/mod.rs @@ -4,10 +4,15 @@ mod tests; use crate::attestation::AttestationRequest; use crate::deposit_address::DepositAddress; use crate::endpoints::{DepositErc20Error, DepositErc20Response, DepositStatus, DetectedDeposit}; -use crate::numeric::{BlockNumber, Erc20Value}; +use crate::eth_rpc::Hash; +use crate::eth_rpc_client::responses::TransactionReceipt; +use crate::numeric::{BlockNumber, Erc20Value, TransactionCount, TransactionNonce}; use crate::state::event::{AutomaticDeposit, DepositAddressRegistration, DepositAddressRegistry}; +use crate::state::transactions::{ + ResubmitTransactionError, SweepId, SweepRequest, SweeperTransactionPipeline, +}; use crate::timed_sized_map::{Entry, InsertError, TimedSizedMap, Timestamp}; -use crate::tx::TransactionSignature; +use crate::tx::{Finalized, GasFeeEstimate, Signed, SweepTransaction, TransactionSignature}; use ic_ethereum_types::Address; use icrc_ledger_types::icrc1::account::Account; use std::collections::BTreeMap; @@ -66,9 +71,128 @@ pub struct AutomaticDeposits { /// entries naming a retired helper stay behind forever. [`Self::attestations_len`] is exported /// as a metric so that growth is visible before it needs bounding. attestations: BTreeMap, + /// The dedicated sweeper address' transaction pipeline: sweeps sent from the sweeper address on + /// its own nonce sequence, independent of the main-address withdrawal pipeline. + sweeper_transactions: SweeperTransactionPipeline, } impl AutomaticDeposits { + pub fn new(initial_sweeper_nonce: TransactionNonce) -> Self { + Self { + sweeper_transactions: SweeperTransactionPipeline::new(initial_sweeper_nonce), + ..Default::default() + } + } + + pub fn has_pending_sweeps(&self) -> bool { + self.sweeper_transactions.has_pending_requests() + } + + pub fn is_sent_sweep_tx_empty(&self) -> bool { + self.sweeper_transactions.is_sent_tx_empty() + } + + pub fn next_sweeper_transaction_nonce(&self) -> TransactionNonce { + self.sweeper_transactions.next_transaction_nonce() + } + + pub fn update_next_sweeper_transaction_nonce(&mut self, new_nonce: TransactionNonce) { + self.sweeper_transactions + .update_next_transaction_nonce(new_nonce) + } + + pub fn sweep_requests_batch(&self, requested_batch_size: usize) -> Vec { + self.sweeper_transactions + .requests_batch(requested_batch_size) + } + + pub fn create_resubmit_sweep_transactions( + &self, + latest_transaction_count: TransactionCount, + current_gas_fee: GasFeeEstimate, + ) -> Vec>> { + self.sweeper_transactions + .create_resubmit_transactions(latest_transaction_count, current_gas_fee) + } + + pub fn sweep_transactions_to_sign_batch( + &self, + batch_size: usize, + ) -> Vec<(SweepId, SweepTransaction)> { + self.sweeper_transactions + .transactions_to_sign_batch(batch_size) + } + + pub fn sweep_transactions_to_send_batch( + &self, + latest_transaction_count: TransactionCount, + batch_size: usize, + ) -> Vec> { + self.sweeper_transactions + .transactions_to_send_batch(latest_transaction_count, batch_size) + } + + pub fn sent_sweep_transactions_to_finalize( + &self, + finalized_transaction_count: &TransactionCount, + ) -> BTreeMap { + self.sweeper_transactions + .sent_transactions_to_finalize(finalized_transaction_count) + } + + pub fn record_sweep_request(&mut self, request: SweepRequest) { + self.sweeper_transactions.record_request(request) + } + + pub fn reschedule_sweep_request(&mut self, id: SweepId) { + self.sweeper_transactions.reschedule_request(id) + } + + pub fn record_created_sweep_transaction(&mut self, id: SweepId, transaction: SweepTransaction) { + self.sweeper_transactions + .record_created_transaction(id, transaction) + } + + pub fn record_signed_sweep_transaction( + &mut self, + signed_transaction: Signed, + ) { + self.sweeper_transactions + .record_signed_transaction(signed_transaction) + } + + pub fn record_resubmit_sweep_transaction(&mut self, new_tx: SweepTransaction) { + self.sweeper_transactions + .record_resubmit_transaction(new_tx) + } + + pub fn record_finalized_sweep_transaction( + &mut self, + id: SweepId, + receipt: &TransactionReceipt, + ) -> Finalized { + self.sweeper_transactions + .record_finalized_transaction(id, receipt) + } + + /// Equality as replay defines it: the sweeper pipeline reorders its queue without recording an + /// event, so it compares itself rather than being compared field by field. + pub fn is_equivalent_to(&self, other: &Self) -> Result<(), String> { + use ic_utils_ensure::ensure_eq; + + let Self { + watchlist, + sweep, + attestations, + sweeper_transactions, + } = self; + + ensure_eq!(watchlist, &other.watchlist); + ensure_eq!(sweep, &other.sweep); + ensure_eq!(attestations, &other.attestations); + sweeper_transactions.is_equivalent_to(&other.sweeper_transactions) + } + /// The signature already stored for `request`, if any: signing another would cost a /// threshold-ECDSA signature for the same digest. pub fn attestation(&self, request: &AttestationRequest) -> Option<&TransactionSignature> { @@ -357,6 +481,7 @@ impl Default for AutomaticDeposits { watchlist: TimedSizedMap::new(DEPOSIT_ADDRESS_SCAN_WINDOW, MAX_ACTIVE_DEPOSITS), sweep: BTreeMap::new(), attestations: BTreeMap::new(), + sweeper_transactions: SweeperTransactionPipeline::new(TransactionNonce::ZERO), } } } diff --git a/rs/ethereum/cketh/minter/src/state/tests.rs b/rs/ethereum/cketh/minter/src/state/tests.rs index 1f433c44003a..888310ed45a8 100644 --- a/rs/ethereum/cketh/minter/src/state/tests.rs +++ b/rs/ethereum/cketh/minter/src/state/tests.rs @@ -15,7 +15,6 @@ use crate::state::eth_logs_scraping::{LogScrapingId, LogScrapings}; use crate::state::event::{Event, EventType}; use crate::state::transactions::{ Erc20WithdrawalRequest, EthWithdrawalRequest, ReimbursementIndex, SweepId, - SweeperTransactionPipeline, }; use crate::state::{Erc20Balances, EthBalance, State}; use crate::test_fixtures::{ @@ -1170,7 +1169,6 @@ fn state_equivalence() { }; let state = State { sweeper_funding: Default::default(), - sweeper_transactions: SweeperTransactionPipeline::new(TransactionNonce::ZERO), next_sweep_id: SweepId(0), ethereum_network: EthereumNetwork::Mainnet, ecdsa_key_name: "test_key".to_string(), diff --git a/rs/ethereum/cketh/minter/src/state/transactions/mod.rs b/rs/ethereum/cketh/minter/src/state/transactions/mod.rs index 793d80422edc..6017519dc61b 100644 --- a/rs/ethereum/cketh/minter/src/state/transactions/mod.rs +++ b/rs/ethereum/cketh/minter/src/state/transactions/mod.rs @@ -893,16 +893,26 @@ where buf } + let Self { + pending_requests, + processed_requests, + created_tx, + sent_tx, + finalized_tx, + next_nonce, + } = self; + // We can reorder request in `reschedule_request`. The audit log won't // reflect this change, so we must sort the queues before comparing them. ensure_eq!( - sorted_requests(&self.pending_requests), + sorted_requests(pending_requests), sorted_requests(&other.pending_requests) ); - ensure_eq!(self.created_tx, other.created_tx); - ensure_eq!(self.sent_tx, other.sent_tx); - ensure_eq!(self.finalized_tx, other.finalized_tx); - ensure_eq!(self.next_nonce, other.next_nonce); + ensure_eq!(processed_requests, &other.processed_requests); + ensure_eq!(created_tx, &other.created_tx); + ensure_eq!(sent_tx, &other.sent_tx); + ensure_eq!(finalized_tx, &other.finalized_tx); + ensure_eq!(next_nonce, &other.next_nonce); Ok(()) } @@ -1031,10 +1041,17 @@ impl WithdrawalTransactions { pub fn is_equivalent_to(&self, other: &Self) -> Result<(), String> { use ic_utils_ensure::ensure_eq; - ensure_eq!(self.maybe_reimburse, other.maybe_reimburse); - ensure_eq!(self.reimbursement_requests, other.reimbursement_requests); - ensure_eq!(self.reimbursed, other.reimbursed); - self.pipeline.is_equivalent_to(&other.pipeline) + let Self { + pipeline, + maybe_reimburse, + reimbursement_requests, + reimbursed, + } = self; + + ensure_eq!(maybe_reimburse, &other.maybe_reimburse); + ensure_eq!(reimbursement_requests, &other.reimbursement_requests); + ensure_eq!(reimbursed, &other.reimbursed); + pipeline.is_equivalent_to(&other.pipeline) } pub fn next_transaction_nonce(&self) -> TransactionNonce { diff --git a/rs/ethereum/cketh/minter/src/sweep.rs b/rs/ethereum/cketh/minter/src/sweep.rs index 2935fd070bea..babeeb2f8276 100644 --- a/rs/ethereum/cketh/minter/src/sweep.rs +++ b/rs/ethereum/cketh/minter/src/sweep.rs @@ -51,7 +51,7 @@ pub async fn process_sweeper_transactions(time_provider: T) { } }; - if read_state(|s| !s.sweeper_transactions.has_pending_requests()) { + if read_state(|s| !s.automatic_deposits.has_pending_sweeps()) { return; } @@ -86,7 +86,7 @@ pub async fn process_sweeper_transactions(time_provider: T) { send_transactions_batch(sender, latest_transaction_count).await; finalize_transactions_batch(sender, &time_provider).await; - if read_state(|s| s.sweeper_transactions.has_pending_requests()) { + if read_state(|s| s.automatic_deposits.has_pending_sweeps()) { schedule_retry(time_provider); } } @@ -103,15 +103,15 @@ async fn resubmit_transactions_batch( gas_fee_estimate: &GasFeeEstimate, time_provider: &T, ) { - if read_state(|s| s.sweeper_transactions.is_sent_tx_empty()) { + if read_state(|s| s.automatic_deposits.is_sent_sweep_tx_empty()) { return; } let Some(latest_transaction_count) = latest_transaction_count else { return; }; let transactions_to_resubmit = read_state(|s| { - s.sweeper_transactions - .create_resubmit_transactions(latest_transaction_count, gas_fee_estimate.clone()) + s.automatic_deposits + .create_resubmit_sweep_transactions(latest_transaction_count, gas_fee_estimate.clone()) }); for result in transactions_to_resubmit { match result { @@ -146,11 +146,11 @@ fn create_transactions_batch( time_provider: &T, ) { for request in read_state(|s| { - s.sweeper_transactions - .requests_batch(SWEEP_REQUESTS_BATCH_SIZE) + s.automatic_deposits + .sweep_requests_batch(SWEEP_REQUESTS_BATCH_SIZE) }) { let ethereum_network = read_state(State::ethereum_network); - let nonce = read_state(|s| s.sweeper_transactions.next_transaction_nonce()); + let nonce = read_state(|s| s.automatic_deposits.next_sweeper_transaction_nonce()); let sweep_id = request.id; match request.create_transaction( nonce, @@ -179,7 +179,7 @@ fn create_transactions_batch( INFO, "[process_sweeper_transactions]: Sweep {sweep_id:?} has prepaid gas {allowed_max_transaction_fee:?}, below the current transaction fee {actual_max_transaction_fee:?}. Sweep moved back to end of queue." ); - mutate_state(|s| s.sweeper_transactions.reschedule_request(sweep_id)); + mutate_state(|s| s.automatic_deposits.reschedule_sweep_request(sweep_id)); } } } @@ -187,8 +187,8 @@ fn create_transactions_batch( async fn sign_transactions_batch(time_provider: &T) { let transactions_batch: Vec<_> = read_state(|s| { - s.sweeper_transactions - .transactions_to_sign_batch(SWEEP_TRANSACTIONS_TO_SIGN_BATCH_SIZE) + s.automatic_deposits + .sweep_transactions_to_sign_batch(SWEEP_TRANSACTIONS_TO_SIGN_BATCH_SIZE) }); let results = join_all( transactions_batch @@ -229,7 +229,7 @@ async fn send_transactions_batch( return; }; let transactions_to_send: Vec<_> = read_state(|s| { - s.sweeper_transactions.transactions_to_send_batch( + s.automatic_deposits.sweep_transactions_to_send_batch( latest_transaction_count, SWEEP_TRANSACTIONS_TO_SEND_BATCH_SIZE, ) @@ -238,14 +238,14 @@ async fn send_transactions_batch( } async fn finalize_transactions_batch(sender: Address, time_provider: &T) { - if read_state(|s| s.sweeper_transactions.is_sent_tx_empty()) { + if read_state(|s| s.automatic_deposits.is_sent_sweep_tx_empty()) { return; } match finalized_transaction_count(sender).await { Ok(finalized_tx_count) => { let txs_to_finalize = read_state(|s| { - s.sweeper_transactions - .sent_transactions_to_finalize(&finalized_tx_count) + s.automatic_deposits + .sent_sweep_transactions_to_finalize(&finalized_tx_count) }); if let Some(receipts) = fetch_finalized_receipts(txs_to_finalize).await { for (sweep_id, transaction_receipt) in receipts {