dfinity · mbjorkqvist · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026
@@ -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

@@ -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;

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

@@ -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<Id> {
+    window: usize,
+    cursor: Option<Id>,
+    rounds_since_chain_read: u32,
+    receipts_total: u64,
+    not_mined_total: u64,
+    failures_total: u64,
+    stalled_ids_total: u64,
+    abandoned_rounds_total: u64,
+}
+
+impl<Id> Default for ReceiptFetchWindow<Id> {
+    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<Id: Copy + Ord> ReceiptFetchWindow<Id> {
+    /// 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<Hash, Id>) -> BTreeMap<Hash, Id> {
+        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<Id, Vec<Hash>>) -> Vec<Id> {
+        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<Id> {
+        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<Id: Copy + Ord>(pending: &BTreeMap<Hash, Id>) -> BTreeMap<Id, Vec<Hash>> {
+    let mut by_id: BTreeMap<Id, Vec<Hash>> = BTreeMap::new();
+    for (hash, id) in pending {
+        by_id.entry(*id).or_default().push(*hash);
+    }
+    by_id
+}