Use slot for pending accumulations

Author: ali-behjati

hermes/src/store.rs

@@ -10,6 +10,7 @@ use {
             MessageType,
             PriceFeedsWithUpdateData,
             RequestTime,
+            Slot,
             Update,
         },
     },
@@ -32,19 +33,14 @@ use {
     derive_builder::Builder,
     moka::future::Cache,
     pyth_sdk::PriceIdentifier,
-    std::{
-        ops::Rem,
-        time::Duration,
-    },
+    std::time::Duration,
     wormhole_sdk::Vaa,
 };
 
 pub mod proof;
 pub mod storage;
 pub mod types;
 
-pub type RingIndex = u32;
-
 #[derive(Clone, PartialEq, Debug, Builder)]
 #[builder(derive(Debug), pattern = "immutable")]
 pub struct AccumulatorState {
@@ -55,7 +51,7 @@ pub struct AccumulatorState {
 #[derive(Clone)]
 pub struct Store {
     pub storage:               StorageInstance,
-    pub pending_accumulations: Cache<RingIndex, AccumulatorStateBuilder>,
+    pub pending_accumulations: Cache<Slot, AccumulatorStateBuilder>,
 }
 
 impl Store {
@@ -73,7 +69,7 @@ impl Store {
 
     /// Stores the update data in the store
     pub async fn store_update(&self, update: Update) -> Result<()> {
-        let ring_index = match update {
+        let slot = match update {
             Update::Vaa(vaa_bytes) => {
                 let vaa = serde_wormhole::from_slice::<Vaa<Vec<u8>>>(&vaa_bytes)?;
                 let payload = WormholePayload::try_from_bytes(&vaa.payload, &vaa_bytes)?;
@@ -83,41 +79,42 @@ impl Store {
 
                 match payload {
                     WormholePayload::Merkle(proof) => {
-                        log::info!("Storing merkle proof for state index {:?}", proof);
+                        log::info!(
+                            "Storing merkle proof for slot {:?}: {:?}",
+                            proof.slot,
+                            proof
+                        );
                         store_wormhole_merkle_verified_message(self, proof.clone()).await?;
-                        proof.state_index
+                        proof.slot
                     }
                 }
             }
             Update::AccumulatorMessages(accumulator_messages) => {
                 // FIXME: Move this constant to a better place
-                const RING_SIZE: u32 = 10_000;
-                let ring_index = accumulator_messages.slot.rem(RING_SIZE as u64) as u32;
+
+                let slot = accumulator_messages.slot;
 
                 log::info!(
-                    "Storing accumulator messages for ring index {:?}: {:?}",
-                    ring_index,
+                    "Storing accumulator messages for slot {:?}: {:?}",
+                    slot,
                     accumulator_messages
                 );
 
                 let pending_acc = self
                     .pending_accumulations
-                    .entry(ring_index)
+                    .entry(slot)
                     .or_default()
                     .await
                     .into_value();
                 self.pending_accumulations
-                    .insert(
-                        ring_index,
-                        pending_acc.accumulator_messages(accumulator_messages),
-                    )
+                    .insert(slot, pending_acc.accumulator_messages(accumulator_messages))
                     .await;
 
-                ring_index
+                slot
             }
         };
 
-        let pending_state = self.pending_accumulations.get(&ring_index);
+        let pending_state = self.pending_accumulations.get(&slot);
         let pending_state = match pending_state {
             Some(pending_state) => pending_state,
             // Due to some race conditions this might happen when it's processed before
@@ -159,7 +156,7 @@ impl Store {
 
         self.storage.store_message_states(message_states)?;
 
-        self.pending_accumulations.invalidate(&ring_index).await;
+        self.pending_accumulations.invalidate(&slot).await;
 
         Ok(())
     }

hermes/src/store/proof/wormhole_merkle.rs

@@ -32,9 +32,10 @@ type Hash = [u8; 20];
 
 #[derive(Clone, PartialEq, Debug)]
 pub struct WormholeMerkleProof {
-    pub vaa:         Vec<u8>,
-    pub state_index: u32,
-    pub root:        Hash,
+    pub vaa:       Vec<u8>,
+    pub slot:      u64,
+    pub ring_size: u32,
+    pub root:      Hash,
 }
 
 #[derive(Clone, PartialEq, Debug)]
@@ -49,13 +50,13 @@ pub async fn store_wormhole_merkle_verified_message(
 ) -> Result<()> {
     let pending_acc = store
         .pending_accumulations
-        .entry(proof.state_index)
+        .entry(proof.slot)
         .or_default()
         .await
         .into_value();
     store
         .pending_accumulations
-        .insert(proof.state_index, pending_acc.wormhole_merkle_proof(proof))
+        .insert(proof.slot, pending_acc.wormhole_merkle_proof(proof))
         .await;
     Ok(())
 }

hermes/src/store/types.rs

@@ -16,6 +16,11 @@ pub enum WormholePayload {
 
 impl WormholePayload {
     pub fn try_from_bytes(bytes: &[u8], vaa_bytes: &[u8]) -> Result<Self> {
+        if bytes.len() != 37 {
+            return Err(anyhow!("Invalid message length"));
+        }
+
+        // TODO: Use byte string literals for this check
         let magic = u32::from_be_bytes(bytes[0..4].try_into()?);
         if magic != 0x41555756u32 {
             return Err(anyhow!("Invalid magic"));
@@ -27,16 +32,15 @@ impl WormholePayload {
             return Err(anyhow!("Invalid message type"));
         }
 
-        let state_index = u32::from_be_bytes(bytes[5..9].try_into()?);
-        let root_digest = bytes[9..29].try_into()?;
+        let slot = u64::from_be_bytes(bytes[5..13].try_into()?);
+        let ring_size = u32::from_be_bytes(bytes[13..17].try_into()?);
+        let root_digest = bytes[17..37].try_into()?;
 
-        if bytes.len() > 29 {
-            return Err(anyhow!("Invalid message length"));
-        }
 
         Ok(Self::Merkle(WormholeMerkleProof {
             root: root_digest,
-            state_index,
+            slot,
+            ring_size,
             vaa: vaa_bytes.to_vec(),
         }))
     }