Relayer performs Pyth updates on demand

common/src/oracle/pyth.rs

@@ -26,7 +26,7 @@ use near_sdk::{
 
 pub type OracleResponse = HashMap<PriceIdentifier, Option<Price>>;
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 #[near(serializers = [borsh, json])]
 pub struct PriceIdentifier(
     #[serde(
@@ -36,6 +36,12 @@ pub struct PriceIdentifier(
     pub [u8; 32],
 );
 
+impl std::fmt::Debug for PriceIdentifier {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", hex::encode(self.0))
+    }
+}
+
 impl Display for PriceIdentifier {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", hex::encode(self.0))

contract/universal-account/examples/02_payload.rs

@@ -15,7 +15,9 @@ pub fn main() {
             index: U64(0),
             nonce: U64(1),
         },
-        account_id: "my-universal-account.testnet".parse().unwrap(),
+        account_id: "default-18843764340.gh-275.templar-in-training.testnet"
+            .parse()
+            .unwrap(),
         payload: vec![Transaction {
             receiver_id: "alice.testnet".parse().unwrap(),
             actions: vec![Action::Transfer {

service/relayer/Cargo.toml

@@ -37,6 +37,7 @@ tower.workspace = true
 tower-http.workspace = true
 tracing-subscriber = { workspace = true, features = ["env-filter", "json"] }
 tracing = { workspace = true, features = ["attributes"] }
+reqwest = {version = "0.12.24", features = ["json"]}
 
 [dev-dependencies]
 near-workspaces = { workspace = true, features = ["experimental"] }

service/relayer/src/app/args.rs

@@ -23,6 +23,8 @@ pub struct Configuration {
     #[clap(flatten)]
     pub ua: UniversalAccount,
     #[clap(flatten)]
+    pub pyth: Pyth,
+    #[clap(flatten)]
     pub cache: Cache,
     /// Broom batch size.
     #[arg(long, env = "BROOM_BATCH_SIZE", default_value_t = 16)]
@@ -36,6 +38,51 @@ fn duration_from_secs(s: &str) -> Result<Duration, std::num::ParseIntError> {
     Ok(Duration::from_secs(u64::from_str(s)?))
 }
 
+#[derive(Args, Debug, Clone)]
+pub struct Pyth {
+    /// Pyth Hermes API URL. See: <https://docs.pyth.network/price-feeds/core/api-reference>
+    #[arg(
+        long = "pyth-hermes-url",
+        env = "PYTH_HERMES_URL",
+        default_value_t = String::from("https://hermes-beta.pyth.network")
+    )]
+    pub hermes_url: String,
+    /// Do not push price updates to Pyth oracle if the last push was less
+    /// than this long ago, even if requested.
+    #[arg(
+        id = "pyth-refresh-secs",
+        long = "pyth-refresh-secs",
+        env = "PYTH_REFRESH_SECS",
+        value_parser = duration_from_secs,
+        default_value = "3"
+    )]
+    pub refresh: Duration,
+    /// Oracle ID to push price updates to.
+    #[arg(
+        id = "pyth-oracle-id",
+        long = "pyth-oracle-id",
+        env = "PYTH_ORACLE_ID",
+        default_value_t = AccountId::from_str("pyth-oracle.testnet").unwrap()
+    )]
+    pub oracle_id: AccountId,
+    /// How much gas (in units of Tgas) to attach to oracle price update calls.
+    #[arg(
+        id = "pyth-update-gas",
+        long = "pyth-update-gas",
+        env = "PYTH_UPDATE_GAS",
+        default_value = "300 Tgas"
+    )]
+    pub update_gas: near_sdk::Gas,
+    /// How much NEAR to attach as a deposit to oracle price update calls.
+    #[arg(
+        id = "pyth-update-deposit",
+        long = "pyth-update-deposit",
+        env = "PYTH_UPDATE_DEPOSIT",
+        default_value = "0.01 NEAR"
+    )]
+    pub update_deposit: NearToken,
+}
+
 #[derive(Args, Debug, Clone)]
 pub struct Cache {
     /// Refresh the cached gas price after X seconds.
@@ -56,15 +103,6 @@ pub struct Cache {
         default_value = "60"
     )]
     pub nonce_refresh: Duration,
-    /// Refresh the cached protocol configuration after X seconds.
-    #[arg(
-        id = "cache-protocol-config-secs",
-        long = "cache-protocol-config-secs",
-        env = "CACHE_PROTOCOL_CONFIG_SECS",
-        value_parser = duration_from_secs,
-        default_value = "3600"
-    )]
-    pub protocol_config_refresh: Duration,
 }
 
 #[derive(Args, Debug, Clone)]

service/relayer/src/app/mod.rs

