feat(minter): finalize withdrawals from their nonce accounts

integration_tests/src/fixtures.rs

@@ -48,6 +48,11 @@ const MOCK_SLOT: u64 = 100_000_000;
 const SUBMITTED_BLOCKHASH: &str = "4sGjMW1sUnHzSxGspuhpqLDx6wiyjNtZAMdL4VZHirAn";
 const SUBMITTED_SIGNATURE: &str =
     "5VERv8NMvzbJMEkV8xnrLkEaWRtSz9CosKDYjCJjBRnbJLgp8uirBgmQpjKhoR4tjF3ZpRzrFmBV6UjKdiSZkQUW";
+/// Seed of the nonce value the mocked nonce account stores when the withdrawal timer reads it.
+const SUBMITTED_NONCE_SEED: [u8; 32] = [0x4E; 32];
+/// Seed of the nonce value the mocked nonce account stores once the submitted withdrawal
+/// transaction has advanced it.
+const ADVANCED_NONCE_SEED: [u8; 32] = [0x4F; 32];
 /// Blockhash the mocks report for a block a timer only reads the height of.
 const IGNORED_BLOCKHASH: &str = "CzBVNFJkh7WkQDfJUiDjLc7kPrJd8kR2yiCvwBUhSe7Y";
 
@@ -154,7 +159,7 @@ impl MockBuilder {
     pub fn get_nonce_account(self) -> Self {
         self.expect(
             get_account_info_request(),
-            get_account_info_nonce_response(),
+            get_account_info_nonce_response(SUBMITTED_NONCE_SEED),
         )
     }
 
@@ -167,11 +172,19 @@ impl MockBuilder {
         )
     }
 
-    /// Mocks for `finalize_transactions` finding only in-flight withdrawal
-    /// transactions, which carry a durable nonce and need no current block:
-    /// a single `getSignatureStatuses` reporting them as finalized.
-    pub fn finalize_withdrawal_transaction(self, signature: &Signature) -> Self {
-        self.check_signature_statuses(signature, get_signature_statuses_finalized_response())
+    /// Mocks for `finalize_transactions` finding only the given in-flight withdrawal
+    /// transaction, which carries a durable nonce and needs no current block:
+    /// `getAccountInfo` reporting its nonce account advanced → `getTransaction`
+    /// reporting it as succeeded.
+    pub fn finalize_withdrawal_transaction(self, withdrawal: &Transaction) -> Self {
+        self.expect(
+            get_account_info_request(),
+            get_account_info_nonce_response(ADVANCED_NONCE_SEED),
+        )
+        .expect(
+            get_transaction_request(&withdrawal.signatures[0]),
+            withdrawal_transaction_response(withdrawal),
+        )
     }
 
     /// Mocks for `finalize_transactions` finding the pending transaction with the given
@@ -263,17 +276,49 @@ fn sweep_transaction_response(sweep: &Transaction, sweepable_amount: Lamport) ->
     }))
 }
 
+/// JSON-RPC `getTransaction` response for the given withdrawal transaction, finalized without
+/// error.
+fn withdrawal_transaction_response(withdrawal: &Transaction) -> JsonRpcResponse {
+    let encoded_transaction = STANDARD.encode(
+        bincode::serialize(withdrawal).expect("serializing the transaction should succeed"),
+    );
+    JsonRpcResponse::from(json!({
+        "jsonrpc": "2.0",
+        "result": {
+            "blockTime": 1700000000_i64,
+            "meta": {
+                "computeUnitsConsumed": 450,
+                "err": null,
+                "fee": FEE_PER_SIGNATURE,
+                "innerInstructions": [],
+                "loadedAddresses": { "readonly": [], "writable": [] },
+                "logMessages": [],
+                "postBalances": [],
+                "postTokenBalances": [],
+                "preBalances": [],
+                "preTokenBalances": [],
+                "rewards": [],
+                "status": { "Ok": null }
+            },
+            "slot": 350_000_000_u64,
+            "transaction": [encoded_transaction, "base64"],
+            "version": "legacy"
+        },
+        "id": 1
+    }))
+}
+
 fn get_account_info_request() -> JsonRpcRequestMatcher {
     JsonRpcRequestMatcher::with_method("getAccountInfo")
 }
 
