refactor(cketh): move the sweeper pipeline into AutomaticDeposits

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<InitArg> 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<InitArg> 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(),
         };

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)
         );
     }

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<EventSource, MintedEvent>,
     pub invalid_events: BTreeMap<EventSource, InvalidEventReason>,
     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<Address, BTreeSet<BlockNumber>>,
@@ -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,16 +708,15 @@ 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
         );
         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)
     }

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,

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<AttestationRequest, TransactionSignature>,
+    /// 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<SweepRequest> {
+        self.sweeper_transactions
+            .requests_batch(requested_batch_size)
+    }
+
+    pub fn create_resubmit_sweep_transactions(
+        &self,
+        latest_transaction_count: TransactionCount,
+        current_gas_fee: GasFeeEstimate,
+    ) -> Vec<Result<(SweepId, SweepTransaction), ResubmitTransactionError<SweepId>>> {
+        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<Signed<SweepTransaction>> {
+        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<Hash, SweepId> {
+        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<SweepTransaction>,
+    ) {
+        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<SweepTransaction> {
+        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),
         }
     }
 }

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(),

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 {