implement better retries for tip-distributor workflows

tip-distributor/src/claim_mev_workflow.rs

@@ -1,5 +1,7 @@
 use {
-    crate::{read_json_from_file, send_transactions_with_retry, GeneratedMerkleTreeCollection},
+    crate::{
+        read_json_from_file, sign_and_send_transactions_with_retries, GeneratedMerkleTreeCollection,
+    },
     anchor_lang::{AccountDeserialize, InstructionData, ToAccountMetas},
     log::{debug, info, warn},
     solana_client::{nonblocking::rpc_client::RpcClient, rpc_request::RpcError},
@@ -37,7 +39,7 @@ pub fn claim_mev_tips(
     keypair_path: &PathBuf,
 ) -> Result<(), ClaimMevError> {
     // roughly how long before blockhash expires
-    const MAX_RETRY_DURATION: Duration = Duration::from_secs(60);
+    const MAX_RETRY_DURATION: Duration = Duration::from_secs(300);
 
     let merkle_trees: GeneratedMerkleTreeCollection =
         read_json_from_file(merkle_root_path).expect("read GeneratedMerkleTreeCollection");
@@ -131,21 +133,19 @@ pub fn claim_mev_tips(
             }
         }
 
-        let blockhash = rpc_client.get_latest_blockhash().await.expect("fetch latest blockhash");
         let transactions = instructions.into_iter().map(|ix|{
-            Transaction::new_signed_with_payer(
+            Transaction::new_with_payer(
                 &[ix],
                 Some(&keypair.pubkey()),
-                &[&keypair],
-                blockhash,
             )
         }).collect::<Vec<_>>();
 
         info!("Sending {} tip claim transactions. {} tried sending zero lamports, {} would be below minimum rent",
             &transactions.len(), zero_lamports_count, below_min_rent_count);
-        let num_failed_txs = send_transactions_with_retry(&rpc_client, &transactions, MAX_RETRY_DURATION).await;
-        if num_failed_txs != 0 {
-            panic!("failed to send {num_failed_txs} transactions");
+
+        let failed_transactions = sign_and_send_transactions_with_retries(&keypair, &rpc_client, transactions, MAX_RETRY_DURATION).await;
+        if !failed_transactions.is_empty() {
+            panic!("failed to send {} transactions", failed_transactions.len());
         }
     });
 

tip-distributor/src/lib.rs

@@ -15,12 +15,13 @@ use {
     solana_client::{nonblocking::rpc_client::RpcClient, rpc_client::RpcClient as SyncRpcClient},
     solana_merkle_tree::MerkleTree,
     solana_metrics::{datapoint_error, datapoint_warn},
+    solana_rpc_client_api::client_error::Error,
     solana_sdk::{
         account::{AccountSharedData, ReadableAccount},
         clock::Slot,
         hash::{Hash, Hasher},
         pubkey::Pubkey,
-        signature::Signature,
+        signature::{Keypair, Signature},
         stake_history::Epoch,
         transaction::Transaction,
     },
@@ -40,7 +41,6 @@ use {
         TIP_ACCOUNT_SEED_3, TIP_ACCOUNT_SEED_4, TIP_ACCOUNT_SEED_5, TIP_ACCOUNT_SEED_6,
         TIP_ACCOUNT_SEED_7,
     },
-    tokio::time::sleep,
 };
 
 #[derive(Deserialize, Serialize, Debug)]
@@ -455,59 +455,59 @@ pub fn derive_tip_distribution_account_address(
     )
 }
 