-fn get_account_info_nonce_response() -> JsonRpcResponse {
+fn get_account_info_nonce_response(nonce_seed: [u8; 32]) -> JsonRpcResponse {
     JsonRpcResponse::from(json!({
         "jsonrpc": "2.0",
         "result": {
             "context": { "apiVersion": "2.0.15", "slot": 341_197_053 },
             "value": {
-                "data": [nonce_account_data(), "base64"],
+                "data": [nonce_account_data(nonce_seed), "base64"],
                 "executable": false,
                 "lamports": 1_447_680,
                 "owner": "11111111111111111111111111111111",
@@ -285,10 +330,10 @@ fn get_account_info_nonce_response() -> JsonRpcResponse {
     }))
 }
 
-fn nonce_account_data() -> String {
+fn nonce_account_data(nonce_seed: [u8; 32]) -> String {
     let nonce_account = Versions::new(State::Initialized(Data::new(
         MINTER_ADDRESS,
-        DurableNonce::from_blockhash(&Hash::from([0x4E; 32])),
+        DurableNonce::from_blockhash(&Hash::from(nonce_seed)),
         FEE_PER_SIGNATURE,
     )));
     STANDARD.encode(

integration_tests/tests/tests.rs

@@ -795,12 +795,14 @@ mod withdrawal_tests {
         );
 
         // Only the in-flight withdrawal is monitored, so the finalization timer
-        // checks the signature statuses without fetching a current block.
+        // reads its nonce account and fetches the landed transaction without
+        // fetching a current block.
         setup.advance_time(FINALIZE_TRANSACTIONS_DELAY).await;
+        let withdrawal = setup.minter().signed_transaction(&transaction_id).await;
         setup
             .execute_http_mocks(
                 MockBuilder::with_start_id(40)
-                    .finalize_withdrawal_transaction(&transaction_id)
+                    .finalize_withdrawal_transaction(&withdrawal)
                     .build(),
             )
             .await;

minter/src/metrics.rs

@@ -1,5 +1,5 @@
 use crate::state::{QuarantineCause, State};
-use crate::storage::{self, FailedCreditReason, FailedMintReason};
+use crate::storage::{self, FailedCreditReason, FailedMintReason, UndecidedWithdrawalReason};
 use ic_metrics_encoder::MetricsEncoder;
 
 const WASM_PAGE_SIZE_IN_BYTES: usize = 65536;
@@ -147,6 +147,16 @@ pub fn encode_metrics(w: &mut MetricsEncoder<Vec<u8>>, s: &State) -> std::io::Re
         storage::with_unstable_metrics(|m| m.withdrawal_transaction_rebroadcasts).metric_value(),
         "Number of re-broadcast attempts of withdrawal transactions since the last upgrade.",
     )?;
+    let mut undecided_withdrawal_transactions = w.gauge_vec(
+        "undecided_withdrawal_transactions",
+        "Number of in-flight withdrawal transactions left undecided by the last finalization round, by reason: a stale nonce read, a failed nonce read, or a nonce advance whose outcome could not be fetched.",
+    )?;
+    for reason in UndecidedWithdrawalReason::ALL {
+        undecided_withdrawal_transactions = undecided_withdrawal_transactions.value(
+            &[("reason", reason.label())],
+            storage::undecided_withdrawal_transaction_count(reason).metric_value(),
+        )?;
+    }
     Ok(())
 }
 

minter/src/monitor/mod.rs

@@ -2,36 +2,35 @@ use crate::{
     constants::MAX_CONCURRENT_RPC_CALLS,
     deposit::sweep::credit_finalized_sweeps,
     guard::TimerGuard,
-    rpc::{
-        BlockHeight, get_recent_block, get_signature_statuses,
-        submit_transaction_skipping_preflight,
+    monitor::withdrawals::{
+        finalize_transactions_batch, read_in_flight_nonces, resubmit_transactions_batch,
     },
+    rpc::{BlockHeight, get_recent_block, get_signature_statuses},
     runtime::CanisterRuntime,
     state::{
-        MinterTransaction, TaskType,
-        audit::process_event,
-        event::{EventType, VersionedMessage},
-        mutate_state, read_state,
+        MinterTransaction, TaskType, audit::process_event, event::EventType, mutate_state,
+        read_state,
     },
-    storage::with_unstable_metrics_mut,
 };
 use canlog::log;
 use cksol_types_internal::log::Priority;
 use itertools::Itertools;
 use solana_signature::Signature;
-use solana_transaction::Transaction;
 use solana_transaction_status_client_types::TransactionConfirmationStatus;
 use std::collections::{BTreeMap, BTreeSet};
 use std::time::Duration;
 
 #[cfg(test)]
 mod tests;
+mod withdrawals;
 
 pub const FINALIZE_TRANSACTIONS_DELAY: Duration = Duration::from_mins(2);
-/// Minimum time since its submission before a withdrawal transaction without a
-/// status is re-broadcast. The minter sends transactions without `maxRetries`,
-/// so an Agave RPC node keeps re-sending a durable-nonce transaction every 2 s
-/// until it lands, its nonce advances, or 150 blocks (about 60 to 90 s) pass.
+pub const FINALIZE_TRANSACTIONS_RETRY_DELAY: Duration = Duration::from_secs(10);
+/// Minimum time since its submission before a withdrawal transaction whose nonce
+/// account still stores its nonce value is re-broadcast. The minter sends
+/// transactions without `maxRetries`, so an Agave RPC node keeps re-sending a
+/// durable-nonce transaction every 2 s until it lands, its nonce advances, or
+/// 150 blocks (about 60 to 90 s) pass.
 /// Re-broadcasting earlier would only duplicate the work of the RPC node.
 /// See https://github.com/anza-xyz/agave/blob/master/rpc/src/rpc.rs
 pub const MIN_REBROADCAST_AGE: Duration = Duration::from_secs(90);
@@ -48,19 +47,23 @@ const MAX_BLOCKHASH_AGE_IN_BLOCKS: BlockHeight = BlockHeight::new(150);
 /// See https://solana.com/docs/rpc/http/getsignaturestatuses
 const MAX_SIGNATURES_PER_STATUS_CHECK: usize = 256;
 
-/// Check the status of all submitted transactions, finalize succeeded/failed
-/// ones, drop expired sweeps, and re-broadcast withdrawals that have no status.
+/// Check the status of all submitted sweeps, finalize succeeded/failed ones and drop
+/// expired ones, then finalize the withdrawals whose nonce advanced and re-broadcast
+/// the withdrawals whose nonce did not.
 pub async fn finalize_transactions<R: CanisterRuntime>(runtime: R) {
     let _guard = match TimerGuard::new(TaskType::FinalizeTransactions) {
         Ok(guard) => guard,
         Err(_) => return,
     };
 
     let reschedule = scopeguard::guard(runtime.clone(), |runtime| {
-        runtime.set_timer(Duration::ZERO, finalize_transactions);
+        runtime.set_timer(FINALIZE_TRANSACTIONS_RETRY_DELAY, finalize_transactions);
     });
 
-    let more_transactions_to_check = check_submitted_transactions(&runtime).await;
+    let more_transactions_to_check = check_sweep_transactions(&runtime).await;
+    let nonces = read_in_flight_nonces(&runtime).await;
+    resubmit_transactions_batch(&runtime, &nonces).await;
+    finalize_transactions_batch(&runtime, &nonces).await;
     let more_sweeps_to_credit = credit_finalized_sweeps(&runtime).await;
 
     if !more_transactions_to_check && !more_sweeps_to_credit {
@@ -69,40 +72,31 @@ pub async fn finalize_transactions<R: CanisterRuntime>(runtime: R) {
 }
 
 /// Returns whether the finalization timer must run again immediately.
-async fn check_submitted_transactions<R: CanisterRuntime>(runtime: &R) -> bool {
-    let (signatures, blockhash_transactions): (
-        BTreeSet<Signature>,
-        BTreeMap<Signature, BlockHeight>,
-    ) = read_state(|state| {
-        let submitted = state.submitted_transactions();
-        (
-            submitted.iter().map(|(sig, _)| *sig).collect(),
-            submitted
-                .iter()
-                .filter_map(|(sig, tx)| match tx {
-                    MinterTransaction::SweepDeposit { block_height, .. } => {
-                        Some((*sig, *block_height))
-                    }
-                    MinterTransaction::Withdrawal { .. } => None,
-                })
-                .collect(),
-        )
+async fn check_sweep_transactions<R: CanisterRuntime>(runtime: &R) -> bool {
+    let blockhash_transactions: BTreeMap<Signature, BlockHeight> = read_state(|state| {
+        state
+            .submitted_transactions()
+            .iter()
+            .filter_map(|(signature, transaction)| match transaction {
+                MinterTransaction::SweepDeposit { block_height } => {
+                    Some((*signature, *block_height))
+                }
+                MinterTransaction::Withdrawal { .. } => None,
+            })
+            .collect()
     });
-    if signatures.is_empty() {
+    if blockhash_transactions.is_empty() {
         return false;
     }
 
     // Fetch the current block before checking statuses: if a transaction finalizes
     // after we snapshot the block, the status check will see it as finalized rather
     // than missing, so it will never be incorrectly marked as expired.
-    let current_block_height = if blockhash_transactions.is_empty() {
-        None
-    } else {
-        fetch_current_block_height(runtime).await
-    };
+    let current_block_height = fetch_current_block_height(runtime).await;
 
-    let num_transactions = signatures.len();
-    let statuses = check_transaction_statuses(runtime, signatures.into_iter().collect()).await;
+    let num_transactions = blockhash_transactions.len();
+    let statuses =
+        check_transaction_statuses(runtime, blockhash_transactions.keys().copied().collect()).await;
 
     for (signature, error) in &statuses.errored {
         log!(
@@ -142,8 +136,6 @@ async fn check_submitted_transactions<R: CanisterRuntime>(runtime: &R) -> bool {
         );
     }
 
-    rebroadcast_withdrawal_transactions(runtime, &statuses.not_found).await;
-
     num_transactions > MAX_CONCURRENT_RPC_CALLS * MAX_SIGNATURES_PER_STATUS_CHECK
 }
 
@@ -186,62 +178,6 @@ fn expire_transactions<R: CanisterRuntime>(
     }
 }
 
-async fn rebroadcast_withdrawal_transactions<R: CanisterRuntime>(
-    runtime: &R,
-    not_found: &BTreeSet<Signature>,
-) {
-    let now = runtime.time();
-    let batches: Vec<Vec<Transaction>> = read_state(|state| {
-        not_found
-            .iter()
-            .filter_map(
-                |signature| match state.submitted_transactions().get(signature)? {
-                    MinterTransaction::Withdrawal {
-                        message: VersionedMessage::Legacy(message),
-                        submitted_at,
-                        ..
-                    } if is_old_enough_to_rebroadcast(*submitted_at, now) => Some(Transaction {
-                        signatures: vec![*signature],
-                        message: message.clone(),
-                    }),
-                    MinterTransaction::Withdrawal { .. }
-                    | MinterTransaction::SweepDeposit { .. } => None,
-                },
-            )
-            .chunks(MAX_CONCURRENT_RPC_CALLS)
-            .into_iter()
-            .map(Iterator::collect)
-            .collect()
-    });
-    for batch in batches {
-        futures::future::join_all(
-            batch
-                .into_iter()
-                .map(|transaction| rebroadcast_transaction(runtime, transaction)),
-        )
-        .await;
-    }
-}
-
-fn is_old_enough_to_rebroadcast(submitted_at: u64, now: u64) -> bool {
-    Duration::from_nanos(now.saturating_sub(submitted_at)) >= MIN_REBROADCAST_AGE
-}
-
-async fn rebroadcast_transaction<R: CanisterRuntime>(runtime: &R, transaction: Transaction) {
-    let signature = transaction.signatures[0];
-    with_unstable_metrics_mut(|m| m.withdrawal_transaction_rebroadcasts += 1);
-    match submit_transaction_skipping_preflight(runtime, transaction).await {
-        Ok(_) => log!(
-            Priority::Info,
-            "Re-broadcast withdrawal transaction {signature}"
-        ),
-        Err(e) => log!(
-            Priority::Info,
-            "Failed to re-broadcast withdrawal transaction {signature} (will retry next round): {e}"
-        ),
-    }
-}
-
 fn is_blockhash_expired(
     transaction_block_height: BlockHeight,
     current_block_height: BlockHeight,