pyth2wormhole-client: refactor into fully-async futures-based approach (#219)

* pyth2wormhole-client: refactor into fully-async futures-based approach

commit-id:2ed35045

* p2w-client: Change inconsistent rpc constructor

commit-id:cb3b2ff2

* p2w-client: Move job creation to a function, simplify comment

commit-id:35328b38

* pyth2wormhole-client: Use get_multiple_accounts

commit-id:7fc85157

* Implement a rate-limited mutex for RPC client

commit-id:1a243063

* pyth2wormhole-client: only guard beginning new requests in RLMutex

commit-id:d8251474

* pyth2wormhole-client: RLMutex: ensure the inner guard is not dropped

commit-id:c3513f5e

* pyth2wormhole-client: Clarify attestation_sched_futs comment

commit-id:97033670

* pyth2wormhole-client: Use CommitmentConfig's native FromStr parsing

commit-id:835d7125

* pyth2wormhole-client: doc comment typo

commit-id:5ee388de

* pyth2wormhole-client: move closures to their own async functions

This makes the main.rs async attestation routines easier to read.

commit-id:3565a744

* pyth2wormhole-client: fix merge typo

* pyth2wormhole-client: Apply Tom's readability advice

* pyth2wormhole-client reword attestation_sched_job() comment

* pyth2wormhole-client: expand attestation_sched_job() comment

* pyth2wormhole-client:   e  x  p  a  n  d   the comment

* Trigger CI

* p2w-client/main.rs: correct missing awaits after merge

Author: Stanisław Drozd

solana/pyth2wormhole/Cargo.lock

@@ -29,6 +29,8 @@ solana-sdk = "=1.10.13"
 solana-transaction-status = "=1.10.13"
 solitaire-client = {path = "../../solitaire/client"}
 solitaire = {path = "../../solitaire/program"}
+tokio = {version = "1", features = ["sync", "rt", "time"]}
+futures = "0.3.21"
 
 [dev-dependencies]
 pyth-client = "0.5.0"

solana/pyth2wormhole/client/Cargo.toml

@@ -26,20 +26,30 @@ pub struct SymbolGroup {
     pub symbols: Vec<P2WSymbol>,
 }
 
-pub const fn DEFAULT_MIN_INTERVAL_SECS() -> u64 {
+pub const fn default_min_interval_secs() -> u64 {
     60
 }
 
+pub const fn default_max_batch_jobs() -> usize {
+    20
+}
+
 /// Spontaneous attestation triggers. Attestation is triggered if any
 /// of the active conditions is met. Option<> fields can be
 /// de-activated with None. All conditions are inactive by default,
-/// except for min_interval_secs set to 1 minute.
+/// except for the non-Option ones.
 #[derive(Clone, Default, Debug, Deserialize, Serialize, PartialEq)]
 pub struct AttestationConditions {
     /// Baseline, unconditional attestation interval. Attestation is triggered if the specified interval elapsed since last attestation.
-    #[serde(default = "DEFAULT_MIN_INTERVAL_SECS")]
+    #[serde(default = "default_min_interval_secs")]
     pub min_interval_secs: u64,
 
+    /// Limit concurrent attestation attempts per batch. This setting
+    /// should act only as a failsafe cap on resource consumption and is
+    /// best set well above the expected average number of jobs.
+    #[serde(default = "default_max_batch_jobs")]
+    pub max_batch_jobs: usize,
+
     /// Trigger attestation if price changes by the specified percentage.
     #[serde(default)]
     pub price_changed_pct: Option<f64>,
@@ -51,7 +61,7 @@ pub struct AttestationConditions {
 }
 
 /// Config entry for a Pyth product + price pair
-#[derive(Default, Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[derive(Clone, Default, Debug, Deserialize, Serialize, PartialEq, Eq)]
 pub struct P2WSymbol {
     /// User-defined human-readable name
     pub name: Option<String>,

solana/pyth2wormhole/client/src/attestation_cfg.rs

@@ -1,8 +1,10 @@
+use futures::future::TryFutureExt;
 use log::{
     debug,
+    trace,
     warn,
 };
-use solana_client::rpc_client::RpcClient;
+use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::signature::Signature;
 
 use pyth_sdk_solana::state::PriceAccount;
@@ -16,17 +18,18 @@ use crate::{
     AttestationConditions,
     ErrBox,
     P2WSymbol,
+    RLMutex,
 };
 
+/// Runtime representation of a batch. It refers to the original group
+/// from the config.
 #[derive(Debug)]
 pub struct BatchState<'a> {
     pub group_name: String,
     pub symbols: &'a [P2WSymbol],
     pub last_known_symbol_states: Vec<Option<PriceAccount>>,
     pub conditions: AttestationConditions,
-    status: BatchTxStatus,
-    status_changed_at: Instant,
-    pub last_success_at: Option<Instant>,
+    pub last_job_finished_at: Instant,
 }
 
 impl<'a> BatchState<'a> {
@@ -40,63 +43,51 @@ impl<'a> BatchState<'a> {
             symbols,
             conditions,
             last_known_symbol_states: vec![None; symbols.len()],
-            status: BatchTxStatus::Sending { attempt_no: 1 },
-            status_changed_at: Instant::now(),
-            last_success_at: None,
+            last_job_finished_at: Instant::now(),
         }
     }
-    /// Ensure only set_status() alters the timestamp
-    pub fn get_status_changed_at(&self) -> &Instant {
-        &self.status_changed_at
-    }
-    pub fn get_status(&self) -> &BatchTxStatus {
-        &self.status
-    }
-
-    /// Ensure that status changes are accompanied by a timestamp bump
-    pub fn set_status(&mut self, s: BatchTxStatus) {
-        self.status_changed_at = Instant::now();
-        self.status = s;
-    }
 
     /// Evaluate the configured attestation conditions for this
     /// batch. RPC is used to update last known state. Returns
     /// Some("<reason>") if any trigger condition was met. Only the
     /// first encountered condition is mentioned.
-    pub fn should_resend(&mut self, c: &RpcClient) -> Option<String> {
+    pub async fn should_resend(&mut self, c: &RpcClient) -> Option<String> {
         let mut ret = None;
 
         let sym_count = self.symbols.len();
-        let mut new_symbol_states: Vec<Option<PriceAccount>> = Vec::with_capacity(sym_count);
-        for (idx, sym) in self.symbols.iter().enumerate() {
-            let new_state = match c
-                .get_account_data(&sym.price_addr)
-                .map_err(|e| e.to_string())
-                .and_then(|bytes| {
-                    pyth_sdk_solana::state::load_price_account(&bytes)
-                        .map(|state| state.clone())
-                        .map_err(|e| e.to_string())
-                }) {
-                Ok(state) => Some(state),
-                Err(e) => {
-                    warn!(
-                        "Symbol {} ({}/{}): Could not look up state: {}",
-                        sym.name
-                            .as_ref()
-                            .unwrap_or(&format!("Unnamed product {}", sym.product_addr)),
-                        idx + 1,
-                        sym_count,
-                        e.to_string()
-                    );
-                    None
-                }
-            };
+        let pubkeys: Vec<_> = self.symbols.iter().map(|s| s.price_addr).collect();
 
-            new_symbol_states.push(new_state);
-        }
+        // Always learn the current on-chain state for each symbol, use None values if lookup fails
+        let mut new_symbol_states: Vec<Option<PriceAccount>> = match c
+            .get_multiple_accounts(&pubkeys)
+            .await
+        {
+            Ok(acc_opts) => {
+                acc_opts
+                    .into_iter()
+                    .enumerate()
+                    .map(|(idx, opt)| {
+                        // Take each Some(acc), make it None and log on load_price_account() error
+                        opt.and_then(|acc| {
+                            pyth_sdk_solana::state::load_price_account(&acc.data)
+                                .cloned() // load_price_account() transmutes the data reference into another reference, and owning acc_opts is not enough
+                                .map_err(|e| {
+                                    warn!("Could not parse symbol {}/{}: {}", idx, sym_count, e);
+                                    e
+                                })
+                                .ok() // Err becomes None
+                        })
+                    })
+                    .collect()
+            }
+            Err(e) => {
+                warn!("Could not look up any symbols on-chain: {}", e);
+                vec![None; sym_count]
+            }
+        };
 
         // min interval
-        if self.get_status_changed_at().elapsed()
+        if self.last_job_finished_at.elapsed()
             > Duration::from_secs(self.conditions.min_interval_secs)
         {
             ret = Some(format!(
@@ -154,7 +145,9 @@ impl<'a> BatchState<'a> {
             }
         }
 
-        // Update with newer state if a condition was met
+        // Update with newer state only if a condition was met. We
+        // don't want to shadow changes that may happen over a larger
+        // period between state lookups.
         if ret.is_some() {
             for (old, new) in self
                 .last_known_symbol_states
@@ -170,23 +163,3 @@ impl<'a> BatchState<'a> {
         return ret;
     }
 }
-
-#[derive(Debug)]
-pub enum BatchTxStatus {
-    Sending {
-        attempt_no: usize,
-    },
-    Confirming {
-        attempt_no: usize,
-        signature: Signature,
-    },
-    Success {
-        seqno: String,
-    },
-    FailedSend {
-        last_err: ErrBox,
-    },
-    FailedConfirm {
-        last_err: ErrBox,
-    },
-}

solana/pyth2wormhole/client/src/batch_state.rs

@@ -1,6 +1,7 @@
 //! CLI options
 
 use solana_program::pubkey::Pubkey;
+use solana_sdk::commitment_config::CommitmentConfig;
 use std::path::PathBuf;
 
 use clap::{
@@ -29,6 +30,14 @@ pub struct Cli {
     pub payer: String,
     #[clap(short, long, default_value = "http://localhost:8899")]
     pub rpc_url: String,
+    #[clap(
+        long = "rpc-interval",
+        default_value = "150",
+        help = "Rate-limiting minimum delay between RPC requests in milliseconds"
+    )]
+    pub rpc_interval_ms: u64,
+    #[clap(long, default_value = "confirmed")]
+    pub commitment: CommitmentConfig,
     #[clap(long)]
     pub p2w_addr: Pubkey,
     #[clap(subcommand)]
@@ -60,10 +69,18 @@ pub enum Action {
         #[clap(
             short = 'n',
             long = "--n-retries",
-            help = "How many times to retry send_transaction() on each batch before flagging a failure.",
+            help = "How many times to retry send_transaction() on each batch before flagging a failure. Only active outside daemon mode",
             default_value = "5"
         )]
         n_retries: usize,
+        #[clap(
+            short = 'i',
+            long = "--retry-interval",
+            help = "How long to wait between send_transaction
+            retries. Only active outside daemon mode",
+            default_value = "5"
+        )]
+        retry_interval_secs: u64,
         #[clap(
             short = 'd',
             long = "--daemon",
@@ -73,17 +90,10 @@ pub enum Action {
         #[clap(
             short = 't',
             long = "--timeout",
-            help = "How many seconds to wait before giving up on get_transaction() for tx confirmation.",
-            default_value = "40"
-        )]
-        conf_timeout_secs: u64,
-        #[clap(
-            short = 'i',
-            long = "--rpc-interval",
-            help = "How many milliseconds to wait between SOL RPC requests",
-            default_value = "200"
+            help = "How many seconds to wait before giving up on  tx confirmation.",
+            default_value = "20"
         )]
-        rpc_interval_ms: u64,
+        confirmation_timeout_secs: u64,
     },
     #[clap(about = "Retrieve a pyth2wormhole program's current settings")]
     GetConfig,