@@ -29,9 +29,10 @@ use crate::{
             Database,
         },
         near::Near,
+        pyth::Pyth,
     },
-    error::PreconditionError,
-    AccountData, AssetTransfer, AssetTransferParseError, ContractData,
+    error::{FunctionCallRejectionReason, PayloadRejectionReason},
+    AccountData, AssetTransfer, ContractData,
 };
 
 pub mod args;
@@ -43,6 +44,7 @@ pub struct App {
     pub accounts: Arc<RwLock<AccountData>>,
     pub relay_near: Near,
     pub ua_near: Near,
+    pub pyth: Pyth,
     pub cache: Arc<Cache>,
     pub database: Database,
 }
@@ -74,6 +76,13 @@ impl App {
 
         let cache = Cache::new(relay_near.clone(), args.cache.clone(), kill.clone());
 
+        let pyth = Pyth::new(
+            args.pyth.clone(),
+            relay_near.clone(),
+            cache.clone(),
+            kill.clone(),
+        );
+
         tokio::spawn(broom::start(
             database.clone(),
             relay_near.clone(),
@@ -87,26 +96,27 @@ impl App {
             accounts: Arc::new(RwLock::new(AccountData::default())),
             relay_near,
             ua_near,
+            pyth,
             cache: Arc::new(cache),
             database,
         }
     }
 
     #[tracing::instrument(skip(self), fields(gas = %gas))]
-    pub async fn estimate_cost_of_gas(&self, gas: u64) -> Option<NearToken> {
+    pub async fn estimate_cost_of_gas(&self, gas: near_sdk::Gas) -> Option<NearToken> {
         const TERA: u128 = near_sdk::Gas::from_tgas(1).as_gas() as u128;
 
         let price_per_tgas = self.cache.gas_price().await;
         let result = price_per_tgas
-            .checked_mul(u128::from(gas))?
-            .checked_div(TERA);
+            .checked_mul(u128::from(gas.as_gas()))
+            .and_then(|x| x.checked_div(TERA));
 
         tracing::debug!(cost = ?result, "Estimated gas cost");
         result
     }
 
     #[allow(clippy::too_many_lines, reason = "procedural")]
-    #[tracing::instrument(skip(self), name = "load_markets")]
+    #[tracing::instrument(skip(self))]
     pub async fn load_markets(&mut self) {
         tracing::info!("Loading markets from registry and individual sources");
         let mut markets = self.args.monitor.market.clone();
@@ -245,80 +255,74 @@ impl App {
     ///
     /// - If the receiver is not known.
     /// - If any of the function call actions are not allowed.
-    #[tracing::instrument(skip(self, accounts, calls), fields(receiver_id = %receiver_id))]
+    #[tracing::instrument(skip(self, accounts, contract_data, calls))]
     pub fn actions_are_allowed<'a>(
         &self,
-        receiver_id: &AccountIdRef,
         accounts: &AccountData,
+        receiver_id: &AccountIdRef,
+        contract_data: &ContractData,
         calls: impl IntoIterator<Item = &'a FunctionCallAction>,
-    ) -> Result<Vec<AccountId>, PreconditionError> {
+    ) -> Result<Vec<AccountId>, Vec<FunctionCallRejectionReason>> {
         let mut other_interactions = Vec::new();
-
-        let Some(contract_data) = accounts.allowed_contract_data.get(receiver_id) else {
-            return Err(PreconditionError::UnknownTransactionReceiverId {
-                account_id: receiver_id.to_owned(),
-            });
-        };
+        let mut errors = vec![];
 
         for (index, call) in calls.into_iter().enumerate() {
             if !contract_data.allowed_methods.contains(&call.method_name) {
-                return Err(PreconditionError::UnknownFunctionName { index });
+                errors.push(FunctionCallRejectionReason::UnknownFunctionName {
+                    index,
+                    function_name: call.method_name.clone(),
+                });
             }
 
-            if let Ok(transfer) =
-                AssetTransfer::parse(receiver_id.to_owned(), call).map_err(|e| match e {
-                    AssetTransferParseError::UnknownFunctionName => {
-                        PreconditionError::UnknownFunctionName { index }
-                    }
-                    AssetTransferParseError::ArgumentDeserialization => {
-                        PreconditionError::ArgumentDeserializationFailure { index }
-                    }
-                })
-            {
+            if let Ok(transfer) = AssetTransfer::parse(receiver_id.to_owned(), call) {
                 let market_id = transfer.token_receiver_id();
                 other_interactions.push(market_id.to_owned());
 
                 let Some(market_account_ids) = accounts.market_data.get(market_id) else {
-                    return Err(PreconditionError::UnknownTransferReceiverId {
+                    errors.push(FunctionCallRejectionReason::UnknownTransferReceiverId {
                         account_id: market_id.to_owned(),
                         index,
                     });
+                    continue;
                 };
 
                 let msg = transfer.args.msg();
                 let Ok(msg) = serde_json::from_str::<DepositMsg>(msg) else {
-                    return Err(PreconditionError::MsgDeserializationFailure {
+                    errors.push(FunctionCallRejectionReason::MsgDeserializationFailure {
                         index,
                         msg: msg.to_string(),
                     });
+                    continue;
                 };
 
                 #[allow(clippy::unwrap_used, reason = "DepositMsg serialization is infallible")]
                 if transfer.asset() == market_account_ids.borrow_asset {
                     if !matches!(msg, DepositMsg::Supply | DepositMsg::Repay) {
-                        return Err(PreconditionError::InvalidMsgForAsset {
+                        errors.push(FunctionCallRejectionReason::InvalidMsgForAsset {
                             index,
                             expected: "\"Supply\" or \"Repay\"".to_string(),
                             actual: serde_json::to_string(&msg).unwrap(),
                         });
                     }
                 } else if transfer.asset() == market_account_ids.collateral_asset {
                     if !matches!(msg, DepositMsg::Collateralize) {
-                        return Err(PreconditionError::InvalidMsgForAsset {
+                        errors.push(FunctionCallRejectionReason::InvalidMsgForAsset {
                             index,
                             expected: "\"Collateralize\"".to_string(),
                             actual: serde_json::to_string(&msg).unwrap(),
                         });
                     }
                 } else {
-                    return Err(PreconditionError::UnknownTransactionReceiverId {
-                        account_id: receiver_id.to_owned(),
-                    });
+                    // Not a standard-compliant function call
                 }
             }
         }
 
-        Ok(other_interactions)
+        if errors.is_empty() {
+            Ok(other_interactions)
+        } else {
+            Err(errors)
+        }
     }
 
     /// Check and calculate gas for a signed delegate action.
@@ -335,20 +339,20 @@ impl App {
         sender_id = %signed_delegate_action.delegate_action.sender_id,
         receiver_id = %signed_delegate_action.delegate_action.receiver_id
     ))]
-    pub async fn check_and_calculate_gas(
+    pub async fn sda_check_and_calculate_gas(
         &self,
         signed_delegate_action: &SignedDelegateAction,
-    ) -> Result<(u64, ContractData), PreconditionError> {
+    ) -> Result<(near_sdk::Gas, ContractData), PayloadRejectionReason> {
         tracing::debug!("Checking and calculating gas for delegate action");
         if !signed_delegate_action.verify() {
-            return Err(PreconditionError::SignatureVerificationFailure);
+            return Err(PayloadRejectionReason::SignatureVerificationFailure);
         }
 
         let receiver_id = &signed_delegate_action.delegate_action.receiver_id;
         let accounts = self.accounts.read().await;
 
         let Some(contract_data) = accounts.allowed_contract_data.get(receiver_id).cloned() else {
-            return Err(PreconditionError::UnknownTransactionReceiverId {
+            return Err(PayloadRejectionReason::UnknownTransactionReceiverId {
                 account_id: receiver_id.clone(),
             });
         };
@@ -357,21 +361,28 @@ impl App {
         let len = actions.len();
         let calls = actions
             .into_iter()
-            .try_fold(Vec::with_capacity(len), |mut v, action| {
+            .enumerate()
+            .try_fold(Vec::with_capacity(len), |mut v, (i, action)| {
                 if let Action::FunctionCall(fc) = action {
                     v.push(fc);
                     Ok(v)
                 } else {
-                    Err((v.len(), action))
+                    Err(i)
                 }
             })
-            .map_err(|(index, action)| PreconditionError::UnsupportedAction { index, action })?;
+            .map_err(|index| PayloadRejectionReason::UnsupportedAction { index })?;
 
-        self.actions_are_allowed(receiver_id, &accounts, calls.iter().map(Borrow::borrow))?;
+        self.actions_are_allowed(
+            &accounts,
+            receiver_id,
+            &contract_data,
+            calls.iter().map(Borrow::borrow),
+        )
+        .map_err(PayloadRejectionReason::FunctionCallRejection)?;
 
         let gas_total = calls.iter().map(|call| call.gas).sum();
 
-        Ok((gas_total, contract_data))
+        Ok((near_sdk::Gas::from_gas(gas_total), contract_data))
     }
 
     /// # Errors

service/relayer/src/cache.rs

@@ -79,7 +79,7 @@ impl<T> CacheRecord<T> {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Cache {
     request: mpsc::Sender<CacheRequest>,
 }

service/relayer/src/client/mod.rs

@@ -1,2 +1,3 @@
 pub mod database;
 pub mod near;
+pub mod pyth;