Merge pull request #46 from pyth-network/drozdziak1/p2w-client-async-retries

p2w client retries

solana/pyth2wormhole/Cargo.lock

@@ -3141,11 +3141,10 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
 
 [[package]]
 name = "tokio"
-version = "1.7.1"
+version = "1.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5fb2ed024293bb19f7a5dc54fe83bf86532a44c12a2bb8ba40d64a4509395ca2"
+checksum = "0c27a64b625de6d309e8c57716ba93021dccf1b3b5c97edd6d3dd2d2135afc0a"
 dependencies = [
- "autocfg",
  "bytes",
  "libc",
  "memchr",
@@ -3161,9 +3160,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-macros"
-version = "1.2.0"
+version = "1.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c49e3df43841dafb86046472506755d8501c5615673955f6aa17181125d13c37"
+checksum = "b557f72f448c511a979e2564e55d74e6c4432fc96ff4f6241bc6bded342643b7"
 dependencies = [
  "proc-macro2 1.0.27",
  "quote 1.0.9",

solana/pyth2wormhole/client/Cargo.toml

@@ -5,6 +5,10 @@ edition = "2018"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+[lib]
+name = "pyth2wormhole_client"
+src = "src/lib.rs"
+
 [features]
 default = ["pyth2wormhole/client", "wormhole-bridge-solana/client"]
 
@@ -24,4 +28,4 @@ solana-program = "=1.9.4"
 solana-sdk = "=1.9.4"
 solana-transaction-status = "=1.9.4"
 solitaire-client = {path = "../../solitaire/client"}
-solitaire = {path = "../../solitaire/program"}
+solitaire = {path = "../../solitaire/program"}

solana/pyth2wormhole/client/src/attestation_cfg.rs

@@ -8,7 +8,6 @@ use serde::{
     Serializer,
 };
 use solana_program::pubkey::Pubkey;
-use solitaire::ErrBox;
 
 /// Pyth2wormhole config specific to attestation requests
 #[derive(Debug, Deserialize, Serialize, PartialEq, Eq)]
@@ -56,6 +55,8 @@ where
 mod tests {
     use super::*;
 
+    use solitaire::ErrBox;
+
     #[test]
     fn test_sanity() -> Result<(), ErrBox> {
         let initial = AttestationConfig {

solana/pyth2wormhole/client/src/cli.rs

@@ -49,10 +49,17 @@ pub enum Action {
     Attest {
         #[clap(short = 'f', long = "--config", about = "Attestation YAML config")]
         attestation_cfg: PathBuf,
+        #[clap(
+            short = 'n',
+            long = "--retries",
+            about = "How many times to retry each batch on failure",
+            default_value = "3"
+        )]
+        n_retries: usize,
     },
-    #[clap(
-        about = "Update an existing pyth2wormhole program's settings (currently set owner only)"
-    )]
+    #[clap(about = "Retrieve a pyth2wormhole program's current settings")]
+    GetConfig,
+    #[clap(about = "Update an existing pyth2wormhole program's settings")]
     SetConfig {
         /// Current owner keypair path
         #[clap(long = "owner", default_value = "~/.config/solana/id.json")]

solana/pyth2wormhole/client/src/lib.rs

@@ -0,0 +1,252 @@
+pub mod attestation_cfg;
+
+use borsh::{
+    BorshDeserialize,
+    BorshSerialize,
+};
+use solana_client::rpc_client::RpcClient;
+use solana_program::{
+    hash::Hash,
+    instruction::{
+        AccountMeta,
+        Instruction,
+    },
+    pubkey::Pubkey,
+    system_program,
+    sysvar::{
+        clock,
+        rent,
+    },
+};
+use solana_sdk::transaction::Transaction;
+use solitaire::{
+    processors::seeded::Seeded,
+    AccountState,
+    ErrBox,
+};
+use solitaire_client::{
+    AccEntry,
+    Keypair,
+    SolSigner,
+    ToInstruction,
+};
+
+use bridge::{
+    accounts::{
+        Bridge,
+        FeeCollector,
+        Sequence,
+        SequenceDerivationData,
+    },
+    types::ConsistencyLevel,
+};
+
+use p2w_sdk::P2WEmitter;
+
+use pyth2wormhole::{
+    attest::P2W_MAX_BATCH_SIZE,
+    config::P2WConfigAccount,
+    initialize::InitializeAccounts,
+    set_config::SetConfigAccounts,
+    AttestData,
+    Pyth2WormholeConfig,
+};
+
+pub use attestation_cfg::{
+    AttestationConfig,
+    P2WSymbol,
+};
+
+pub fn gen_init_tx(
+    payer: Keypair,
+    p2w_addr: Pubkey,
+    new_owner_addr: Pubkey,
+    wh_prog: Pubkey,
+    pyth_owner_addr: Pubkey,
+    latest_blockhash: Hash,
+) -> Result<Transaction, ErrBox> {
+    use AccEntry::*;
+
+    let payer_pubkey = payer.pubkey();
+
+    let accs = InitializeAccounts {
+        payer: Signer(payer),
+        new_config: Derived(p2w_addr),
+    };
+
+    let config = Pyth2WormholeConfig {
+        max_batch_size: P2W_MAX_BATCH_SIZE,
+        owner: new_owner_addr,
+        wh_prog: wh_prog,
+        pyth_owner: pyth_owner_addr,
+    };
+    let ix_data = (pyth2wormhole::instruction::Instruction::Initialize, config);
+
+    let (ix, signers) = accs.to_ix(p2w_addr, ix_data.try_to_vec()?.as_slice())?;
+
+    let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
+        &[ix],
+        Some(&payer_pubkey),
+        signers.iter().collect::<Vec<_>>().as_ref(),
+        latest_blockhash,
+    );
+    Ok(tx_signed)
+}
+
+pub fn gen_set_config_tx(
+    payer: Keypair,
+    p2w_addr: Pubkey,
+    owner: Keypair,
+    new_owner_addr: Pubkey,
+    new_wh_prog: Pubkey,
+    new_pyth_owner_addr: Pubkey,
+    latest_blockhash: Hash,
+) -> Result<Transaction, ErrBox> {
+    use AccEntry::*;
+
+    let payer_pubkey = payer.pubkey();
+
+    let accs = SetConfigAccounts {
+        payer: Signer(payer),
+        current_owner: Signer(owner),
+        config: Derived(p2w_addr),
+    };
+
+    let config = Pyth2WormholeConfig {
+        max_batch_size: P2W_MAX_BATCH_SIZE,
+        owner: new_owner_addr,
+        wh_prog: new_wh_prog,
+        pyth_owner: new_pyth_owner_addr,
+    };
+    let ix_data = (pyth2wormhole::instruction::Instruction::SetConfig, config);
+
+    let (ix, signers) = accs.to_ix(p2w_addr, ix_data.try_to_vec()?.as_slice())?;
+
+    let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
+        &[ix],
+        Some(&payer_pubkey),
+        signers.iter().collect::<Vec<_>>().as_ref(),
+        latest_blockhash,
+    );
+    Ok(tx_signed)
+}
+
+/// Get the current config account data for given p2w program address
+pub fn get_config_account(
+    rpc_client: &RpcClient,
+    p2w_addr: &Pubkey,
+) -> Result<Pyth2WormholeConfig, ErrBox> {
+    let p2w_config_addr = P2WConfigAccount::<{ AccountState::Initialized }>::key(None, p2w_addr);
+
+    let config = Pyth2WormholeConfig::try_from_slice(
+        rpc_client.get_account_data(&p2w_config_addr)?.as_slice(),
+    )?;
+
+    Ok(config)
+}
+
+/// Generate an Instruction for making the attest() contract
+/// call.
+pub fn gen_attest_tx(
+    p2w_addr: Pubkey,
+    p2w_config: &Pyth2WormholeConfig, // Must be fresh, not retrieved inside to keep side effects away
+    payer: &Keypair,
+    symbols: &[P2WSymbol],
+    wh_msg: &Keypair,
+    latest_blockhash: Hash,
+) -> Result<Transaction, ErrBox> {
+    let emitter_addr = P2WEmitter::key(None, &p2w_addr);
+
+    let seq_addr = Sequence::key(
+        &SequenceDerivationData {
+            emitter_key: &emitter_addr,
+        },
+        &p2w_config.wh_prog,
+    );
+
+    let p2w_config_addr = P2WConfigAccount::<{ AccountState::Initialized }>::key(None, &p2w_addr);
+    if symbols.len() > p2w_config.max_batch_size as usize {
+        return Err(format!(
+            "Expected up to {} symbols for batch, {} were found",
+            p2w_config.max_batch_size,
+            symbols.len()
+        )
+        .into());
+    }
+    // Initial attest() accounts
+    let mut acc_metas = vec![
+        // payer
+        AccountMeta::new(payer.pubkey(), true),
+        // system_program
+        AccountMeta::new_readonly(system_program::id(), false),
+        // config
+        AccountMeta::new_readonly(p2w_config_addr, false),
+    ];
+
+    // Batch contents and padding if applicable
+    let mut padded_symbols = {
+        let mut not_padded: Vec<_> = symbols
+            .iter()
+            .map(|s| {
+                vec![
+                    AccountMeta::new_readonly(s.product_addr, false),
+                    AccountMeta::new_readonly(s.price_addr, false),
+                ]
+            })
+            .flatten()
+            .collect();
+
+        // Align to max batch size with null accounts
+        let mut padding_accounts =
+            vec![
+                AccountMeta::new_readonly(Pubkey::new_from_array([0u8; 32]), false);
+                2 * (p2w_config.max_batch_size as usize - symbols.len())
+            ];
+        not_padded.append(&mut padding_accounts);
+
+        not_padded
+    };
+
+    acc_metas.append(&mut padded_symbols);
+
+    // Continue with other pyth2wormhole accounts
+    let mut acc_metas_remainder = vec![
+        // clock
+        AccountMeta::new_readonly(clock::id(), false),
+        // wh_prog
+        AccountMeta::new_readonly(p2w_config.wh_prog, false),
+        // wh_bridge
+        AccountMeta::new(
+            Bridge::<{ AccountState::Initialized }>::key(None, &p2w_config.wh_prog),
+            false,
+        ),
+        // wh_message
+        AccountMeta::new(wh_msg.pubkey(), true),
+        // wh_emitter
+        AccountMeta::new_readonly(emitter_addr, false),
+        // wh_sequence
+        AccountMeta::new(seq_addr, false),
+        // wh_fee_collector
+        AccountMeta::new(FeeCollector::<'_>::key(None, &p2w_config.wh_prog), false),
+        AccountMeta::new_readonly(rent::id(), false),
+    ];
+
+    acc_metas.append(&mut acc_metas_remainder);
+
+    let ix_data = (
+        pyth2wormhole::instruction::Instruction::Attest,
+        AttestData {
+            consistency_level: ConsistencyLevel::Finalized,
+        },
+    );
+
+    let ix = Instruction::new_with_bytes(p2w_addr, ix_data.try_to_vec()?.as_slice(), acc_metas);
+
+    let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
+        &[ix],
+        Some(&payer.pubkey()),
+        &vec![&payer, &wh_msg],
+        latest_blockhash,
+    );
+    Ok(tx_signed)
+}

solana/pyth2wormhole/client/src/main.rs

@@ -1,18 +1,7 @@
-pub mod attestation_cfg;
 pub mod cli;
 
-use std::{
-    fs::File,
-    path::{
-        Path,
-        PathBuf,
-    },
-};
+use std::fs::File;
 
-use borsh::{
-    BorshDeserialize,
-    BorshSerialize,
-};
 use clap::Clap;
 use log::{
     debug,
@@ -22,68 +11,26 @@ use log::{
     LevelFilter,
 };
 use solana_client::rpc_client::RpcClient;
-use solana_program::{
-    hash::Hash,
-    instruction::{
-        AccountMeta,
-        Instruction,
-    },
-    pubkey::Pubkey,
-    system_program,
-    sysvar::{
-        clock,
-        rent,
-    },
-};
+use solana_program::pubkey::Pubkey;
 use solana_sdk::{
     commitment_config::CommitmentConfig,
     signature::read_keypair_file,
-    transaction::Transaction,
 };
 use solana_transaction_status::UiTransactionEncoding;
 use solitaire::{
     processors::seeded::Seeded,
-    AccountState,
-    Derive,
-    Info,
-};
-use solitaire_client::{
-    AccEntry,
-    Keypair,
-    SolSigner,
-    ToInstruction,
+    ErrBox,
 };
+use solitaire_client::Keypair;
 
 use cli::{
     Action,
     Cli,
 };
 
-use bridge::{
-    accounts::{
-        Bridge,
-        FeeCollector,
-        Sequence,
-        SequenceDerivationData,
-    },
-    types::ConsistencyLevel,
-    CHAIN_ID_SOLANA,
-};
-
-use pyth2wormhole::{
-    attest::P2W_MAX_BATCH_SIZE,
-    config::P2WConfigAccount,
-    initialize::InitializeAccounts,
-    set_config::SetConfigAccounts,
-    AttestData,
-    Pyth2WormholeConfig,
-};
-
 use p2w_sdk::P2WEmitter;
 
-use crate::attestation_cfg::AttestationConfig;
-
-pub type ErrBox = Box<dyn std::error::Error>;
+use pyth2wormhole_client::*;
 
 pub const SEQNO_PREFIX: &'static str = "Program log: Sequence: ";
 
@@ -104,7 +51,7 @@ fn main() -> Result<(), ErrBox> {
             pyth_owner_addr,
             wh_prog,
         } => {
-            let tx = handle_init(
+            let tx = gen_init_tx(
                 payer,
                 p2w_addr,
                 owner_addr,
@@ -114,13 +61,16 @@ fn main() -> Result<(), ErrBox> {
             )?;
             rpc_client.send_and_confirm_transaction_with_spinner(&tx)?;
         }
+        Action::GetConfig => {
+            println!("{:?}", get_config_account(&rpc_client, &p2w_addr)?);
+        }
         Action::SetConfig {
             ref owner,
             new_owner_addr,
             new_wh_prog,
             new_pyth_owner_addr,
         } => {
-            let tx = handle_set_config(
+            let tx = gen_set_config_tx(
                 payer,
                 p2w_addr,
                 read_keypair_file(&*shellexpand::tilde(&owner))?,
@@ -133,118 +83,32 @@ fn main() -> Result<(), ErrBox> {
         }
         Action::Attest {
             ref attestation_cfg,
+            n_retries,
         } => {
             // Load the attestation config yaml
             let attestation_cfg: AttestationConfig =
                 serde_yaml::from_reader(File::open(attestation_cfg)?)?;
 
-            handle_attest(&rpc_client, payer, p2w_addr, &attestation_cfg)?;
+            handle_attest(&rpc_client, payer, p2w_addr, &attestation_cfg, n_retries)?;
         }
     }
 
     Ok(())
 }
 
-fn handle_init(
-    payer: Keypair,
-    p2w_addr: Pubkey,
-    new_owner_addr: Pubkey,
-    wh_prog: Pubkey,
-    pyth_owner_addr: Pubkey,
-    latest_blockhash: Hash,
-) -> Result<Transaction, ErrBox> {
-    use AccEntry::*;
-
-    let payer_pubkey = payer.pubkey();
-
-    let accs = InitializeAccounts {
-        payer: Signer(payer),
-        new_config: Derived(p2w_addr),
-    };
-
-    let config = Pyth2WormholeConfig {
-        max_batch_size: P2W_MAX_BATCH_SIZE,
-        owner: new_owner_addr,
-        wh_prog: wh_prog,
-        pyth_owner: pyth_owner_addr,
-    };
-    let ix_data = (pyth2wormhole::instruction::Instruction::Initialize, config);
-
-    let (ix, signers) = accs.to_ix(p2w_addr, ix_data.try_to_vec()?.as_slice())?;
-
-    let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
-        &[ix],
-        Some(&payer_pubkey),
-        signers.iter().collect::<Vec<_>>().as_ref(),
-        latest_blockhash,
-    );
-    Ok(tx_signed)
-}
-
-fn handle_set_config(
-    payer: Keypair,
-    p2w_addr: Pubkey,
-    owner: Keypair,
-    new_owner_addr: Pubkey,
-    new_wh_prog: Pubkey,
-    new_pyth_owner_addr: Pubkey,
-    latest_blockhash: Hash,
-) -> Result<Transaction, ErrBox> {
-    use AccEntry::*;
-
-    let payer_pubkey = payer.pubkey();
-
-    let accs = SetConfigAccounts {
-        payer: Signer(payer),
-        current_owner: Signer(owner),
-        config: Derived(p2w_addr),
-    };
-
-    let config = Pyth2WormholeConfig {
-        max_batch_size: P2W_MAX_BATCH_SIZE,
-        owner: new_owner_addr,
-        wh_prog: new_wh_prog,
-        pyth_owner: new_pyth_owner_addr,
-    };
-    let ix_data = (pyth2wormhole::instruction::Instruction::SetConfig, config);
-
-    let (ix, signers) = accs.to_ix(p2w_addr, ix_data.try_to_vec()?.as_slice())?;
-
-    let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
-        &[ix],
-        Some(&payer_pubkey),
-        signers.iter().collect::<Vec<_>>().as_ref(),
-        latest_blockhash,
-    );
-    Ok(tx_signed)
-}
-
 fn handle_attest(
     rpc_client: &RpcClient, // Needed for reading Pyth account data
     payer: Keypair,
     p2w_addr: Pubkey,
     attestation_cfg: &AttestationConfig,
+    n_retries: usize,
 ) -> Result<(), ErrBox> {
     // Derive seeded accounts
     let emitter_addr = P2WEmitter::key(None, &p2w_addr);
 
     info!("Using emitter addr {}", emitter_addr);
 
-    let p2w_config_addr = P2WConfigAccount::<{ AccountState::Initialized }>::key(None, &p2w_addr);
-
-    let config = Pyth2WormholeConfig::try_from_slice(
-        rpc_client.get_account_data(&p2w_config_addr)?.as_slice(),
-    )?;
-
-    let seq_addr = Sequence::key(
-        &SequenceDerivationData {
-            emitter_key: &emitter_addr,
-        },
-        &config.wh_prog,
-    );
-
-    // Read the current max batch size from the contract's settings
-    let max_batch_size = config.max_batch_size;
+    let config = get_config_account(rpc_client, &p2w_addr)?;
 
     let batch_count = {
         let whole_batches = attestation_cfg.symbols.len() / config.max_batch_size as usize;
@@ -262,156 +126,132 @@ fn handle_attest(
     info!(
         "{} symbols read, max batch size {}, dividing into {} batches",
         attestation_cfg.symbols.len(),
-        max_batch_size,
+        config.max_batch_size,
         batch_count
     );
 
     let mut errors = Vec::new();
 
-    for (idx, symbols) in attestation_cfg
+    // Reused for failed batch retries
+    let mut batches: Vec<_> = attestation_cfg
         .symbols
         .as_slice()
-        .chunks(max_batch_size as usize)
+        .chunks(config.max_batch_size as usize)
         .enumerate()
-    {
-        let batch_no = idx + 1;
-        let sym_msg_keypair = Keypair::new();
-        info!(
-            "Batch {}/{} contents: {:?}",
-            batch_no,
-            batch_count,
-            symbols
-                .iter()
-                .map(|s| s
-                    .name
-                    .clone()
-                    .unwrap_or(format!("unnamed product {:?}", s.product_addr)))
-                .collect::<Vec<_>>()
-        );
-
-        let mut sym_metas_vec: Vec<_> = symbols
-            .iter()
-            .map(|s| {
-                vec![
-                    AccountMeta::new_readonly(s.product_addr, false),
-                    AccountMeta::new_readonly(s.price_addr, false),
-                ]
-            })
-            .flatten()
-            .collect();
-
-        // Align to max batch size with null accounts
-        let mut blank_accounts =
-            vec![
-                AccountMeta::new_readonly(Pubkey::new_from_array([0u8; 32]), false);
-                2 * (max_batch_size as usize - symbols.len())
-            ];
-        sym_metas_vec.append(&mut blank_accounts);
-
-        // Arrange Attest accounts
-        let mut acc_metas = vec![
-            // payer
-            AccountMeta::new(payer.pubkey(), true),
-            // system_program
-            AccountMeta::new_readonly(system_program::id(), false),
-            // config
-            AccountMeta::new_readonly(p2w_config_addr, false),
-        ];
-
-        // Insert max_batch_size metas
-        acc_metas.append(&mut sym_metas_vec);
-
-        // Continue with other pyth2wormhole accounts
-        let mut acc_metas_remainder = vec![
-            // clock
-            AccountMeta::new_readonly(clock::id(), false),
-            // wh_prog
-            AccountMeta::new_readonly(config.wh_prog, false),
-            // wh_bridge
-            AccountMeta::new(
-                Bridge::<{ AccountState::Initialized }>::key(None, &config.wh_prog),
-                false,
-            ),
-            // wh_message
-            AccountMeta::new(sym_msg_keypair.pubkey(), true),
-            // wh_emitter
-            AccountMeta::new_readonly(emitter_addr, false),
-            // wh_sequence
-            AccountMeta::new(seq_addr, false),
-            // wh_fee_collector
-            AccountMeta::new(FeeCollector::<'_>::key(None, &config.wh_prog), false),
-            AccountMeta::new_readonly(rent::id(), false),
-        ];
-
-        acc_metas.append(&mut acc_metas_remainder);
-
-        let ix_data = (
-            pyth2wormhole::instruction::Instruction::Attest,
-            AttestData {
-                consistency_level: ConsistencyLevel::Finalized,
-            },
-        );
-
-        let ix = Instruction::new_with_bytes(p2w_addr, ix_data.try_to_vec()?.as_slice(), acc_metas);
-
-        // Execute the transaction, obtain the resulting sequence
-        // number. The and_then() calls enforce error handling
-        // location near loop end.
-        let res = rpc_client
-            .get_latest_blockhash()
-            .and_then(|latest_blockhash| {
-                let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
-                    &[ix],
-                    Some(&payer.pubkey()),
-                    &vec![&payer, &sym_msg_keypair],
-                    latest_blockhash,
-                );
-                rpc_client.send_and_confirm_transaction_with_spinner(&tx_signed)
-            })
-            .and_then(|sig| rpc_client.get_transaction(&sig, UiTransactionEncoding::Json))
-            .map_err(|e| -> ErrBox { e.into() })
-            .and_then(|this_tx| {
-                this_tx
-                    .transaction
-                    .meta
-                    .and_then(|meta| meta.log_messages)
-                    .and_then(|logs| {
-                        let mut seqno = None;
-                        for log in logs {
-                            if log.starts_with(SEQNO_PREFIX) {
-                                seqno = Some(log.replace(SEQNO_PREFIX, ""));
-                                break;
+        .map(|(idx, symbols)| (idx + 1, symbols, 1))
+        .collect();
+
+    let mut batches4retry = Vec::new();
+
+    // If no batches are scheduled for retry, the vector eventually drains
+    while !batches.is_empty() {
+        for (batch_no, symbols, attempt_no) in batches {
+            info!(
+                "Batch {}/{} contents: {:?}",
+                batch_no,
+                batch_count,
+                symbols
+                    .iter()
+                    .map(|s| s
+                        .name
+                        .clone()
+                        .unwrap_or(format!("unnamed product {:?}", s.product_addr)))
+                    .collect::<Vec<_>>()
+            );
+
+            // Execute the transaction, obtain the resulting sequence
+            // number. The and_then() calls enforce permissible error
+            // handling location near loop end.
+            let res = rpc_client
+                .get_latest_blockhash()
+                .map_err(|e| -> ErrBox { e.into() })
+                .and_then(|latest_blockhash| {
+                    let tx_signed = gen_attest_tx(
+                        p2w_addr,
+                        &config,
+                        &payer,
+                        symbols,
+                        &Keypair::new(),
+                        latest_blockhash,
+                    )?;
+
+                    rpc_client
+                        .send_and_confirm_transaction_with_spinner(&tx_signed)
+                        .map_err(|e| -> ErrBox { e.into() })
+                })
+                .and_then(|sig| {
+                    rpc_client
+                        .get_transaction(&sig, UiTransactionEncoding::Json)
+                        .map_err(|e| -> ErrBox { e.into() })
+                })
+                .and_then(|this_tx| {
+                    this_tx
+                        .transaction
+                        .meta
+                        .and_then(|meta| meta.log_messages)
+                        .and_then(|logs| {
+                            let mut seqno = None;
+                            for log in logs {
+                                if log.starts_with(SEQNO_PREFIX) {
+                                    seqno = Some(log.replace(SEQNO_PREFIX, ""));
+                                    break;
+                                }
                             }
-                        }
-                        seqno
-                    })
-                    .ok_or_else(|| format!("No seqno in program logs").into())
-            });
-
-        // Individual batch errors mustn't prevent other batches from being sent.
-        match res {
-            Ok(seqno) => {
-                println!("Sequence number: {}", seqno);
-                info!("Batch {}/{}: OK, seqno {}", batch_no, batch_count, seqno);
-            }
-            Err(e) => {
-                let msg = format!(
-                    "Batch {}/{} tx error: {}",
-                    batch_no,
-                    batch_count,
-                    e.to_string()
-                );
-                error!("{}", &msg);
-
-                errors.push(msg)
+                            seqno
+                        })
+                        .ok_or_else(|| format!("No seqno in program logs").into())
+                });
+
+            // Individual batch errors mustn't prevent other batches from being sent.
+            match res {
+                Ok(seqno) => {
+                    println!("Sequence number: {}", seqno);
+                    info!("Batch {}/{}: OK, seqno {}", batch_no, batch_count, seqno);
+                }
+                Err(e) => {
+                    let msg = format!(
+                        "Batch {}/{} tx error (attempt {} of {}): {}",
+                        batch_no,
+                        batch_count,
+                        attempt_no,
+                        n_retries + 1,
+                        e.to_string()
+                    );
+                    warn!("{}", &msg);
+
+                    if attempt_no < n_retries + 1 {
+                        // Schedule this batch for a retry if we have retries left
+                        batches4retry.push((batch_no, symbols, attempt_no + 1));
+                    } else {
+                        // This batch failed all attempts, note the error but do not schedule for retry
+                        error!(
+                            "Batch {}/{}: All {} attempts failed",
+                            batch_no,
+                            batch_count,
+                            n_retries + 1
+                        );
+                        errors.push(msg);
+                    }
+                }
             }
         }
+
+        // Batches scheduled for retry become the list of batches for
+        // next attempt round, clear retry vec for future failed attempts.
+        batches = batches4retry;
+        batches4retry = Vec::new();
     }
 
-    if errors.len() > 0 {
+    if !errors.is_empty() {
         let err_list = errors.join("\n");
 
-        Err(format!("{} of {} batches failed:\n{}", errors.len(), batch_count, err_list).into())
+        Err(format!(
+            "{} of {} batches failed:\n{}",
+            errors.len(),
+            batch_count,
+            err_list
+        )
+        .into())
     } else {
         Ok(())
     }

solana/pyth2wormhole/program/Cargo.toml

@@ -26,3 +26,4 @@ p2w-sdk = { path = "../../../third_party/pyth/p2w-sdk/rust" }
 serde = { version = "1", optional = true}
 serde_derive = { version = "1", optional = true}
 serde_json = { version = "1", optional = true}
+

solana/pyth2wormhole/program/src/config.rs

@@ -19,6 +19,7 @@ use solitaire::{
 };
 
 #[derive(Default, BorshDeserialize, BorshSerialize)]
+#[cfg_attr(feature = "client", derive(Debug))]
 pub struct Pyth2WormholeConfig {
     ///  Authority owning this contract
     pub owner: Pubkey,