Wormhole message reuse via post_message_unreliable (#261)

* pyth2wormhole: Implement reusable message PDAs

This changeset converts attest() to use a new PDA for
reusable/unreliable wormhole message account. Each PDA is tied to a
given attest() payer with an index that lets them rotate a number of
message accounts. Keeping the appropriate timing of the reuses is up
to the message owner, who should rotate a number of PDAs.

* pyth2wormhole-client: Add a message acc rotation impl

* p2w attest(): Integrate with bumped wormhole and fix call issues

* p2w-client: Format code, fix test_attest, refactor message index

* Dockerfile.client, Dockerfile.p2w-attest: Improve caching

The main improvement comes from running cargo-install from within a
workspace which lets us cache target/

* p2w-client: Make reusable messages configurable in yaml

* p2w on-chain: refactor to unreliable-only, adjust msg balance, nits

* p2w-client: P2WMessageIndex -> P2WMessageQueue, queue tests

* p2w-client: Add a hard limit to the message account queue

Author: Stanisław Drozd

Dockerfile.client

@@ -9,7 +9,7 @@ RUN apt-get update && apt-get install -yq libudev-dev ncat
 RUN curl -fsSL https://deb.nodesource.com/setup_16.x | bash - && apt-get install -y nodejs
 
 COPY solana /usr/src/solana 
-WORKDIR /usr/src/solana
+WORKDIR /usr/src/solana/pyth2wormhole
 
 RUN --mount=type=cache,target=/root/.cache \
     cargo install --version =2.0.12 --locked spl-token-cli
@@ -23,6 +23,7 @@ ENV BRIDGE_ADDRESS="Bridge1p5gheXUvJ6jGWGeCsgPKgnE3YgdGKRVCMY9o"
 
 RUN --mount=type=cache,target=/root/.cache \
     --mount=type=cache,target=/usr/local/cargo/registry,id=cargo_registry \
+    --mount=type=cache,target=target,id=cargo_registry \
 	set -xe && \
     cargo install bridge_client --git https://github.com/certusone/wormhole --tag $WORMHOLE_TAG --locked --root /usr/local && \
     cargo install token_bridge_client --git https://github.com/certusone/wormhole --tag $WORMHOLE_TAG --locked --root /usr/local

solana/pyth2wormhole/client/src/attestation_cfg.rs

@@ -15,6 +15,10 @@ use solana_program::pubkey::Pubkey;
 /// Pyth2wormhole config specific to attestation requests
 #[derive(Debug, Deserialize, Serialize, PartialEq)]
 pub struct AttestationConfig {
+    #[serde(default = "default_min_msg_reuse_interval_ms")]
+    pub min_msg_reuse_interval_ms: u64,
+    #[serde(default = "default_max_msg_accounts")]
+    pub max_msg_accounts: u64,
     pub symbol_groups: Vec<SymbolGroup>,
 }
 
@@ -26,6 +30,14 @@ pub struct SymbolGroup {
     pub symbols: Vec<P2WSymbol>,
 }
 
+pub const fn default_max_msg_accounts() -> u64 {
+    1_000_000
+}
+
+pub const fn default_min_msg_reuse_interval_ms() -> u64 {
+    10_000 // 10s
+}
+
 pub const fn default_min_interval_secs() -> u64 {
     60
 }
@@ -149,6 +161,8 @@ mod tests {
         };
 
         let cfg = AttestationConfig {
+            min_msg_reuse_interval_ms: 1000,
+            max_msg_accounts: 100_000,
             symbol_groups: vec![fastbois, slowbois],
         };
 

solana/pyth2wormhole/client/src/cli.rs

@@ -116,7 +116,9 @@ pub enum Action {
         #[clap(long = "is-active")]
         is_active: Option<bool>,
     },
-    #[clap(about = "Migrate existing pyth2wormhole program settings to a newer format version. Client version must match the deployed contract.")]
+    #[clap(
+        about = "Migrate existing pyth2wormhole program settings to a newer format version. Client version must match the deployed contract."
+    )]
     Migrate {
         /// owner keypair path
         #[clap(

solana/pyth2wormhole/client/src/lib.rs

@@ -1,5 +1,6 @@
 pub mod attestation_cfg;
 pub mod batch_state;
+pub mod message;
 pub mod util;
 
 use borsh::{
@@ -20,7 +21,13 @@ use solana_program::{
         rent,
     },
 };
-use solana_sdk::{transaction::Transaction, signer::{Signer, keypair::Keypair}};
+use solana_sdk::{
+    signer::{
+        keypair::Keypair,
+        Signer,
+    },
+    transaction::Transaction,
+};
 use solitaire::{
     processors::seeded::Seeded,
     AccountState,
@@ -41,7 +48,14 @@ use p2w_sdk::P2WEmitter;
 
 use pyth2wormhole::{
     attest::P2W_MAX_BATCH_SIZE,
-    config::{OldP2WConfigAccount, P2WConfigAccount},
+    config::{
+        OldP2WConfigAccount,
+        P2WConfigAccount,
+    },
+    message::{
+        P2WMessage,
+        P2WMessageDrvData,
+    },
     AttestData,
 };
 
@@ -58,6 +72,8 @@ pub use util::{
     RLMutexGuard,
 };
 
+pub use message::P2WMessageQueue;
+
 /// Future-friendly version of solitaire::ErrBox
 pub type ErrBoxSend = Box<dyn std::error::Error + Send + Sync>;
 
@@ -67,26 +83,22 @@ pub fn gen_init_tx(
     config: Pyth2WormholeConfig,
     latest_blockhash: Hash,
 ) -> Result<Transaction, ErrBox> {
-
     let payer_pubkey = payer.pubkey();
     let acc_metas = vec![
         // new_config
-        AccountMeta::new(P2WConfigAccount::<{AccountState::Uninitialized}>::key(None, &p2w_addr), false),
+        AccountMeta::new(
+            P2WConfigAccount::<{ AccountState::Uninitialized }>::key(None, &p2w_addr),
+            false,
+        ),
         // payer
         AccountMeta::new(payer.pubkey(), true),
         // system_program
         AccountMeta::new(system_program::id(), false),
-        ];
+    ];
 
     let ix_data = (pyth2wormhole::instruction::Instruction::Initialize, config);
 
-    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()?.as_slice(), acc_metas);
 
     let signers = vec![&payer];
 
@@ -110,25 +122,22 @@ pub fn gen_set_config_tx(
 
     let acc_metas = vec![
         // config
-        AccountMeta::new(P2WConfigAccount::<{AccountState::Initialized}>::key(None, &p2w_addr), false),
+        AccountMeta::new(
+            P2WConfigAccount::<{ AccountState::Initialized }>::key(None, &p2w_addr),
+            false,
+        ),
         // current_owner
         AccountMeta::new(owner.pubkey(), true),
         // payer
         AccountMeta::new(payer.pubkey(), true),
-        ];
+    ];
 
     let ix_data = (
         pyth2wormhole::instruction::Instruction::SetConfig,
         new_config,
     );
 
-    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()?.as_slice(), acc_metas);
 
     let signers = vec![&owner, &payer];
     let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
