From dfc2df779c87e470d71e7080c26a43448d4b9de2 Mon Sep 17 00:00:00 2001 From: Ali Behjati Date: Fri, 23 Jun 2023 19:31:44 +0200 Subject: [PATCH] [hermes] Use raw_message for future-compatibility (#913) --- hermes/src/network/pythnet.rs | 39 +---------------------- hermes/src/store.rs | 20 ++++++++---- hermes/src/store/proof/wormhole_merkle.rs | 22 +++++-------- hermes/src/store/storage.rs | 36 +++++++++++---------- hermes/src/store/storage/local_storage.rs | 9 +++--- hermes/src/store/types.rs | 24 ++++++++------ 6 files changed, 62 insertions(+), 88 deletions(-) diff --git a/hermes/src/network/pythnet.rs b/hermes/src/network/pythnet.rs index 88dc155d..4f3c8929 100644 --- a/hermes/src/network/pythnet.rs +++ b/hermes/src/network/pythnet.rs @@ -20,9 +20,7 @@ use { Result, }, borsh::BorshDeserialize, - byteorder::BE, futures::stream::StreamExt, - pythnet_sdk::messages::Message, solana_account_decoder::UiAccountEncoding, solana_client::{ nonblocking::{ @@ -157,44 +155,9 @@ pub async fn run(store: Arc, pythnet_ws_endpoint: String) -> Result { } }; - // The validators writes the accumulator messages using Borsh with - // the following struct. We cannot directly have messages as Vec - // because they are serialized using big-endian byte order and Borsh - // uses little-endian byte order. - #[derive(BorshDeserialize)] - struct RawAccumulatorMessages { - pub magic: [u8; 4], - pub slot: u64, - pub ring_size: u32, - pub raw_messages: Vec>, - } - - let accumulator_messages = RawAccumulatorMessages::try_from_slice(&account.data); + let accumulator_messages = AccumulatorMessages::try_from_slice(&account.data); match accumulator_messages { Ok(accumulator_messages) => { - let messages = accumulator_messages - .raw_messages - .iter() - .map(|message| { - pythnet_sdk::wire::from_slice::(message.as_slice()) - }) - .collect::, _>>(); - - let messages = match messages { - Ok(messages) => messages, - Err(err) => { - log::error!("Failed to parse messages: {:?}", err); - continue; - } - }; - - let accumulator_messages = AccumulatorMessages { - magic: accumulator_messages.magic, - slot: accumulator_messages.slot, - ring_size: accumulator_messages.ring_size, - messages, - }; - let (candidate, _) = Pubkey::find_program_address( &[ b"AccumulatorState", diff --git a/hermes/src/store.rs b/hermes/src/store.rs index e9e026e1..f8a741b9 100644 --- a/hermes/src/store.rs +++ b/hermes/src/store.rs @@ -33,15 +33,19 @@ use { anyhow, Result, }, + byteorder::BigEndian, pyth_sdk::PriceIdentifier, pythnet_sdk::{ messages::{ Message, MessageType, }, - wire::v1::{ - WormholeMessage, - WormholePayload, + wire::{ + from_slice, + v1::{ + WormholeMessage, + WormholePayload, + }, }, }, std::{ @@ -199,12 +203,14 @@ impl Store { let message_states = completed_state .accumulator_messages - .messages - .iter() + .raw_messages + .into_iter() .enumerate() - .map(|(idx, message)| { + .map(|(idx, raw_message)| { Ok(MessageState::new( - *message, + from_slice::(raw_message.as_ref()) + .map_err(|e| anyhow!("Failed to deserialize message: {:?}", e))?, + raw_message, ProofSet { wormhole_merkle_proof: wormhole_merkle_message_states_proofs .get(idx) diff --git a/hermes/src/store/proof/wormhole_merkle.rs b/hermes/src/store/proof/wormhole_merkle.rs index 818c1710..7450a55b 100644 --- a/hermes/src/store/proof/wormhole_merkle.rs +++ b/hermes/src/store/proof/wormhole_merkle.rs @@ -76,26 +76,20 @@ pub fn construct_message_states_proofs( let accumulator_messages = &completed_accumulator_state.accumulator_messages; let wormhole_merkle_state = &completed_accumulator_state.wormhole_merkle_state; - let raw_messages = accumulator_messages - .messages - .iter() - .map(|m| { - to_vec::<_, byteorder::BE>(m).map_err(|e| anyhow!("Failed to serialize message: {}", e)) - }) - .collect::>>>()?; - // Check whether the state is valid - let merkle_acc = - match MerkleTree::::from_set(raw_messages.iter().map(|m| m.as_ref())) { - Some(merkle_acc) => merkle_acc, - None => return Ok(vec![]), // It only happens when the message set is empty - }; + let merkle_acc = match MerkleTree::::from_set( + accumulator_messages.raw_messages.iter().map(|m| m.as_ref()), + ) { + Some(merkle_acc) => merkle_acc, + None => return Ok(vec![]), // It only happens when the message set is empty + }; if merkle_acc.root.as_bytes() != wormhole_merkle_state.root.root { return Err(anyhow!("Invalid merkle root")); } - raw_messages + accumulator_messages + .raw_messages .iter() .map(|m| { Ok(WormholeMerkleMessageProof { diff --git a/hermes/src/store/storage.rs b/hermes/src/store/storage.rs index 7a2c681f..371d708c 100644 --- a/hermes/src/store/storage.rs +++ b/hermes/src/store/storage.rs @@ -4,6 +4,7 @@ use { types::{ AccumulatorMessages, ProofSet, + RawMessage, RequestTime, Slot, UnixTimestamp, @@ -71,6 +72,7 @@ pub struct MessageStateTime { pub struct MessageState { pub slot: Slot, pub message: Message, + pub raw_message: RawMessage, pub proof_set: ProofSet, pub received_at: UnixTimestamp, } @@ -92,6 +94,7 @@ impl MessageState { pub fn new( message: Message, + raw_message: RawMessage, proof_set: ProofSet, slot: Slot, received_at: UnixTimestamp, @@ -99,6 +102,7 @@ impl MessageState { Self { slot, message, + raw_message, proof_set, received_at, } @@ -151,20 +155,20 @@ mod test { wormhole_merkle_state: None, }; - assert!(CompletedAccumulatorState::try_from(accumulator_state.clone()).is_err()); + assert!(CompletedAccumulatorState::try_from(accumulator_state).is_err()); let accumulator_state = AccumulatorState { slot: 1, accumulator_messages: Some(AccumulatorMessages { - slot: 1, - magic: [0; 4], - ring_size: 10, - messages: vec![], + slot: 1, + magic: [0; 4], + ring_size: 10, + raw_messages: vec![], }), wormhole_merkle_state: None, }; - assert!(CompletedAccumulatorState::try_from(accumulator_state.clone()).is_err()); + assert!(CompletedAccumulatorState::try_from(accumulator_state).is_err()); let accumulator_state = AccumulatorState { slot: 1, @@ -179,15 +183,15 @@ mod test { }), }; - assert!(CompletedAccumulatorState::try_from(accumulator_state.clone()).is_err()); + assert!(CompletedAccumulatorState::try_from(accumulator_state).is_err()); let accumulator_state = AccumulatorState { slot: 1, accumulator_messages: Some(AccumulatorMessages { - slot: 1, - magic: [0; 4], - ring_size: 10, - messages: vec![], + slot: 1, + magic: [0; 4], + ring_size: 10, + raw_messages: vec![], }), wormhole_merkle_state: Some(WormholeMerkleState { vaa: vec![], @@ -200,14 +204,14 @@ mod test { }; assert_eq!( - CompletedAccumulatorState::try_from(accumulator_state.clone()).unwrap(), + CompletedAccumulatorState::try_from(accumulator_state).unwrap(), CompletedAccumulatorState { slot: 1, accumulator_messages: AccumulatorMessages { - slot: 1, - magic: [0; 4], - ring_size: 10, - messages: vec![], + slot: 1, + magic: [0; 4], + ring_size: 10, + raw_messages: vec![], }, wormhole_merkle_state: WormholeMerkleState { vaa: vec![], diff --git a/hermes/src/store/storage/local_storage.rs b/hermes/src/store/storage/local_storage.rs index 518168b7..b6cbc09d 100644 --- a/hermes/src/store/storage/local_storage.rs +++ b/hermes/src/store/storage/local_storage.rs @@ -211,6 +211,7 @@ mod test { ) -> MessageState { MessageState { slot, + raw_message: vec![], message: Message::PriceFeedMessage(PriceFeedMessage { feed_id, publish_time, @@ -574,10 +575,10 @@ mod test { // Change the state to have accumulator messages // We mutate the existing state because the normal flow is like this. accumulator_state.accumulator_messages = Some(AccumulatorMessages { - magic: [0; 4], - slot: 10, - ring_size: 3, - messages: vec![], + magic: [0; 4], + slot: 10, + ring_size: 3, + raw_messages: vec![], }); // Store the accumulator state again. diff --git a/hermes/src/store/types.rs b/hermes/src/store/types.rs index a73cf27c..fe98f236 100644 --- a/hermes/src/store/types.rs +++ b/hermes/src/store/types.rs @@ -1,9 +1,7 @@ use { super::proof::wormhole_merkle::WormholeMerkleMessageProof, - pythnet_sdk::messages::{ - Message, - PriceFeedMessage, - }, + borsh::BorshDeserialize, + pythnet_sdk::messages::PriceFeedMessage, }; #[derive(Clone, PartialEq, Debug)] @@ -20,12 +18,20 @@ pub enum RequestTime { FirstAfter(UnixTimestamp), } -#[derive(Clone, PartialEq, Debug)] +pub type RawMessage = Vec; + +/// Accumulator messages coming from Pythnet validators. +/// +/// The validators writes the accumulator messages using Borsh with +/// the following struct. We cannot directly have messages as Vec +/// because they are serialized using big-endian byte order and Borsh +/// uses little-endian byte order. +#[derive(Clone, PartialEq, Debug, BorshDeserialize)] pub struct AccumulatorMessages { - pub magic: [u8; 4], - pub slot: Slot, - pub ring_size: u32, - pub messages: Vec, + pub magic: [u8; 4], + pub slot: u64, + pub ring_size: u32, + pub raw_messages: Vec, } impl AccumulatorMessages {