diff --git a/rs/ethereum/cketh/minter/src/main.rs b/rs/ethereum/cketh/minter/src/main.rs index d6da7eb9dddf..7254e4e3b5fa 100644 --- a/rs/ethereum/cketh/minter/src/main.rs +++ b/rs/ethereum/cketh/minter/src/main.rs @@ -1516,6 +1516,72 @@ fn http_request(req: HttpRequest) -> HttpResponse { "Age of the sweeper funding awaiting finalization; 0 if none is outstanding.", )?; + let withdrawal_pipeline = s.withdrawal_transactions.pipeline(); + let sweeper_pipeline = s.automatic_deposits.sweeper_pipeline(); + let receipt_fetch = [ + ("withdrawal", withdrawal_pipeline.receipt_fetch_counters()), + ("sweeper", sweeper_pipeline.receipt_fetch_counters()), + ]; + w.gauge_vec( + "cketh_minter_receipt_fetch_window", + "Maximum pipeline ids one finalization round fetches receipts for, per pipeline.", + )? + .value( + &[("pipeline", "withdrawal")], + withdrawal_pipeline.receipt_fetch_window() as f64, + )? + .value( + &[("pipeline", "sweeper")], + sweeper_pipeline.receipt_fetch_window() as f64, + )?; + w.gauge_vec( + "cketh_minter_receipt_fetch_rounds_since_chain_read", + "Consecutive rounds failing or skipping the finalized-count read, per pipeline.", + )? + .value( + &[("pipeline", "withdrawal")], + withdrawal_pipeline.rounds_since_chain_read() as f64, + )? + .value( + &[("pipeline", "sweeper")], + sweeper_pipeline.rounds_since_chain_read() as f64, + )?; + let mut receipt_lookups = w.counter_vec( + "cketh_minter_receipt_lookups_total", + "Transaction receipt lookups, by pipeline and outcome. Resets on upgrade.", + )?; + for (pipeline, counters) in receipt_fetch { + receipt_lookups = receipt_lookups + .value( + &[("pipeline", pipeline), ("outcome", "receipt")], + counters.receipts as f64, + )? + .value( + &[("pipeline", pipeline), ("outcome", "not_mined")], + counters.not_mined as f64, + )? + .value( + &[("pipeline", pipeline), ("outcome", "error")], + counters.failures as f64, + )?; + } + let mut abandoned_rounds = w.counter_vec( + "cketh_minter_receipt_fetch_abandoned_rounds_total", + "Rounds dropped because two receipts named the same id. Resets on upgrade.", + )?; + for (pipeline, counters) in receipt_fetch { + abandoned_rounds = abandoned_rounds + .value(&[("pipeline", pipeline)], counters.abandoned_rounds as f64)?; + } + let mut stalled_ids = w.counter_vec( + "cketh_minter_receipt_fetch_stalled_ids_total", + "Ids with no receipt though their nonce is finalized. Resets on upgrade.", + )?; + for (pipeline, counters) in receipt_fetch { + stalled_ids = stalled_ids + .value(&[("pipeline", pipeline)], counters.stalled_ids as f64)?; + } + w.encode_gauge( "cketh_minter_last_max_fee_per_gas", s.last_transaction_price_estimate diff --git a/rs/ethereum/cketh/minter/src/state.rs b/rs/ethereum/cketh/minter/src/state.rs index d908b125794b..951286a45dd3 100644 --- a/rs/ethereum/cketh/minter/src/state.rs +++ b/rs/ethereum/cketh/minter/src/state.rs @@ -40,6 +40,7 @@ pub mod audit; pub mod automatic_deposits; pub mod eth_logs_scraping; pub mod event; +pub mod receipt_fetch; pub mod sweep_observations; pub mod sweeper_funding; pub mod transactions; 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 d1953b9a8b4a..7e9ffcbcdb8c 100644 --- a/rs/ethereum/cketh/minter/src/state/automatic_deposits/mod.rs +++ b/rs/ethereum/cketh/minter/src/state/automatic_deposits/mod.rs @@ -5,7 +5,6 @@ use crate::asset::{Asset, Erc20Asset, EthAsset}; use crate::attestation::AttestationRequest; use crate::balance_scan::batcher::Delegation; use crate::deposit_address::DepositAddress; -use crate::eth_rpc::Hash; use crate::eth_rpc_client::responses::{TransactionReceipt, TransactionStatus}; use crate::logs::INFO; use crate::numeric::{BlockNumber, Erc20Value, TransactionCount, TransactionNonce}; @@ -120,6 +119,15 @@ impl AutomaticDeposits { } } + /// The sweeper address' pipeline, whose receipt fetch the sweeper round drives. + pub fn sweeper_pipeline(&self) -> &SweeperTransactionPipeline { + &self.sweeper_transactions + } + + pub fn sweeper_pipeline_mut(&mut self) -> &mut SweeperTransactionPipeline { + &mut self.sweeper_transactions + } + pub fn has_pending_sweeps(&self) -> bool { self.sweeper_transactions.has_pending_requests() } @@ -168,14 +176,6 @@ impl AutomaticDeposits { .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) } diff --git a/rs/ethereum/cketh/minter/src/state/receipt_fetch.rs b/rs/ethereum/cketh/minter/src/state/receipt_fetch.rs new file mode 100644 index 000000000000..312fcdd061e6 --- /dev/null +++ b/rs/ethereum/cketh/minter/src/state/receipt_fetch.rs @@ -0,0 +1,227 @@ +//! How many pipeline ids one finalization round fetches transaction receipts for, and where in the +//! pending set the next round resumes. Soft state, reset on upgrade. + +#[cfg(test)] +mod tests; + +use crate::eth_rpc::Hash; +use std::collections::BTreeMap; +use std::ops::Bound; + +pub const INITIAL_RECEIPT_FETCH_WINDOW: usize = 10; +pub const MIN_RECEIPT_FETCH_WINDOW: usize = 1; +pub const MAX_RECEIPT_FETCH_WINDOW: usize = 20; +/// A pipeline whose chain read keeps failing makes no lookup, so has no failure to shrink its +/// window with, and would otherwise repeat at full cadence forever. +pub const ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING: u32 = 3; +/// One round out of this many is still attempted while rounds are being skipped. +pub const ROUNDS_PER_ATTEMPT_WHILE_SKIPPING: u32 = 4; + +/// What one round's receipt lookups returned. Only provider-level failures shrink the window: +/// "not mined" is the ordinary answer for a superseded resubmission. +#[derive(Clone, Copy, Debug, Default)] +pub struct RoundOutcome { + receipts: u32, + not_mined: u32, + failures: u32, + stalled_ids: u32, + abandoned: bool, +} + +impl RoundOutcome { + pub fn record_receipt(&mut self) { + self.receipts = self.receipts.saturating_add(1); + } + + pub fn record_not_mined(&mut self) { + self.not_mined = self.not_mined.saturating_add(1); + } + + pub fn record_failure(&mut self) { + self.failures = self.failures.saturating_add(1); + } + + pub fn record_stalled_id(&mut self) { + self.stalled_ids = self.stalled_ids.saturating_add(1); + } + + /// A flag rather than a count, so a round with several conflicting ids counts once. + pub fn abandon(&mut self) { + self.abandoned = true; + } + + pub fn receipts(&self) -> u32 { + self.receipts + } + + pub fn not_mined(&self) -> u32 { + self.not_mined + } + + pub fn failures(&self) -> u32 { + self.failures + } + + pub fn stalled_ids(&self) -> u32 { + self.stalled_ids + } + + pub fn is_abandoned(&self) -> bool { + self.abandoned + } + + pub fn lookups(&self) -> u32 { + self.receipts + .saturating_add(self.not_mined) + .saturating_add(self.failures) + } + + fn next_window(&self, window: usize) -> usize { + if self.abandoned || (self.lookups() > 0 && self.failures == self.lookups()) { + return MIN_RECEIPT_FETCH_WINDOW; + } + if self.failures > 0 { + return (window / 2).max(MIN_RECEIPT_FETCH_WINDOW); + } + window + .saturating_mul(2) + .clamp(MIN_RECEIPT_FETCH_WINDOW, MAX_RECEIPT_FETCH_WINDOW) + } +} + +/// One window per pipeline, so a sweeper problem cannot throttle user withdrawals. +#[derive(Clone, Copy, Eq, PartialEq, Debug)] +pub struct ReceiptFetchWindow { + window: usize, + cursor: Option, + rounds_since_chain_read: u32, + receipts_total: u64, + not_mined_total: u64, + failures_total: u64, + stalled_ids_total: u64, + abandoned_rounds_total: u64, +} + +impl Default for ReceiptFetchWindow { + fn default() -> Self { + Self { + window: INITIAL_RECEIPT_FETCH_WINDOW, + cursor: None, + rounds_since_chain_read: 0, + receipts_total: 0, + not_mined_total: 0, + failures_total: 0, + stalled_ids_total: 0, + abandoned_rounds_total: 0, + } + } +} + +impl ReceiptFetchWindow { + /// Takes whole ids, resuming past the cursor and wrapping around. A resubmitted id spans + /// several hashes of which only one has a receipt, so slicing the hash-keyed map would split it. + /// + /// The cursor moves past the selected ids whether or not their lookups succeed, so an id whose + /// receipt cannot be retrieved is retried only once the window has gone round the whole pending + /// set, rather than pinning every round onto itself and starving the ids behind it. + pub fn select_next_round(&mut self, pending: &BTreeMap) -> BTreeMap { + let by_id = group_by_id(pending); + let ids = self.next_ids(&by_id); + if let Some(last) = ids.last() { + self.cursor = Some(*last); + } + ids.iter() + .flat_map(|id| by_id[id].iter().map(move |hash| (*hash, *id))) + .collect() + } + + fn next_ids(&self, by_id: &BTreeMap>) -> Vec { + let after_cursor = match self.cursor { + Some(cursor) => Bound::Excluded(cursor), + None => Bound::Unbounded, + }; + by_id + .range((after_cursor, Bound::Unbounded)) + .map(|(id, _hashes)| *id) + .chain(by_id.keys().copied()) + .take(self.window.min(by_id.len())) + .collect() + } + + /// Deliberately not a general backoff: only a pipeline with no failure to shrink its window + /// with skips rounds. + pub fn should_skip_round(&self) -> bool { + self.rounds_since_chain_read >= ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING + && !(self.rounds_since_chain_read - ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING) + .is_multiple_of(ROUNDS_PER_ATTEMPT_WHILE_SKIPPING) + } + + pub fn record_round_without_chain_read(&mut self) { + self.rounds_since_chain_read = self.rounds_since_chain_read.saturating_add(1); + } + + /// An empty outcome still proves the providers answered, so rounds stop being skipped. + pub fn record_round(&mut self, outcome: RoundOutcome) { + self.rounds_since_chain_read = 0; + if outcome.lookups() == 0 { + return; + } + self.receipts_total = self + .receipts_total + .saturating_add(outcome.receipts() as u64); + self.not_mined_total = self + .not_mined_total + .saturating_add(outcome.not_mined() as u64); + self.failures_total = self + .failures_total + .saturating_add(outcome.failures() as u64); + self.stalled_ids_total = self + .stalled_ids_total + .saturating_add(outcome.stalled_ids() as u64); + if outcome.is_abandoned() { + self.abandoned_rounds_total = self.abandoned_rounds_total.saturating_add(1); + } + self.window = outcome.next_window(self.window); + } + + pub fn window(&self) -> usize { + self.window + } + + pub fn cursor(&self) -> Option { + self.cursor + } + + pub fn rounds_since_chain_read(&self) -> u32 { + self.rounds_since_chain_read + } + + pub fn counters(&self) -> ReceiptFetchCounters { + ReceiptFetchCounters { + receipts: self.receipts_total, + not_mined: self.not_mined_total, + failures: self.failures_total, + stalled_ids: self.stalled_ids_total, + abandoned_rounds: self.abandoned_rounds_total, + } + } +} + +/// Reset on upgrade, so alerts on these must be written against rates. +#[derive(Clone, Copy, Eq, PartialEq, Debug, Default)] +pub struct ReceiptFetchCounters { + pub receipts: u64, + pub not_mined: u64, + pub failures: u64, + /// Counted once per round, so an id that cannot be finalized keeps adding to it. + pub stalled_ids: u64, + pub abandoned_rounds: u64, +} + +fn group_by_id(pending: &BTreeMap) -> BTreeMap> { + let mut by_id: BTreeMap> = BTreeMap::new(); + for (hash, id) in pending { + by_id.entry(*id).or_default().push(*hash); + } + by_id +} diff --git a/rs/ethereum/cketh/minter/src/state/receipt_fetch/tests.rs b/rs/ethereum/cketh/minter/src/state/receipt_fetch/tests.rs new file mode 100644 index 000000000000..73f7e95ef1dc --- /dev/null +++ b/rs/ethereum/cketh/minter/src/state/receipt_fetch/tests.rs @@ -0,0 +1,357 @@ +use crate::eth_rpc::Hash; +use crate::numeric::LedgerBurnIndex; +use crate::state::receipt_fetch::{ + INITIAL_RECEIPT_FETCH_WINDOW, MAX_RECEIPT_FETCH_WINDOW, MIN_RECEIPT_FETCH_WINDOW, + ROUNDS_PER_ATTEMPT_WHILE_SKIPPING, ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING, + ReceiptFetchCounters, ReceiptFetchWindow, RoundOutcome, +}; +use std::collections::BTreeMap; + +mod adaptation { + use super::*; + + #[test] + fn should_double_the_window_when_no_lookup_failed() { + let mut window = window_of(1); + + for expected in [2, 4, 8, 16] { + window.record_round(round(1, 0, 0)); + assert_eq!(window.window(), expected); + } + } + + #[test] + fn should_not_grow_past_the_ceiling() { + let mut window = window_of(16); + + window.record_round(round(3, 0, 0)); + assert_eq!(window.window(), MAX_RECEIPT_FETCH_WINDOW); + + window.record_round(round(3, 0, 0)); + assert_eq!(window.window(), MAX_RECEIPT_FETCH_WINDOW); + } + + #[test] + fn should_halve_the_window_when_some_lookups_failed() { + let mut window = window_of(INITIAL_RECEIPT_FETCH_WINDOW); + + for expected in [5, 2, 1, 1] { + window.record_round(round(1, 0, 1)); + assert_eq!(window.window(), expected); + } + } + + #[test] + fn should_drop_to_the_floor_when_every_lookup_failed() { + let mut window = window_of(MAX_RECEIPT_FETCH_WINDOW); + + window.record_round(round(0, 0, 7)); + + assert_eq!(window.window(), MIN_RECEIPT_FETCH_WINDOW); + } + + #[test] + fn should_drop_to_the_floor_on_an_abandoned_round() { + let mut window = window_of(MAX_RECEIPT_FETCH_WINDOW); + let mut outcome = round(2, 0, 0); + outcome.abandon(); + + window.record_round(outcome); + + assert_eq!(window.window(), MIN_RECEIPT_FETCH_WINDOW); + } + + #[test] + fn should_count_a_round_with_several_conflicts_as_one_abandoned_round() { + let mut window = window_of(INITIAL_RECEIPT_FETCH_WINDOW); + let mut outcome = round(4, 0, 0); + outcome.abandon(); + outcome.abandon(); + + window.record_round(outcome); + + assert_eq!(window.counters().abandoned_rounds, 1); + } + + #[test] + fn should_not_shrink_on_transactions_that_were_not_mined() { + let mut window = window_of(INITIAL_RECEIPT_FETCH_WINDOW); + + window.record_round(round(1, 9, 0)); + + assert_eq!(window.window(), MAX_RECEIPT_FETCH_WINDOW); + } + + #[test] + fn should_leave_the_window_alone_on_a_round_with_nothing_to_fetch() { + let mut window = window_of(INITIAL_RECEIPT_FETCH_WINDOW); + + window.record_round(RoundOutcome::default()); + window.record_round_without_chain_read(); + + assert_eq!(window.window(), INITIAL_RECEIPT_FETCH_WINDOW); + assert_eq!(window.rounds_since_chain_read(), 1); + } + + #[test] + fn should_accumulate_what_the_lookups_returned() { + let mut window = window_of(INITIAL_RECEIPT_FETCH_WINDOW); + let mut outcome = round(1, 2, 3); + outcome.record_stalled_id(); + + window.record_round(outcome); + window.record_round(outcome); + + assert_eq!( + window.counters(), + ReceiptFetchCounters { + receipts: 2, + not_mined: 4, + failures: 6, + stalled_ids: 2, + abandoned_rounds: 0, + } + ); + } +} + +mod selection { + use super::*; + + #[test] + fn should_select_nothing_when_nothing_is_pending() { + let mut window = ReceiptFetchWindow::::default(); + + assert_eq!(window.select_next_round(&BTreeMap::new()), BTreeMap::new()); + assert_eq!(window.cursor(), None); + } + + #[test] + fn should_select_every_transaction_of_the_ids_it_takes() { + let pending = pending_with_variants(&[(1, 3), (2, 1), (3, 2)]); + let mut window = window_of(2); + + let selected = window.select_next_round(&pending); + + assert_eq!(ids_of(&selected), vec![id(1), id(2)]); + assert_eq!(selected.len(), 4); + assert_eq!( + selected, + pending + .iter() + .filter(|(_hash, id)| **id != LedgerBurnIndex::new(3)) + .map(|(hash, id)| (*hash, *id)) + .collect::>() + ); + } + + #[test] + fn should_not_split_the_transactions_of_one_id_across_rounds() { + let pending = pending_with_variants(&[(1, 3), (2, 3), (3, 3)]); + let mut window = window_of(1); + + for expected in [id(1), id(2), id(3), id(1)] { + let selected = window.select_next_round(&pending); + assert_eq!(ids_of(&selected), vec![expected]); + assert_eq!(selected.len(), 3); + } + } + + #[test] + fn should_select_at_most_the_whole_pending_set() { + let pending = pending_with_variants(&[(1, 1), (2, 1)]); + let mut window = window_of(MAX_RECEIPT_FETCH_WINDOW); + + let selected = window.select_next_round(&pending); + + assert_eq!(ids_of(&selected), vec![id(1), id(2)]); + assert_eq!(window.cursor(), Some(id(2))); + } +} + +mod cursor { + use super::*; + + #[test] + fn should_walk_the_pending_set_round_after_round() { + let pending = pending_with_variants(&[(1, 1), (2, 1), (3, 1), (4, 1), (5, 1)]); + let mut window = window_of(2); + + let rounds: Vec<_> = (0..4) + .map(|_| { + let selected = ids_of(&window.select_next_round(&pending)); + (selected, window.cursor()) + }) + .collect(); + + assert_eq!( + rounds, + vec![ + (vec![id(1), id(2)], Some(id(2))), + (vec![id(3), id(4)], Some(id(4))), + (vec![id(1), id(5)], Some(id(1))), + (vec![id(2), id(3)], Some(id(3))), + ] + ); + } + + #[test] + fn should_resume_past_the_cursor_after_the_ids_around_it_finalized() { + let mut window = window_of(2); + + window.select_next_round(&pending_with_variants(&[(1, 1), (2, 1), (3, 1), (4, 1)])); + assert_eq!(window.cursor(), Some(id(2))); + + let selected = window.select_next_round(&pending_with_variants(&[(3, 1), (4, 1)])); + + assert_eq!(ids_of(&selected), vec![id(3), id(4)]); + assert_eq!(window.cursor(), Some(id(4))); + } + + #[test] + fn should_wrap_around_when_the_cursor_is_past_everything_pending() { + let mut window = window_of(2); + + window.select_next_round(&pending_with_variants(&[(1, 1), (2, 1), (9, 1)])); + assert_eq!(window.cursor(), Some(id(2))); + + let selected = window.select_next_round(&pending_with_variants(&[(1, 1), (2, 1)])); + + assert_eq!(ids_of(&selected), vec![id(1), id(2)]); + } + + #[test] + fn should_take_the_whole_set_once_when_the_window_spans_it_from_a_cursor() { + let pending = pending_with_variants(&[(1, 1), (2, 1), (3, 1)]); + let mut window = window_of(3); + window.cursor = Some(id(2)); + + let selected = window.select_next_round(&pending); + + assert_eq!(ids_of(&selected), vec![id(1), id(2), id(3)]); + assert_eq!( + selected.len(), + pending.len(), + "a wrap across the whole set must take every transaction exactly once" + ); + assert_eq!(window.cursor(), Some(id(2))); + } + + #[test] + fn should_leave_the_cursor_alone_on_a_round_that_selected_nothing() { + let mut window = window_of(2); + window.select_next_round(&pending_with_variants(&[(1, 1), (2, 1)])); + + window.select_next_round(&BTreeMap::new()); + + assert_eq!(window.cursor(), Some(id(2))); + } +} + +mod skipping { + use super::*; + + #[test] + fn should_attempt_every_round_below_the_threshold() { + let mut window = ReceiptFetchWindow::::default(); + + for _ in 0..ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING { + assert!(!skip_round(&mut window)); + window.record_round_without_chain_read(); + } + assert!(!skip_round(&mut window)); + } + + #[test] + fn should_attempt_one_round_in_a_few_past_the_threshold() { + let mut window = ReceiptFetchWindow::::default(); + for _ in 0..ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING { + window.record_round_without_chain_read(); + } + + let mut attempted = 0; + let rounds = 4 * ROUNDS_PER_ATTEMPT_WHILE_SKIPPING; + for _ in 0..rounds { + if !skip_round(&mut window) { + attempted += 1; + window.record_round_without_chain_read(); + } + } + + assert_eq!(attempted, rounds / ROUNDS_PER_ATTEMPT_WHILE_SKIPPING); + } + + #[test] + fn should_stop_skipping_once_a_round_read_the_chain() { + let mut window = ReceiptFetchWindow::::default(); + for _ in 0..10 * ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING { + window.record_round_without_chain_read(); + } + assert!(skip_round(&mut window)); + + window.record_round(RoundOutcome::default()); + + assert_eq!(window.rounds_since_chain_read(), 0); + assert!(!skip_round(&mut window)); + } + + fn skip_round(window: &mut ReceiptFetchWindow) -> bool { + let skip = window.should_skip_round(); + if skip { + window.record_round_without_chain_read(); + } + skip + } +} + +fn round(receipts: u32, not_mined: u32, failures: u32) -> RoundOutcome { + let mut outcome = RoundOutcome::default(); + for _ in 0..receipts { + outcome.record_receipt(); + } + for _ in 0..not_mined { + outcome.record_not_mined(); + } + for _ in 0..failures { + outcome.record_failure(); + } + outcome +} + +fn window_of(window: usize) -> ReceiptFetchWindow { + ReceiptFetchWindow { + window, + ..Default::default() + } +} + +fn pending_with_variants(ids: &[(u8, u8)]) -> BTreeMap { + ids.iter() + .flat_map(|(id, variants)| { + (0..*variants).map(move |variant| { + ( + scrambled_hash(*id, variant), + LedgerBurnIndex::new(*id as u64), + ) + }) + }) + .collect() +} + +fn ids_of(selected: &BTreeMap) -> Vec { + let mut ids: Vec<_> = selected.values().copied().collect(); + ids.sort_unstable(); + ids.dedup(); + ids +} + +fn id(id: u8) -> LedgerBurnIndex { + LedgerBurnIndex::new(id as u64) +} + +fn scrambled_hash(id: u8, variant: u8) -> Hash { + let mut bytes = [0_u8; 32]; + bytes[0] = u8::MAX - id; + bytes[1] = variant; + Hash(bytes) +} diff --git a/rs/ethereum/cketh/minter/src/state/tests.rs b/rs/ethereum/cketh/minter/src/state/tests.rs index ea7284781e2f..7857d2396b36 100644 --- a/rs/ethereum/cketh/minter/src/state/tests.rs +++ b/rs/ethereum/cketh/minter/src/state/tests.rs @@ -1540,7 +1540,7 @@ mod eth_balance { use crate::eth_rpc_client::responses::{TransactionReceipt, TransactionStatus}; use crate::lifecycle::EthereumNetwork; use crate::numeric::{ - BlockNumber, GasAmount, LedgerBurnIndex, TransactionNonce, Wei, WeiPerGas, + BlockNumber, GasAmount, LedgerBurnIndex, TransactionCount, TransactionNonce, Wei, WeiPerGas, }; use crate::state::audit::{EventType, apply_state_transition}; use crate::state::tests::checked_sub; @@ -1825,6 +1825,89 @@ mod eth_balance { ); } + #[test] + fn should_finalize_a_later_nonce_while_an_earlier_one_stays_pending() { + let spiked_fee = GasFeeEstimate { + base_fee_per_gas: WeiPerGas::from(1_000_000_u32), + max_priority_fee_per_gas: WeiPerGas::from(1_000_000_u32), + }; + let straggler = withdrawal_flow(LedgerBurnIndex::new(0), TransactionNonce::ZERO); + let ahead = withdrawal_flow(LedgerBurnIndex::new(1), TransactionNonce::ONE); + + let mut out_of_order = deposited_state(); + let straggler_tx = straggler.send(&mut out_of_order); + let ahead_tx = ahead.send(&mut out_of_order); + ahead.finalize(&mut out_of_order, &ahead_tx); + + assert_eq!( + out_of_order + .withdrawal_transactions + .finalized_transactions_iter() + .map(|(nonce, id, _tx)| (*nonce, *id)) + .collect::>(), + vec![(TransactionNonce::ONE, LedgerBurnIndex::new(1))], + "the later nonce must finalize on its own" + ); + assert_eq!( + out_of_order + .withdrawal_transactions + .create_resubmit_transactions(TransactionCount::TWO, spiked_fee.clone()), + vec![], + "the chain has moved past the straggler's nonce, so it must not be resubmitted" + ); + assert!( + !out_of_order + .withdrawal_transactions + .create_resubmit_transactions(TransactionCount::ZERO, spiked_fee) + .is_empty(), + "a straggler the chain has not passed is still considered for resubmission, so the \ + emptiness above is the nonce filter and not the fee" + ); + + straggler.finalize(&mut out_of_order, &straggler_tx); + + let mut in_order = deposited_state(); + let straggler_tx = straggler.send(&mut in_order); + let ahead_tx = ahead.send(&mut in_order); + straggler.finalize(&mut in_order, &straggler_tx); + ahead.finalize(&mut in_order, &ahead_tx); + assert_eq!( + out_of_order.withdrawal_transactions, in_order.withdrawal_transactions, + "both withdrawals must end up finalized, whichever order their receipts arrived in" + ); + assert_eq!(out_of_order.eth_balance, in_order.eth_balance); + } + + fn deposited_state() -> State { + let mut state = initial_state(); + apply_state_transition( + &mut state, + &EventType::AcceptedDeposit(received_eth_event()), + ); + state + } + + fn withdrawal_flow( + ledger_burn_index: LedgerBurnIndex, + nonce: TransactionNonce, + ) -> WithdrawalFlow { + WithdrawalFlow { + nonce, + ..WithdrawalFlow::for_request(EthWithdrawalRequest { + withdrawal_amount: Wei::new(4_000_000_000_000_000), + destination: "0xb44B5e756A894775FC32EDdf3314Bb1B1944dC34" + .parse() + .unwrap(), + ledger_burn_index, + from: "k2t6j-2nvnp-4zjm3-25dtz-6xhaa-c7boj-5gayf-oj3xs-i43lp-teztq-6ae" + .parse() + .unwrap(), + from_subaccount: None, + created_at: Some(1699527697000000000), + }) + } + } + #[test] fn should_update_after_successful_and_failed_sweeper_funding() { let mut state_before_funding = initial_state(); @@ -2021,6 +2104,11 @@ mod eth_balance { } fn apply(self, state: &mut State) -> TransactionReceipt { + let signed_tx = self.send(state); + self.finalize(state, &signed_tx) + } + + fn send(&self, state: &mut State) -> SignedEip1559TransactionRequest { let accepted_withdrawal_request_event = accepted_withdrawal_request_event(self.withdrawal_request.clone()); apply_state_transition(state, &accepted_withdrawal_request_event); @@ -2029,7 +2117,7 @@ mod eth_balance { .withdrawal_request .create_transaction( self.nonce, - self.tx_fee, + self.tx_fee.clone(), self.gas_limit, EthereumNetwork::Sepolia, ) @@ -2047,8 +2135,7 @@ mod eth_balance { r: Default::default(), s: Default::default(), }; - let signed_tx = - SignedEip1559TransactionRequest::from((transaction.clone(), dummy_signature)); + let signed_tx = SignedEip1559TransactionRequest::from((transaction, dummy_signature)); apply_state_transition( state, &EventType::SignedTransaction { @@ -2056,7 +2143,14 @@ mod eth_balance { transaction: signed_tx.clone(), }, ); + signed_tx + } + fn finalize( + &self, + state: &mut State, + signed_tx: &SignedEip1559TransactionRequest, + ) -> TransactionReceipt { let tx_receipt = TransactionReceipt { block_hash: "0xce67a85c9fb8bc50213815c32814c159fd75160acf7cb8631e8e7b7cf7f1d472" .parse() diff --git a/rs/ethereum/cketh/minter/src/state/transactions/mod.rs b/rs/ethereum/cketh/minter/src/state/transactions/mod.rs index 136afda6f7e6..3af23860c8b0 100644 --- a/rs/ethereum/cketh/minter/src/state/transactions/mod.rs +++ b/rs/ethereum/cketh/minter/src/state/transactions/mod.rs @@ -17,6 +17,7 @@ use crate::numeric::{ CkTokenAmount, Erc20Value, GasAmount, LedgerBurnIndex, LedgerMintIndex, TransactionCount, TransactionNonce, Wei, }; +use crate::state::receipt_fetch::{ReceiptFetchCounters, ReceiptFetchWindow, RoundOutcome}; use crate::sweeper_contract::{SweepItem, encode_sweep_erc20_batch, encode_sweep_eth_batch}; use crate::tx::{ Eip1559TransactionRequest, Finalized, FinalizedEip1559Transaction, GasFeeEstimate, @@ -568,6 +569,9 @@ pub struct TransactionPipeline { sent_tx: MultiKeyMap>>, finalized_tx: MultiKeyMap>, next_nonce: TransactionNonce, + /// Ids the next finalization round fetches receipts for, and where in the pending set it + /// resumes. Not event-sourced, so it is reset on upgrade. + receipt_fetch: ReceiptFetchWindow, } /// The pipeline sending from the minter's main address, on which user withdrawals travel. @@ -641,6 +645,7 @@ where sent_tx: MultiKeyMap::default(), finalized_tx: MultiKeyMap::default(), next_nonce, + receipt_fetch: ReceiptFetchWindow::default(), } } @@ -809,6 +814,41 @@ where assert_eq!(self.created_tx.try_insert(nonce, *id, new_tx), Ok(())); } + /// The transactions whose receipts the next round fetches: as many pending ids as this + /// pipeline's window allows, resumed past where the previous round stopped. + pub fn select_receipt_fetch_round( + &mut self, + finalized_transaction_count: &TransactionCount, + ) -> BTreeMap { + let pending = self.sent_transactions_to_finalize(finalized_transaction_count); + self.receipt_fetch.select_next_round(&pending) + } + + /// Records how a round that reached its receipt lookups fared, adapting the window to it. + pub fn record_receipt_fetch_round(&mut self, outcome: RoundOutcome) { + self.receipt_fetch.record_round(outcome) + } + + pub fn should_skip_receipt_fetch_round(&self) -> bool { + self.receipt_fetch.should_skip_round() + } + + pub fn record_round_without_chain_read(&mut self) { + self.receipt_fetch.record_round_without_chain_read() + } + + pub fn receipt_fetch_window(&self) -> usize { + self.receipt_fetch.window() + } + + pub fn rounds_since_chain_read(&self) -> u32 { + self.receipt_fetch.rounds_since_chain_read() + } + + pub fn receipt_fetch_counters(&self) -> ReceiptFetchCounters { + self.receipt_fetch.counters() + } + pub fn sent_transactions_to_finalize( &self, finalized_transaction_count: &TransactionCount, @@ -1040,6 +1080,9 @@ where sent_tx, finalized_tx, next_nonce, + // Not event-sourced: a replayed pipeline has a default window while the live one may + // have advanced, so it is deliberately left out of the comparison. + receipt_fetch: _, } = self; // We can reorder request in `reschedule_request`. The audit log won't @@ -1194,6 +1237,15 @@ impl WithdrawalTransactions { pipeline.is_equivalent_to(&other.pipeline) } + /// The pipeline carrying user withdrawals, whose receipt fetch the finalization round drives. + pub fn pipeline(&self) -> &MinterTransactionPipeline { + &self.pipeline + } + + pub fn pipeline_mut(&mut self) -> &mut MinterTransactionPipeline { + &mut self.pipeline + } + pub fn next_transaction_nonce(&self) -> TransactionNonce { self.pipeline.next_transaction_nonce() } diff --git a/rs/ethereum/cketh/minter/src/state/transactions/request.rs b/rs/ethereum/cketh/minter/src/state/transactions/request.rs index 2a92f9e0fe75..1930d13b152f 100644 --- a/rs/ethereum/cketh/minter/src/state/transactions/request.rs +++ b/rs/ethereum/cketh/minter/src/state/transactions/request.rs @@ -24,6 +24,9 @@ pub trait PipelineRequest { /// for sweeps. type Id: Copy + Ord + fmt::Debug; + /// The log prefix of the task driving this pipeline. + const TASK_NAME: &'static str; + /// The transaction this request turns into. type Transaction: SignableTransaction; @@ -64,6 +67,8 @@ impl PipelineRequest for WithdrawalRequest { type Transaction = Eip1559TransactionRequest; type Error = CreateTransactionError; + const TASK_NAME: &'static str = "finalize_transactions_batch"; + fn id(&self) -> LedgerBurnIndex { self.cketh_ledger_burn_index() } @@ -205,6 +210,8 @@ impl PipelineRequest for SweepRequest { type Transaction = SweepTransaction; type Error = CreateSweepTransactionError; + const TASK_NAME: &'static str = "process_sweeper_transactions"; + fn id(&self) -> SweepId { self.id } diff --git a/rs/ethereum/cketh/minter/src/state/transactions/tests.rs b/rs/ethereum/cketh/minter/src/state/transactions/tests.rs index 8ba22e8b5016..1a7b508d5a2c 100644 --- a/rs/ethereum/cketh/minter/src/state/transactions/tests.rs +++ b/rs/ethereum/cketh/minter/src/state/transactions/tests.rs @@ -3884,6 +3884,7 @@ impl WithdrawalTransactionsBuilder { sent_tx: self.sent_tx, finalized_tx: self.finalized_tx, next_nonce: self.next_nonce, + receipt_fetch: Default::default(), }, maybe_reimburse: self.maybe_reimburse, reimbursement_requests: self.reimbursement_requests, diff --git a/rs/ethereum/cketh/minter/src/sweep/mod.rs b/rs/ethereum/cketh/minter/src/sweep/mod.rs index 1715e76bc9eb..72565b560d97 100644 --- a/rs/ethereum/cketh/minter/src/sweep/mod.rs +++ b/rs/ethereum/cketh/minter/src/sweep/mod.rs @@ -5,9 +5,9 @@ //! It drives the sweeper [`TransactionPipeline`] through the same //! create → sign → send → resubmit → finalize state machine as user withdrawals //! ([`crate::withdraw`]), reusing that module's sender-agnostic RPC helpers -//! (`latest_transaction_count`, `finalized_transaction_count`, `send_signed_transactions`, -//! `fetch_finalized_receipts`), but signing with the sweeper derivation path (`[3]`) and reading -//! the sweeper address' own transaction count. +//! (`latest_transaction_count`, `send_signed_transactions`, `fetch_receipts_for_round`), but +//! signing with the sweeper derivation path (`[3]`) and reading the sweeper address' own +//! transaction count. #[cfg(test)] mod tests; @@ -40,10 +40,7 @@ use crate::{ }, time::TimeProvider, tx::{AuthorizationRequest, GasFeeEstimate, lazy_refresh_gas_fee_estimate, sign_digest}, - withdraw::{ - fetch_finalized_receipts, finalized_transaction_count, latest_transaction_count, - send_signed_transactions, - }, + withdraw::{fetch_receipts_for_round, latest_transaction_count, send_signed_transactions}, }; use evm_rpc_client::{CandidResponseConverter, DoubleCycles, EvmRpcClient}; use futures::future::join_all; @@ -561,36 +558,26 @@ async fn send_transactions_batch( send_signed_transactions(sender, &transactions_to_send).await; } -async fn finalize_transactions_batch(sender: Address, time_provider: &T) { +async fn finalize_transactions_batch(sender: Address, runtime: &R) { 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.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 { - mutate_state(|s| { - process_event( - s, - EventType::FinalizedSweeperTransaction { - sweep_id, - transaction_receipt: transaction_receipt.into(), - }, - time_provider, - ); - }); - } - } - } - Err(e) => { - log!( - INFO, - "[process_sweeper_transactions]: failed to get finalized transaction count: {e:?}" + + let receipts = fetch_receipts_for_round(sender, runtime, |s| { + s.automatic_deposits.sweeper_pipeline_mut() + }) + .await; + + for (sweep_id, transaction_receipt) in receipts { + mutate_state(|s| { + process_event( + s, + EventType::FinalizedSweeperTransaction { + sweep_id, + transaction_receipt: transaction_receipt.into(), + }, + runtime, ); - } + }); } } diff --git a/rs/ethereum/cketh/minter/src/sweep/tests.rs b/rs/ethereum/cketh/minter/src/sweep/tests.rs index a4f9e33fe47d..d4f7fb3785a7 100644 --- a/rs/ethereum/cketh/minter/src/sweep/tests.rs +++ b/rs/ethereum/cketh/minter/src/sweep/tests.rs @@ -9,6 +9,7 @@ use crate::numeric::{BlockNumber, GasAmount, TransactionNonce, Wei, WeiPerGas}; use crate::state::audit::{EventType, apply_state_transition, process_event}; use crate::state::eth_logs_scraping::LogScrapings; use crate::state::event::AutomaticDeposit; +use crate::state::receipt_fetch::ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING; use crate::state::transactions::{PipelineRequest, SweepId, SweepRequest}; use crate::state::{State, mutate_state, read_state}; use crate::storage::with_event_iter; @@ -16,13 +17,14 @@ use crate::sweep::create_pending_sweeper_requests; use crate::test_fixtures::mock::MockCanisterRuntime; use crate::test_fixtures::{ LATEST_BLOCK, account, another_account, automatic_deposit, delegation_response, - deposit_address, gas_fee_estimate, init_state, initial_state, only_one, prepay_sweep_gas, + deposit_address, gas_fee_estimate, init_state, initial_state, mock, only_one, prepay_sweep_gas, state_with_deposit_helper, stub_rpc_client, transaction_signature, usdc, usdt, }; use crate::tx::{ AuthorizationRequest, GasFeeEstimate, SignableTransaction, Signed, SignedAuthorization, TransactionSignature, }; +use crate::withdraw::fetch_receipts_for_round; use ethnum::u256; use evm_rpc_types::{Hex, MultiRpcResult}; use ic_canister_runtime::IcError; @@ -651,6 +653,45 @@ fn finalize_sweep_through_the_event_log(request: &SweepRequest, runtime: &MockCa }); } +#[tokio::test] +async fn should_skip_a_sweeper_round_without_touching_the_withdrawal_window() { + init_state(initial_state()); + mutate_state(|s| { + for _ in 0..=ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING { + s.automatic_deposits + .sweeper_pipeline_mut() + .record_round_without_chain_read(); + } + }); + let withdrawals_before = read_state(|s| s.withdrawal_transactions.clone()); + + let receipts: BTreeMap = + fetch_receipts_for_round(Address::new([0_u8; 20]), &no_rpc_runtime(), |s| { + s.automatic_deposits.sweeper_pipeline_mut() + }) + .await; + + assert_eq!(receipts, BTreeMap::new()); + assert_eq!( + read_state(|s| s + .automatic_deposits + .sweeper_pipeline() + .rounds_since_chain_read()), + ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING + 2 + ); + assert_eq!( + read_state(|s| s.withdrawal_transactions.clone()), + withdrawals_before, + "a sweeper problem must not throttle user withdrawals" + ); +} + +fn no_rpc_runtime() -> mock::MockCanisterRuntime { + let mut runtime = mock::MockCanisterRuntime::new(); + runtime.expect_evm_rpc_client().never(); + runtime +} + fn one_pending_sweep() -> SweepRequest { only_one(&pending_sweeps()).clone() } diff --git a/rs/ethereum/cketh/minter/src/withdraw.rs b/rs/ethereum/cketh/minter/src/withdraw.rs index 774545b52a64..a612312fc2fc 100644 --- a/rs/ethereum/cketh/minter/src/withdraw.rs +++ b/rs/ethereum/cketh/minter/src/withdraw.rs @@ -1,3 +1,6 @@ +#[cfg(test)] +mod tests; + use crate::eth_rpc::Hash; use crate::{ MAIN_DERIVATION_PATH, @@ -14,9 +17,10 @@ use crate::{ State, TaskType, audit::{EventType, process_event}, minter_address, mutate_state, read_state, + receipt_fetch::RoundOutcome, transactions::{ CreateTransactionError, PipelineRequest, Reimbursed, ReimbursementIndex, - ReimbursementRequest, WithdrawalRequest, + ReimbursementRequest, TransactionPipeline, WithdrawalRequest, }, }, time::TimeProvider, @@ -428,48 +432,92 @@ pub(crate) async fn send_signed_transactions(sender: Address, time_provider: &T) { +async fn finalize_transactions_batch(sender: Address, runtime: &R) { if read_state(|s| s.withdrawal_transactions.is_sent_tx_empty()) { return; } - match finalized_transaction_count(sender).await { - Ok(finalized_tx_count) => { - let txs_to_finalize = read_state(|s| { - s.withdrawal_transactions - .sent_transactions_to_finalize(&finalized_tx_count) - }); - if let Some(receipts) = fetch_finalized_receipts(txs_to_finalize).await { - for (withdrawal_id, transaction_receipt) in receipts { - mutate_state(|s| { - process_event( - s, - EventType::FinalizedTransaction { - withdrawal_id, - transaction_receipt: transaction_receipt.into(), - }, - time_provider, - ); - }); - } - } + let receipts = fetch_receipts_for_round(sender, runtime, |s| { + s.withdrawal_transactions.pipeline_mut() + }) + .await; + + for (withdrawal_id, transaction_receipt) in receipts { + mutate_state(|s| { + process_event( + s, + EventType::FinalizedTransaction { + withdrawal_id, + transaction_receipt: transaction_receipt.into(), + }, + runtime, + ); + }); + } +} + +/// One round of a pipeline's receipt fetch, bounded by that pipeline's window. Both pipelines +/// reuse it: naming one of them picks its ids, so a round can never pair them up. +pub(crate) async fn fetch_receipts_for_round( + sender: Address, + runtime: &R, + pipeline: fn(&mut State) -> &mut TransactionPipeline, +) -> BTreeMap +where + Req: PipelineRequest + Clone + Eq + std::fmt::Debug, + Req::Transaction: Clone + Eq + std::fmt::Debug, + R: CanisterRuntime, +{ + let context = Req::TASK_NAME; + let skipped = mutate_state(|s| { + let pipeline = pipeline(s); + if !pipeline.should_skip_receipt_fetch_round() { + return None; } + pipeline.record_round_without_chain_read(); + Some(pipeline.rounds_since_chain_read()) + }); + if let Some(rounds_since_chain_read) = skipped { + log!( + INFO, + "[{context}]: SKIPPING: the last {rounds_since_chain_read} rounds could not read the \ + chain to fetch a single receipt" + ); + return BTreeMap::new(); + } + let finalized_tx_count = match finalized_transaction_count(sender, runtime).await { + Ok(finalized_tx_count) => finalized_tx_count, Err(e) => { - log!(INFO, "Failed to get finalized transaction count: {e:?}"); + log!( + INFO, + "[{context}]: failed to get the finalized transaction count of {sender}: {e:?}" + ); + mutate_state(|s| pipeline(s).record_round_without_chain_read()); + return BTreeMap::new(); } + }; + + let txs_to_finalize = + mutate_state(|s| pipeline(s).select_receipt_fetch_round(&finalized_tx_count)); + if txs_to_finalize.is_empty() { + mutate_state(|s| pipeline(s).record_receipt_fetch_round(RoundOutcome::default())); + return BTreeMap::new(); } + + let (receipts, outcome) = fetch_finalized_receipts(txs_to_finalize, runtime).await; + mutate_state(|s| pipeline(s).record_receipt_fetch_round(outcome)); + receipts } -/// Fetch the finalized receipts for the given (transaction hash → pipeline id) map. Returns `None` when -/// the batch should be retried later (a receipt fetch failed, or the same hash came back with two -/// different receipts). On success the map is keyed by pipeline id, and its keys are asserted to be -/// exactly the ids expected to finalize. Sender/id-agnostic, so both pipelines reuse it. -pub(crate) async fn fetch_finalized_receipts( +type ReceiptResult = + Result, MultiCallError>>; + +async fn fetch_finalized_receipts( txs_to_finalize: BTreeMap, -) -> Option> { - let expected_finalized_ids: BTreeSet = txs_to_finalize.values().copied().collect(); - let rpc_client = read_state(rpc_client); + runtime: &R, +) -> (BTreeMap, RoundOutcome) { + let rpc_client = runtime.evm_rpc_client(); let results = join_all(txs_to_finalize.keys().map(async |hash| { rpc_client .get_transaction_receipt(*hash) @@ -479,7 +527,17 @@ pub(crate) async fn fetch_finalized_receipts( .reduce_with_strategy(NoReduction) })) .await; + collect_finalized_receipts(txs_to_finalize, results) +} + +fn collect_finalized_receipts( + txs_to_finalize: BTreeMap, + results: Vec, +) -> (BTreeMap, RoundOutcome) { + let expected_finalized_ids: BTreeSet = txs_to_finalize.values().copied().collect(); + let mut outcome = RoundOutcome::default(); let mut receipts: BTreeMap = BTreeMap::new(); + let mut unanswered: BTreeSet = BTreeSet::new(); for ((hash, id), result) in zip(txs_to_finalize, results) { match result { Ok(Some(receipt)) => { @@ -487,6 +545,7 @@ pub(crate) async fn fetch_finalized_receipts( DEBUG, "Received transaction receipt {receipt:?} for transaction {hash} and id {id:?}" ); + outcome.record_receipt(); match receipts.get(&id) { // by construction we never query twice the same transaction hash, which is a field in TransactionReceipt. Some(existing_receipt) => { @@ -494,7 +553,7 @@ pub(crate) async fn fetch_finalized_receipts( INFO, "ERROR: received different receipts for transaction {hash} with id {id:?}: {existing_receipt:?} and {receipt:?}. Will retry later" ); - return None; + outcome.abandon(); } None => { receipts.insert(id, receipt); @@ -502,32 +561,49 @@ pub(crate) async fn fetch_finalized_receipts( } } Ok(None) => { + outcome.record_not_mined(); log!( DEBUG, "Transaction {hash} for id {id:?} was not mined, it's probably a resubmitted transaction", ) } Err(e) => { + outcome.record_failure(); + unanswered.insert(id); log!( INFO, "Failed to get transaction receipt for {hash} and id {id:?}: {e:?}. Will retry later", ); - return None; } } } - let actual_finalized_ids: BTreeSet = receipts.keys().copied().collect(); - assert_eq!( - expected_finalized_ids, actual_finalized_ids, - "ERROR: unexpected transaction receipts for some ids" - ); - Some(receipts) + // The ids of an abandoned round were answered and thrown away, not left unanswered. + if outcome.is_abandoned() { + return (BTreeMap::new(), outcome); + } + // A selected id's nonce is below the finalized transaction count, so one of its transactions + // must have a receipt: none having one means the chain, the providers or our own bookkeeping is + // wrong. Counted and left pending rather than trapped, which would take the whole minter down. + // An id a provider failed to answer is not one of these, and is already counted as a failure. + for id in expected_finalized_ids + .iter() + .filter(|id| !receipts.contains_key(id) && !unanswered.contains(id)) + { + outcome.record_stalled_id(); + log!( + INFO, + "No transaction receipt for any of the transactions of id {id:?}: leaving it pending", + ); + } + (receipts, outcome) } -pub(crate) async fn finalized_transaction_count( +pub(crate) async fn finalized_transaction_count( sender: Address, + runtime: &R, ) -> Result> { - read_state(rpc_client) + runtime + .evm_rpc_client() .get_transaction_count((sender.into_bytes(), BlockTag::Finalized)) .with_cycles(MIN_ATTACHED_CYCLES) .try_send() diff --git a/rs/ethereum/cketh/minter/src/withdraw/tests.rs b/rs/ethereum/cketh/minter/src/withdraw/tests.rs new file mode 100644 index 000000000000..c325d1b2c6b6 --- /dev/null +++ b/rs/ethereum/cketh/minter/src/withdraw/tests.rs @@ -0,0 +1,255 @@ +use crate::eth_rpc::Hash; +use crate::eth_rpc_client::MultiCallError; +use crate::numeric::LedgerBurnIndex; +use crate::state::receipt_fetch::{ + INITIAL_RECEIPT_FETCH_WINDOW, ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING, RoundOutcome, +}; +use crate::state::{mutate_state, read_state}; +use crate::test_fixtures::{init_state, initial_state, mock, stub_rpc_client}; +use crate::withdraw::{ReceiptResult, collect_finalized_receipts, fetch_receipts_for_round}; +use evm_rpc_types::{ + Hex20, Hex32, Hex256, HexByte, Nat256, TransactionReceipt as EvmTransactionReceipt, +}; +use ic_canister_runtime::IcError; +use ic_ethereum_types::Address; +use std::collections::BTreeMap; + +mod collect { + use super::*; + + #[test] + fn should_return_nothing_for_an_empty_round() { + let (receipts, outcome) = collect_in_hash_order(vec![]); + + assert_eq!(receipts, BTreeMap::new()); + assert_eq!(outcome.lookups(), 0); + assert_eq!(outcome.stalled_ids(), 0); + } + + #[test] + fn should_finalize_the_ids_that_answered_and_leave_the_others_pending() { + let (receipts, outcome) = collect_in_hash_order(vec![ + (hash(1), id(1), Ok(Some(receipt(hash(1))))), + (hash(2), id(2), Err(failed_lookup())), + (hash(3), id(3), Ok(Some(receipt(hash(3))))), + ]); + + assert_eq!( + receipts, + BTreeMap::from([(id(1), receipt(hash(1))), (id(3), receipt(hash(3)))]) + ); + assert_eq!(outcome.receipts(), 2); + assert_eq!(outcome.failures(), 1); + assert_eq!( + outcome.stalled_ids(), + 0, + "an id a provider failed to answer is unanswered, not stalled" + ); + assert!(!outcome.is_abandoned()); + } + + #[test] + fn should_finalize_a_resubmitted_id_on_the_variant_that_was_mined() { + let (receipts, outcome) = collect_in_hash_order(vec![ + (hash(1), id(1), Ok(None)), + (hash(2), id(1), Ok(Some(receipt(hash(2))))), + (hash(3), id(1), Err(failed_lookup())), + ]); + + assert_eq!(receipts, BTreeMap::from([(id(1), receipt(hash(2)))])); + assert_eq!(outcome.receipts(), 1); + assert_eq!(outcome.not_mined(), 1); + assert_eq!(outcome.failures(), 1); + assert_eq!(outcome.stalled_ids(), 0); + } + + #[test] + fn should_leave_an_id_pending_without_trapping_when_none_of_its_transactions_was_mined() { + let (receipts, outcome) = collect_in_hash_order(vec![ + (hash(1), id(1), Ok(None)), + (hash(2), id(1), Ok(None)), + (hash(3), id(2), Ok(Some(receipt(hash(3))))), + ]); + + assert_eq!(receipts, BTreeMap::from([(id(2), receipt(hash(3)))])); + assert_eq!(outcome.not_mined(), 2); + assert_eq!(outcome.failures(), 0); + assert_eq!(outcome.stalled_ids(), 1); + } + + #[test] + fn should_tell_a_stalled_id_apart_from_one_a_provider_failed_to_answer() { + let (receipts, outcome) = collect_in_hash_order(vec![ + (hash(1), id(1), Ok(None)), + (hash(2), id(1), Ok(None)), + (hash(3), id(2), Err(failed_lookup())), + ]); + + assert_eq!(receipts, BTreeMap::new()); + assert_eq!(outcome.not_mined(), 2); + assert_eq!(outcome.failures(), 1); + assert_eq!( + outcome.stalled_ids(), + 1, + "only the id every provider answered counts as stalled" + ); + } + + #[test] + fn should_leave_every_id_pending_when_every_lookup_failed() { + let (receipts, outcome) = collect_in_hash_order(vec![ + (hash(1), id(1), Err(failed_lookup())), + (hash(2), id(2), Err(failed_lookup())), + ]); + + assert_eq!(receipts, BTreeMap::new()); + assert_eq!(outcome.failures(), 2); + assert_eq!(outcome.failures(), outcome.lookups()); + assert_eq!(outcome.stalled_ids(), 0); + } + + #[test] + fn should_abandon_the_round_but_count_every_lookup_on_two_receipts_for_the_same_id() { + let (receipts, outcome) = collect_in_hash_order(vec![ + (hash(1), id(1), Ok(Some(receipt(hash(1))))), + (hash(2), id(1), Ok(Some(receipt(hash(2))))), + (hash(3), id(2), Ok(Some(receipt(hash(3))))), + (hash(4), id(3), Ok(None)), + (hash(5), id(4), Err(failed_lookup())), + ]); + + assert_eq!(receipts, BTreeMap::new()); + assert!(outcome.is_abandoned()); + assert_eq!(outcome.receipts(), 3); + assert_eq!(outcome.not_mined(), 1); + assert_eq!(outcome.failures(), 1); + assert_eq!(outcome.lookups(), 5); + assert_eq!( + outcome.stalled_ids(), + 0, + "the ids of an abandoned round were answered and thrown away, not left unanswered" + ); + } +} + +mod round { + use super::*; + + #[tokio::test] + async fn should_skip_a_round_rather_than_read_the_chain_again() { + init_state(initial_state()); + mutate_state(|s| { + for _ in 0..=ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING { + s.withdrawal_transactions + .pipeline_mut() + .record_round_without_chain_read(); + } + }); + + let receipts = fetch_receipts_for_round(Address::new([0_u8; 20]), &no_rpc_runtime(), |s| { + s.withdrawal_transactions.pipeline_mut() + }) + .await; + + assert_eq!(receipts, BTreeMap::new()); + assert_eq!( + read_state(|s| s + .withdrawal_transactions + .pipeline() + .rounds_since_chain_read()), + ROUNDS_SINCE_CHAIN_READ_BEFORE_SKIPPING + 2, + "a skipped round is one more round that read nothing" + ); + } + + #[tokio::test] + async fn should_count_a_round_whose_chain_read_failed() { + init_state(initial_state()); + let mut runtime = mock::MockCanisterRuntime::new(); + runtime + .expect_evm_rpc_client() + .times(1) + .return_once(|| stub_rpc_client(vec![Err(IcError::CallPerformFailed)])); + + let receipts: BTreeMap = + fetch_receipts_for_round(Address::new([0_u8; 20]), &runtime, |s| { + s.withdrawal_transactions.pipeline_mut() + }) + .await; + + assert_eq!(receipts, BTreeMap::new()); + assert_eq!( + read_state(|s| s + .withdrawal_transactions + .pipeline() + .rounds_since_chain_read()), + 1, + "a round that could not read the chain is what starts the skipping" + ); + assert_eq!( + read_state(|s| s.withdrawal_transactions.pipeline().receipt_fetch_window()), + INITIAL_RECEIPT_FETCH_WINDOW, + "a round that never reached its lookups says nothing about the providers" + ); + } +} + +fn no_rpc_runtime() -> mock::MockCanisterRuntime { + let mut runtime = mock::MockCanisterRuntime::new(); + runtime.expect_evm_rpc_client().never(); + runtime +} + +fn collect_in_hash_order( + lookups: Vec<(Hash, LedgerBurnIndex, ReceiptResult)>, +) -> ( + BTreeMap, + RoundOutcome, +) { + let txs_to_finalize: BTreeMap = lookups + .iter() + .map(|(hash, id, _result)| (*hash, *id)) + .collect(); + assert_eq!(txs_to_finalize.len(), lookups.len(), "BUG: duplicate hash"); + let mut by_hash: BTreeMap = lookups + .into_iter() + .map(|(hash, _id, result)| (hash, result)) + .collect(); + let results = txs_to_finalize + .keys() + .map(|hash| by_hash.remove(hash).unwrap()) + .collect(); + collect_finalized_receipts(txs_to_finalize, results) +} + +fn failed_lookup() -> MultiCallError> { + MultiCallError::from_client_error(IcError::CallPerformFailed) +} + +fn id(id: u8) -> LedgerBurnIndex { + LedgerBurnIndex::new(id as u64) +} + +fn hash(seed: u8) -> Hash { + Hash([seed; 32]) +} + +fn receipt(transaction_hash: Hash) -> EvmTransactionReceipt { + EvmTransactionReceipt { + block_hash: Hex32::from([0x11_u8; 32]), + block_number: Nat256::from(0x4132ec_u64), + effective_gas_price: Nat256::from(0xfefbee3e_u64), + gas_used: Nat256::from(0x5208_u64), + cumulative_gas_used: Nat256::from(0x8b2e10_u64), + status: Some(Nat256::from(1_u8)), + root: None, + transaction_hash: Hex32::from(transaction_hash.0), + contract_address: None, + from: Hex20::from([0x17_u8; 20]), + logs: vec![], + logs_bloom: Hex256::from([0_u8; 256]), + to: Some(Hex20::from([0xdd_u8; 20])), + transaction_index: Nat256::from(0x32_u8), + tx_type: HexByte::from(0x02_u8), + } +} diff --git a/rs/ethereum/cketh/minter/tests/cketh.rs b/rs/ethereum/cketh/minter/tests/cketh.rs index 2c7e562caf0c..c047169647a7 100644 --- a/rs/ethereum/cketh/minter/tests/cketh.rs +++ b/rs/ethereum/cketh/minter/tests/cketh.rs @@ -1396,6 +1396,54 @@ fn should_export_the_sweep_pipeline_metrics() { .assert_contains_metric_matching(r"cketh_minter_last_balance_scan_age_seconds 90 \d+"); } +#[test] +fn should_export_the_receipt_fetch_metrics() { + CkEthSetup::default() + .check_minter_metrics() + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_window\{pipeline="withdrawal"\} 10 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_window\{pipeline="sweeper"\} 10 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_rounds_since_chain_read\{pipeline="withdrawal"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_rounds_since_chain_read\{pipeline="sweeper"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_lookups_total\{pipeline="withdrawal",outcome="receipt"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_lookups_total\{pipeline="withdrawal",outcome="not_mined"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_lookups_total\{pipeline="withdrawal",outcome="error"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_lookups_total\{pipeline="sweeper",outcome="receipt"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_lookups_total\{pipeline="sweeper",outcome="not_mined"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_lookups_total\{pipeline="sweeper",outcome="error"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_abandoned_rounds_total\{pipeline="withdrawal"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_abandoned_rounds_total\{pipeline="sweeper"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_stalled_ids_total\{pipeline="withdrawal"\} 0 \d+"#, + ) + .assert_contains_metric_matching( + r#"cketh_minter_receipt_fetch_stalled_ids_total\{pipeline="sweeper"\} 0 \d+"#, + ); +} + #[test] fn should_export_the_stored_attestation_and_authorization_metrics() { CkEthSetup::default()