@@ -146,12 +155,14 @@ pub fn gen_migrate_tx(
     owner: Keypair,
     latest_blockhash: Hash,
 ) -> Result<Transaction, ErrBox> {
-
     let payer_pubkey = payer.pubkey();
 
     let acc_metas = vec![
         // new_config
-        AccountMeta::new(P2WConfigAccount::<{AccountState::Uninitialized}>::key(None, &p2w_addr), false),
+        AccountMeta::new(
+            P2WConfigAccount::<{ AccountState::Uninitialized }>::key(None, &p2w_addr),
+            false,
+        ),
         // old_config
         AccountMeta::new(OldP2WConfigAccount::key(None, &p2w_addr), false),
         // owner
@@ -160,20 +171,11 @@ pub fn gen_migrate_tx(
         AccountMeta::new(payer.pubkey(), true),
         // system_program
         AccountMeta::new(system_program::id(), false),
-        ];
+    ];
 
-    let ix_data = (
-        pyth2wormhole::instruction::Instruction::Migrate,
-        (),
-    );
+    let ix_data = (pyth2wormhole::instruction::Instruction::Migrate, ());
 
-    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()?.as_slice(), acc_metas);
 
     let signers = vec![&owner, &payer];
 
@@ -209,8 +211,8 @@ pub fn gen_attest_tx(
     p2w_addr: Pubkey,
     p2w_config: &Pyth2WormholeConfig, // Must be fresh, not retrieved inside to keep side effects away
     payer: &Keypair,
+    wh_msg_id: u64,
     symbols: &[P2WSymbol],
-    wh_msg: &Keypair,
     latest_blockhash: Hash,
 ) -> Result<Transaction, ErrBoxSend> {
     let emitter_addr = P2WEmitter::key(None, &p2w_addr);
@@ -279,7 +281,16 @@ pub fn gen_attest_tx(
             false,
         ),
         // wh_message
-        AccountMeta::new(wh_msg.pubkey(), true),
+        AccountMeta::new(
+            P2WMessage::key(
+                &P2WMessageDrvData {
+                    id: wh_msg_id,
+                    message_owner: payer.pubkey(),
+                },
+                &p2w_addr,
+            ),
+            false,
+        ),
         // wh_emitter
         AccountMeta::new_readonly(emitter_addr, false),
         // wh_sequence
@@ -295,21 +306,16 @@ pub fn gen_attest_tx(
         pyth2wormhole::instruction::Instruction::Attest,
         AttestData {
             consistency_level: ConsistencyLevel::Confirmed,
+            message_account_id: wh_msg_id,
         },
     );
 
-    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()?.as_slice(), acc_metas);
 
     let tx_signed = Transaction::new_signed_with_payer::<Vec<&Keypair>>(
         &[ix],
         Some(&payer.pubkey()),
-        &vec![&payer, &wh_msg],
+        &vec![&payer],
         latest_blockhash,
     );
     Ok(tx_signed)