-pub async fn send_transactions_with_retry(
+pub async fn sign_and_send_transactions_with_retries(
+    signer: &Keypair,
     rpc_client: &RpcClient,
-    transactions: &[Transaction],
-    max_send_duration: Duration,
-) -> usize {
-    let mut transactions_to_send: HashMap<Signature, Transaction> = transactions
-        .iter()
-        .map(|tx| (tx.signatures[0], tx.clone()))
-        .collect();
+    mut transactions: Vec<Transaction>,
+    max_retry_duration: Duration,
+) -> HashMap<Signature, Error> {
+    let mut errors = HashMap::default();
+    let mut blockhash = rpc_client
+        .get_latest_blockhash()
+        .await
+        .expect("fetch latest blockhash");
 
     let start = Instant::now();
-    while !transactions_to_send.is_empty() && start.elapsed() < max_send_duration {
-        info!("sending {} transactions", transactions_to_send.len());
-
-        for (signature, tx) in &transactions_to_send {
-            match rpc_client.send_transaction(tx).await {
-                Ok(send_tx_sig) => {
-                    info!("send transaction: {:?}", send_tx_sig);
-                }
-                Err(e) => {
-                    error!("error sending transaction {:?} error: {:?}", signature, e);
-                }
-            }
+    while start.elapsed() < max_retry_duration {
+        if start.elapsed() > Duration::from_secs(60) {
+            blockhash = rpc_client
+                .get_latest_blockhash()
+                .await
+                .expect("fetch latest blockhash");
         }
 
-        sleep(Duration::from_secs(10)).await;
+        transactions
+            .iter_mut()
+            .for_each(|tx| tx.sign(&[signer], blockhash));
+        let mut signatures_to_transactions = transactions
+            .clone()
+            .into_iter()
+            .map(|tx| (tx.signatures[0], tx))
+            .collect::<HashMap<Signature, Transaction>>();
 
-        let mut signatures_confirmed: Vec<Signature> = Vec::new();
-        for signature in transactions_to_send.keys() {
-            match rpc_client.confirm_transaction(signature).await {
-                Ok(true) => {
-                    info!("confirmed signature: {:?}", signature);
-                    signatures_confirmed.push(*signature);
-                }
-                Ok(false) => {
-                    info!("didn't confirmed signature: {:?}", signature);
-                }
-                Err(e) => {
-                    error!(
-                        "error confirming signature: {:?}, signature: {:?}",
-                        signature, e
-                    );
-                }
-            }
-        }
+        let futs = transactions.iter().map(|tx| async {
+            let result = rpc_client.send_and_confirm_transaction(tx).await;
+            (tx.signatures[0], result)
+        });
 
-        info!("confirmed {} signatures", signatures_confirmed.len());
-        for sig in signatures_confirmed {
-            transactions_to_send.remove(&sig);
-        }
+        errors = futures::future::join_all(futs)
+            .await
+            .into_iter()
+            .filter(|(_sig, result)| result.is_err())
+            .map(|(sig, result)| {
+                let e = result.err().unwrap();
+                warn!("error sending transaction: [error={e}, signature={sig}]");
+                (sig, e)
+            })
+            .collect::<HashMap<_, _>>();
+
+        transactions = errors
+            .iter()
+            .map(|(sig, _)| signatures_to_transactions.remove(sig).unwrap())
+            .collect::<Vec<_>>();
     }
 
-    transactions_to_send.len()
+    errors
 }
 
 mod pubkey_string_conversion {

tip-distributor/src/merkle_root_upload_workflow.rs

@@ -1,6 +1,6 @@
 use {
     crate::{
-        read_json_from_file, send_transactions_with_retry, GeneratedMerkleTree,
+        read_json_from_file, sign_and_send_transactions_with_retries, GeneratedMerkleTree,
         GeneratedMerkleTreeCollection,
     },
     anchor_lang::AccountDeserialize,
@@ -40,7 +40,7 @@ pub fn upload_merkle_root(
     tip_distribution_program_id: &Pubkey,
 ) -> Result<(), MerkleRootUploadError> {
     // max amount of time before blockhash expires
-    const MAX_RETRY_DURATION: Duration = Duration::from_secs(60);
+    const MAX_RETRY_DURATION: Duration = Duration::from_secs(300);
 
     let merkle_tree: GeneratedMerkleTreeCollection =
         read_json_from_file(merkle_root_path).expect("read GeneratedMerkleTreeCollection");
@@ -58,11 +58,6 @@ pub fn upload_merkle_root(
     runtime.block_on(async move {
         let rpc_client =
             RpcClient::new_with_commitment(rpc_url.to_string(), CommitmentConfig::confirmed());
-        let recent_blockhash = rpc_client
-            .get_latest_blockhash()
-            .await
-            .expect("get blockhash");
-
         let trees: Vec<GeneratedMerkleTree> = merkle_tree
             .generated_merkle_trees
             .into_iter()
@@ -124,17 +119,15 @@ pub fn upload_merkle_root(
                         tip_distribution_account: tree.tip_distribution_account,
                     },
                 );
-                Transaction::new_signed_with_payer(
+                Transaction::new_with_payer(
                     &[ix],
                     Some(&keypair.pubkey()),
-                    &[&keypair],
-                    recent_blockhash,
                 )
             })
             .collect();
-        let num_failed_txs = send_transactions_with_retry(&rpc_client, &transactions, MAX_RETRY_DURATION).await;
-        if num_failed_txs != 0 {
-            panic!("failed to send {num_failed_txs} transactions");
+        let failed_transactions = sign_and_send_transactions_with_retries(&keypair, &rpc_client, transactions, MAX_RETRY_DURATION).await;
+        if !failed_transactions.is_empty() {
+            panic!("failed to send {} transactions", failed_transactions.len());
         }
     });
 

tip-distributor/src/reclaim_rent_workflow.rs

@@ -1,5 +1,5 @@
 use {
-    crate::send_transactions_with_retry,
+    crate::sign_and_send_transactions_with_retries,
     anchor_lang::AccountDeserialize,
     log::info,
     solana_client::nonblocking::rpc_client::RpcClient,
@@ -145,29 +145,23 @@ pub async fn reclaim_rent(
             )
             .collect::<Vec<_>>()
             .chunks(4)
-            .map(|instructions| {
-                Transaction::new_signed_with_payer(
-                    instructions,
-                    Some(&signer.pubkey()),
-                    &[&signer],
-                    recent_blockhash,
-                )
-            })
+            .map(|instructions| Transaction::new_with_payer(instructions, Some(&signer.pubkey())))
             .collect::<Vec<_>>();
         info!("create close_tip_distribution_account transactions took {}us, closing {} tip distribution accounts", now.elapsed().as_micros(), close_tda_txs.len());
 
         transactions.extend(close_tda_txs);
     }
 
     info!("sending {} transactions", transactions.len());
-    let num_failed_txs = send_transactions_with_retry(
+    let failed_txs = sign_and_send_transactions_with_retries(
+        &signer,
         &rpc_client,
-        transactions.as_slice(),
-        Duration::from_secs(60),
+        transactions,
+        Duration::from_secs(300),
     )
     .await;
-    if num_failed_txs != 0 {
-        panic!("failed to send {num_failed_txs} transactions");
+    if !failed_txs.is_empty() {
+        panic!("failed to send {} transactions", failed_txs.len());
     }
 
     Ok(())