solana/pyth2wormhole/client/src/cli.rs

@@ -1,11 +1,12 @@
 pub mod attestation_cfg;
 pub mod batch_state;
+pub mod util;
 
 use borsh::{
     BorshDeserialize,
     BorshSerialize,
 };
-use solana_client::rpc_client::RpcClient;
+use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_program::{
     hash::Hash,
     instruction::{
@@ -51,19 +52,24 @@ use pyth2wormhole::{
     migrate::MigrateAccounts,
     set_config::SetConfigAccounts,
     AttestData,
-    Pyth2WormholeConfig,
 };
 
+pub use pyth2wormhole::Pyth2WormholeConfig;
+
 pub use attestation_cfg::{
     AttestationConditions,
     AttestationConfig,
     P2WSymbol,
 };
-pub use batch_state::{
-    BatchState,
-    BatchTxStatus,
+pub use batch_state::BatchState;
+pub use util::{
+    RLMutex,
+    RLMutexGuard,
 };
 
+/// Future-friendly version of solitaire::ErrBox
+pub type ErrBoxSend = Box<dyn std::error::Error + Send + Sync>;
+
 pub fn gen_init_tx(
     payer: Keypair,
     p2w_addr: Pubkey,
@@ -159,14 +165,17 @@ pub fn gen_migrate_tx(
 }
 
 /// Get the current config account data for given p2w program address
-pub fn get_config_account(
+pub async 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(),
+        rpc_client
+            .get_account_data(&p2w_config_addr)
+            .await?
+            .as_slice(),
     )?;
 
     Ok(config)
@@ -181,7 +190,7 @@ pub fn gen_attest_tx(
     symbols: &[P2WSymbol],
     wh_msg: &Keypair,
     latest_blockhash: Hash,
-) -> Result<Transaction, ErrBox> {
+) -> Result<Transaction, ErrBoxSend> {
     let emitter_addr = P2WEmitter::key(None, &p2w_addr);
 
     let seq_addr = Sequence::key(
@@ -193,11 +202,11 @@ pub fn gen_attest_tx(
 
     let p2w_config_addr = P2WConfigAccount::<{ AccountState::Initialized }>::key(None, &p2w_addr);
     if symbols.len() > p2w_config.max_batch_size as usize {
-        return Err(format!(
+        return Err((format!(
             "Expected up to {} symbols for batch, {} were found",
             p2w_config.max_batch_size,
             symbols.len()
-        )
+        ))
         .into());
     }
     // Initial attest() accounts
@@ -267,7 +276,14 @@ pub fn gen_attest_tx(
         },
     );
 
-    let ix = Instruction::new_with_bytes(p2w_addr, ix_data.try_to_vec()?.as_slice(), acc_metas);
+    let ix = Instruction::new_with_bytes(
+        p2w_addr,
+        ix_data
+            .try_to_vec()
+            .map_err(|e| -> ErrBoxSend { Box::new(e) })?
+            .as_slice(),
+        acc_metas,
+    );
 
     let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
         &[ix],