solana/pyth2wormhole/client/src/main.rs

@@ -45,7 +45,10 @@ use solitaire::{
     ErrBox,
 };
 use tokio::{
-    sync::Semaphore,
+    sync::{
+        Mutex,
+        Semaphore,
+    },
     task::JoinHandle,
 };
 
@@ -136,16 +139,16 @@ async fn main() -> Result<(), ErrBox> {
                 get_config_account(&rpc_client, &p2w_addr).await?
             );
         }
-        Action::Migrate {
-            ref owner,
-        } => {
+        Action::Migrate { ref owner } => {
             let tx = gen_migrate_tx(
                 payer,
                 p2w_addr,
                 read_keypair_file(&*shellexpand::tilde(&owner))?,
                 latest_blockhash,
             )?;
-            rpc_client.send_and_confirm_transaction_with_spinner(&tx).await?;
+            rpc_client
+                .send_and_confirm_transaction_with_spinner(&tx)
+                .await?;
             println!(
                 "Applied conifg:\n{:?}",
                 get_config_account(&rpc_client, &p2w_addr).await?
@@ -252,6 +255,8 @@ async fn handle_attest(
         rpc_interval,
     ));
 
+    let message_q_mtx = Arc::new(Mutex::new(P2WMessageQueue::new(Duration::from_millis(attestation_cfg.min_msg_reuse_interval_ms), attestation_cfg.max_msg_accounts as usize)));
+
     // Create attestation scheduling routines; see attestation_sched_job() for details
     let mut attestation_sched_futs = batches.into_iter().map(|(batch_no, batch)| {
         attestation_sched_job(
@@ -265,6 +270,7 @@ async fn handle_attest(
             p2w_addr,
             config.clone(),
             Keypair::from_bytes(&payer.to_bytes()).unwrap(),
+            message_q_mtx.clone(),
         )
     });
 
@@ -337,6 +343,7 @@ async fn attestation_sched_job(
     p2w_addr: Pubkey,
     config: Pyth2WormholeConfig,
     payer: Keypair,
+    message_q_mtx: Arc<Mutex<P2WMessageQueue>>,
 ) -> Result<(), ErrBoxSend> {
     let mut retries_left = n_retries;
     // Enforces the max batch job count
@@ -357,6 +364,7 @@ async fn attestation_sched_job(
             Keypair::from_bytes(&payer.to_bytes()).unwrap(), // Keypair has no clone
             batch.symbols.to_vec(),
             sema.clone(),
+            message_q_mtx.clone(),
         );
 
         if daemon {
@@ -456,6 +464,7 @@ async fn attestation_job(
     payer: Keypair,
     symbols: Vec<P2WSymbol>,
     max_jobs_sema: Arc<Semaphore>,
+    message_q_mtx: Arc<Mutex<P2WMessageQueue>>,
 ) -> Result<(), ErrBoxSend> {
     // Will be dropped after attestation is complete
     let _permit = max_jobs_sema.acquire().await?;
@@ -470,17 +479,18 @@ async fn attestation_job(
         .map_err(|e| -> ErrBoxSend { e.into() })
         .await?;
 
+    let wh_msg_id = message_q_mtx.lock().await.get_account()?.id;
+
     let tx_res: Result<_, ErrBoxSend> = gen_attest_tx(
         p2w_addr,
         &config,
         &payer,
+        wh_msg_id,
         symbols.as_slice(),
-        &Keypair::new(),
         latest_blockhash,
     );
-    let tx = tx_res?;
     let sig = rpc
-        .send_and_confirm_transaction(&tx)
+        .send_and_confirm_transaction(&tx_res?)
         .map_err(|e| -> ErrBoxSend { e.into() })
         .await?;
     let tx_data = rpc