Terra contract public api (#79)

* Use pyth-sdk in terra contract
* Update terra contract according to agreed API
- Also adds v2 suffix to price_info key because this migration is breaking.

terra/Cargo.lock

@@ -1433,6 +1433,7 @@ dependencies = [
  "k256",
  "lazy_static",
  "p2w-sdk",
+ "pyth-sdk",
  "schemars",
  "serde",
  "serde_derive",

terra/contracts/pyth-bridge/Cargo.toml

@@ -33,6 +33,7 @@ lazy_static = "1.4.0"
 bigint = "4"
 p2w-sdk = { path = "../../../third_party/pyth/p2w-sdk/rust" }
 solana-program = "=1.8.16"
+pyth-sdk = { version = "0.2.0" }
 
 [dev-dependencies]
 cosmwasm-vm = { version = "0.16.0", default-features = false }

terra/contracts/pyth-bridge/src/contract.rs

@@ -8,9 +8,14 @@ use cosmwasm_std::{
     MessageInfo,
     QueryRequest,
     Response,
-    StdError,
     StdResult,
     WasmQuery,
+    Timestamp,
+};
+
+use pyth_sdk::{
+    PriceFeed,
+    PriceStatus,
 };
 
 use crate::{
@@ -19,21 +24,21 @@ use crate::{
         InstantiateMsg,
         MigrateMsg,
         QueryMsg,
+        PriceFeedResponse,
     },
     state::{
         config,
         config_read,
         price_info,
         price_info_read,
-        sequence,
-        sequence_read,
         ConfigInfo,
+        PriceInfo,
+        VALID_TIME_PERIOD,
     },
 };
 
 use p2w_sdk::{
     BatchPriceAttestation,
-    PriceAttestation,
 };
 
 use wormhole::{
@@ -57,14 +62,13 @@ pub fn instantiate(
     _info: MessageInfo,
     msg: InstantiateMsg,
 ) -> StdResult<Response> {
-    // Save general wormhole info
+    // Save general wormhole and pyth info
     let state = ConfigInfo {
         wormhole_contract: msg.wormhole_contract,
         pyth_emitter: msg.pyth_emitter.as_slice().to_vec(),
         pyth_emitter_chain: msg.pyth_emitter_chain,
     };
     config(deps.storage).save(&state)?;
-    sequence(deps.storage).save(&0)?;
 
     Ok(Response::default())
 }
@@ -97,57 +101,118 @@ fn submit_vaa(
     let state = config_read(deps.storage).load()?;
 
     let vaa = parse_vaa(deps.branch(), env.block.time.seconds(), data)?;
-    let data = vaa.payload;
-
-    // IMPORTANT: VAA replay-protection is not implemented in this code-path
-    // Sequences are used to prevent replay or price rollbacks
-
-    let message = BatchPriceAttestation::deserialize(&data[..])
-        .map_err(|_| ContractError::InvalidVAA.std())?;
+ 
+    // This checks the emitter to be the pyth emitter in wormhole and it comes from emitter chain (Solana)
     if vaa.emitter_address != state.pyth_emitter || vaa.emitter_chain != state.pyth_emitter_chain {
         return ContractError::InvalidVAA.std_err();
     }
 
-    // Check sequence
-    let last_sequence = sequence_read(deps.storage).load()?;
-    if vaa.sequence <= last_sequence && last_sequence != 0 {
-        return Err(StdError::generic_err(
-            "price sequences need to be monotonically increasing",
-        ));
-    }
-    sequence(deps.storage).save(&vaa.sequence)?;
+    let data = vaa.payload;
+
+    let message = BatchPriceAttestation::deserialize(&data[..])
+        .map_err(|_| ContractError::InvalidVAA.std())?;
+    
+    let mut new_attestations_cnt: u8 = 0;
 
-    // Update price
+    // Update prices
     for price_attestation in message.price_attestations.iter() {
-        price_info(deps.storage).save(
-            &price_attestation.price_id.to_bytes()[..],
-            &price_attestation.serialize(),
-        )?;
+        let price_feed = PriceFeed::new(
+                price_attestation.price_id.to_bytes(),
+                price_attestation.status,
+                price_attestation.expo,
+                0, // max_num_publishers data is currently unavailable
+                0, // num_publishers data is currently unavailable
+                price_attestation.product_id.to_bytes(),
+                price_attestation.price,
+            price_attestation.confidence_interval,
+            price_attestation.ema_price.val,
+            price_attestation.ema_conf.val as u64,
+        );
+
+        let attestation_time = Timestamp::from_seconds(price_attestation.timestamp as u64);
+
+        price_info(deps.storage).update(
+            price_feed.id.as_ref(),
+        |maybe_price_info| -> StdResult<PriceInfo> {
+            match maybe_price_info {
+                Some(price_info) => {
+                    // This check ensures that a price won't be updated with the same or older message.
+                    // Attestation_time is guaranteed increasing in solana
+                    if price_info.attestation_time < attestation_time {
+                        new_attestations_cnt += 1;
+                        Ok(PriceInfo {
+                            arrival_time: env.block.time,
+                            arrival_block: env.block.height,
+                            price_feed,
+                            attestation_time
+                        })
+                    } else {
+                        Ok(price_info)
+                    }
+                },
+                None => {
+                    new_attestations_cnt += 1;
+                    Ok(PriceInfo {
+                        arrival_time: env.block.time,
+                        arrival_block: env.block.height,
+                        price_feed,
+                        attestation_time
+                    })
+                }
+            }
+        })?;
     }
 
     Ok(Response::new()
         .add_attribute("action", "price_update")
         .add_attribute(
-            "num_price_feeds",
+            "batch_size",
             format!("{}", message.price_attestations.len()),
-        ))
+        )
+        .add_attribute(
+            "num_updates",
+            format!("{}", new_attestations_cnt),
+        )
+    )
 }
 
 
 #[cfg_attr(not(feature = "library"), entry_point)]
-pub fn query(deps: Deps, _env: Env, msg: QueryMsg) -> StdResult<Binary> {
+pub fn query(deps: Deps, env: Env, msg: QueryMsg) -> StdResult<Binary> {
     match msg {
-        QueryMsg::PriceInfo { price_id } => {
-            to_binary(&query_price_info(deps, price_id.as_slice())?)
+        QueryMsg::PriceFeed { id } => {
+            to_binary(&query_price_info(deps, env, id.as_ref())?)
         }
     }
 }
 
-pub fn query_price_info(deps: Deps, address: &[u8]) -> StdResult<PriceAttestation> {
+pub fn query_price_info(deps: Deps, env: Env, address: &[u8]) -> StdResult<PriceFeedResponse> {
     match price_info_read(deps.storage).load(address) {
-        Ok(data) => PriceAttestation::deserialize(&data[..]).map_err(|_| {
-            StdError::parse_err("PriceAttestation", "failed to decode price attestation")
-        }),
+        Ok(mut terra_price_info) => {
+            // Attestation time is very close to the actual price time (maybe a few seconds older).
+            // Cases that it will cover:
+            // - This will ensure to set status unknown if the price has become very old and hasn't updated yet.
+            // - If a price has arrived very late to terra it will set the status to unknown.
+            // - If a price is coming from future it's tolerated up to VALID_TIME_PERIOD seconds (using abs diff)
+            //   but more than that is set to unknown, the reason is huge clock difference means there exists a 
+            //   problem in a either Terra or Solana blockchain and if it is Solana we don't want to propagate
+            //   Solana internal problems to Terra
+            let time_abs_diff = if env.block.time.seconds() > terra_price_info.attestation_time.seconds() {
+                env.block.time.seconds() - terra_price_info.attestation_time.seconds()
+            } else {
+                terra_price_info.attestation_time.seconds() - env.block.time.seconds()
+            };
+
+            if time_abs_diff > VALID_TIME_PERIOD.as_secs() {
+                terra_price_info.price_feed.status = PriceStatus::Unknown;
+            }
+
+            Ok(
+                PriceFeedResponse {
+                    price_feed: terra_price_info.price_feed,
+                }
+            )
+        },
         Err(_) => ContractError::AssetNotFound.std_err(),
     }
 }

terra/contracts/pyth-bridge/src/msg.rs

@@ -1,4 +1,7 @@
-use cosmwasm_std::Binary;
+use cosmwasm_std::{
+    Binary,
+};
+use pyth_sdk::{PriceFeed, PriceIdentifier};
 use schemars::JsonSchema;
 use serde::{
     Deserialize,
@@ -27,5 +30,12 @@ pub struct MigrateMsg {}
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum QueryMsg {
-    PriceInfo { price_id: Binary },
+    PriceFeed { id: PriceIdentifier },
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub struct PriceFeedResponse {
+    /// Pyth Price Feed
+    pub price_feed:        PriceFeed,
 }

terra/contracts/pyth-bridge/src/state.rs

@@ -1,3 +1,6 @@
+use std::time::Duration;
+
+use pyth_sdk::PriceFeed;
 use schemars::JsonSchema;
 use serde::{
     Deserialize,
@@ -7,7 +10,9 @@ use serde::{
 use cosmwasm_std::{
     StdResult,
     Storage,
+    Timestamp,
 };
+
 use cosmwasm_storage::{
     bucket,
     bucket_read,
@@ -24,8 +29,12 @@ use wormhole::byte_utils::ByteUtils;
 type HumanAddr = String;
 
 pub static CONFIG_KEY: &[u8] = b"config";
-pub static PRICE_INFO_KEY: &[u8] = b"price_info";
-pub static SEQUENCE_KEY: &[u8] = b"sequence";
+pub static PRICE_INFO_KEY: &[u8] = b"price_info_v2";
+
+/// Maximum acceptable time period before price is considered to be stale.
+/// 
+/// This value considers attestation delay which currently might up to a minute.
+pub const VALID_TIME_PERIOD: Duration = Duration::from_secs(3*60);
 
 // Guardian set information
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
@@ -35,6 +44,16 @@ pub struct ConfigInfo {
     pub pyth_emitter_chain: u16,
 }
 
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub struct PriceInfo {
+    pub arrival_time:         Timestamp,
+    pub arrival_block:        u64,
+    pub attestation_time:     Timestamp,
+    pub price_feed:           PriceFeed,
+}
+
 pub fn config(storage: &mut dyn Storage) -> Singleton<ConfigInfo> {
     singleton(storage, CONFIG_KEY)
 }
@@ -43,19 +62,11 @@ pub fn config_read(storage: &dyn Storage) -> ReadonlySingleton<ConfigInfo> {
     singleton_read(storage, CONFIG_KEY)
 }
 
-pub fn sequence(storage: &mut dyn Storage) -> Singleton<u64> {
-    singleton(storage, SEQUENCE_KEY)
-}
-
-pub fn sequence_read(storage: &dyn Storage) -> ReadonlySingleton<u64> {
-    singleton_read(storage, SEQUENCE_KEY)
-}
-
-pub fn price_info(storage: &mut dyn Storage) -> Bucket<Vec<u8>> {
+pub fn price_info(storage: &mut dyn Storage) -> Bucket<PriceInfo> {
     bucket(storage, PRICE_INFO_KEY)
 }
 
-pub fn price_info_read(storage: &dyn Storage) -> ReadonlyBucket<Vec<u8>> {
+pub fn price_info_read(storage: &dyn Storage) -> ReadonlyBucket<PriceInfo> {
     bucket_read(storage, PRICE_INFO_KEY)
 }