[hermes] add /v2/price_feeds endpoint (#1277)

* initial stab

* fix comments

* add filter feature

* fix deprecated warnings

* use cache

* Update price_feeds API query

* Update PriceFeedsQueryParams struct in price_feeds.rs

* fix merge conflict

* fix default value

* add tracing info

* fix comment

* address comments

* change var name

* refactor

* refactor

* refactor

* refactor

* undo changes in cache.rs

* undo changes in aggregate.rs

* address comments

* address comments

* address comments and improve fetching data speed

* address comments

* address comments

* bump

* change chunk size

* change function name

* address comment

* address comments

* address comments

* address comments

* Remove debug print statement

* address comments and add to openapi

Author: cctdaniel

hermes/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name        = "hermes"
-version     = "0.5.2"
+version     = "0.5.3"
 description = "Hermes is an agent that provides Verified Prices from the Pythnet Pyth Oracle."
 edition     = "2021"
 
@@ -30,6 +30,7 @@ nonzero_ext        = { version = "0.3.0" }
 prometheus-client  = { version = "0.21.2" }
 prost              = { version = "0.12.1" }
 pyth-sdk           = { version = "0.8.0" }
+pyth-sdk-solana    = { version = "0.9.0" }
 pythnet-sdk        = { path = "../pythnet/pythnet_sdk/", version = "2.0.0", features = ["strum"] }
 rand               = { version = "0.8.5" }
 reqwest            = { version = "0.11.14", features = ["blocking", "json"] }
@@ -50,9 +51,9 @@ utoipa-swagger-ui  = { version = "3.1.4", features = ["axum"] }
 wormhole-sdk       = { git     = "https://github.com/wormhole-foundation/wormhole", tag = "v2.17.1" }
 
 # We are bound to this Solana version in order to match pyth-oracle.
-solana-client          = { version = "=1.13.3" }
-solana-sdk             = { version = "=1.13.3" }
-solana-account-decoder = { version = "=1.13.3" }
+solana-client          = { version = "=1.16.19" }
+solana-sdk             = { version = "=1.16.19" }
+solana-account-decoder = { version = "=1.16.19" }
 
 
 [build-dependencies]

hermes/Cargo.toml

@@ -431,6 +431,7 @@ where
 
 pub async fn is_ready(state: &State) -> bool {
     let metadata = state.aggregate_state.read().await;
+    let price_feeds_metadata = state.price_feeds_metadata.read().await;
 
     let has_completed_recently = match metadata.latest_completed_update_at.as_ref() {
         Some(latest_completed_update_time) => {
@@ -449,7 +450,9 @@ pub async fn is_ready(state: &State) -> bool {
         _ => false,
     };
 
-    has_completed_recently && is_not_behind
+    let is_metadata_loaded = !price_feeds_metadata.is_empty();
+
+    has_completed_recently && is_not_behind && is_metadata_loaded
 }
 
 #[cfg(test)]

hermes/src/aggregate.rs

@@ -123,6 +123,7 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
             rest::price_feed_ids,
             rest::latest_price_updates,
             rest::timestamp_price_updates,
+            rest::price_feeds_metadata,
         ),
         components(
             schemas(
@@ -139,6 +140,8 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
                 types::BinaryPriceUpdate,
                 types::ParsedPriceUpdate,
                 types::RpcPriceFeedMetadataV2,
+                types::PriceFeedMetadata,
+                types::AssetType
             )
         ),
         tags(
@@ -164,6 +167,7 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
             "/v2/updates/price/:publish_time",
             get(rest::timestamp_price_updates),
         )
+        .route("/v2/price_feeds", get(rest::price_feeds_metadata))
         .route("/live", get(rest::live))
         .route("/ready", get(rest::ready))
         .route("/ws", get(ws::ws_route_handler))

hermes/src/api.rs

@@ -33,6 +33,7 @@ pub use {
     ready::*,
     v2::{
         latest_price_updates::*,
+        price_feeds_metadata::*,
         timestamp_price_updates::*,
     },
 };
@@ -43,6 +44,7 @@ pub enum RestError {
     CcipUpdateDataNotFound,
     InvalidCCIPInput,
     PriceIdsNotFound { missing_ids: Vec<PriceIdentifier> },
+    RpcConnectionError { message: String },
 }
 
 impl IntoResponse for RestError {
@@ -80,6 +82,9 @@ impl IntoResponse for RestError {
                 )
                     .into_response()
             }
+            RestError::RpcConnectionError { message } => {
+                (StatusCode::INTERNAL_SERVER_ERROR, message).into_response()
+            }
         }
     }
 }

hermes/src/api/rest.rs

@@ -18,5 +18,6 @@ pub async fn index() -> impl IntoResponse {
         "/api/get_vaa_ccip?data=<0x<price_feed_id_32_bytes>+<publish_time_unix_timestamp_be_8_bytes>>",
         "/v2/updates/price/latest?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..(&encoding=hex|base64)(&parsed=false)",
         "/v2/updates/price/<timestamp>?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..(&encoding=hex|base64)(&parsed=false)",
+        "/v2/price_feeds?(query=btc)(&asset_type=crypto|equity|fx|metal|rates)",
     ])
 }

hermes/src/api/rest/index.rs

@@ -1,2 +1,3 @@
 pub mod latest_price_updates;
+pub mod price_feeds_metadata;
 pub mod timestamp_price_updates;

hermes/src/api/rest/v2/mod.rs

@@ -0,0 +1,64 @@
+use {
+    crate::{
+        api::{
+            rest::RestError,
+            types::{
+                AssetType,
+                PriceFeedMetadata,
+            },
+        },
+        price_feeds_metadata::get_price_feeds_metadata,
+    },
+    anyhow::Result,
+    axum::{
+        extract::State,
+        Json,
+    },
+    serde::Deserialize,
+    serde_qs::axum::QsQuery,
+    utoipa::IntoParams,
+};
+
+
+#[derive(Debug, Deserialize, IntoParams)]
+#[into_params(parameter_in=Query)]
+pub struct PriceFeedsMetadataQueryParams {
+    /// Optional query parameter. If provided, the results will be filtered to all price feeds whose symbol contains the query string. Query string is case insensitive.
+    #[param(example = "bitcoin")]
+    query: Option<String>,
+
+    /// Optional query parameter. If provided, the results will be filtered by asset type. Possible values are crypto, equity, fx, metal, rates. Filter string is case insensitive.
+    #[param(example = "crypto")]
+    asset_type: Option<AssetType>,
+}
+
+/// Get the set of price feeds.
+///
+/// This endpoint fetches all price feeds from the Pyth network. It can be filtered by asset type
+/// and query string.
+#[utoipa::path(
+    get,
+    path = "/v2/price_feeds",
+    responses(
+        (status = 200, description = "Price feeds metadata retrieved successfully", body = Vec<RpcPriceIdentifier>)
+    ),
+    params(
+        PriceFeedsMetadataQueryParams
+    )
+)]
+pub async fn price_feeds_metadata(
+    State(state): State<crate::api::ApiState>,
+    QsQuery(params): QsQuery<PriceFeedsMetadataQueryParams>,
+) -> Result<Json<Vec<PriceFeedMetadata>>, RestError> {
+    let price_feeds_metadata =
+        get_price_feeds_metadata(&*state.state, params.query, params.asset_type)
+            .await
+            .map_err(|e| {
+                tracing::warn!("RPC connection error: {}", e);
+                RestError::RpcConnectionError {
+                    message: format!("RPC connection error: {}", e),
+                }
+            })?;
+
+    Ok(Json(price_feeds_metadata))
+}

hermes/src/api/rest/v2/price_feeds_metadata.rs

@@ -28,6 +28,14 @@ use {
         Deserialize,
         Serialize,
     },
+    std::{
+        collections::BTreeMap,
+        fmt::{
+            Display,
+            Formatter,
+            Result as FmtResult,
+        },
+    },
     utoipa::ToSchema,
     wormhole_sdk::Chain,
 };
@@ -52,7 +60,7 @@ impl From<PriceIdInput> for PriceIdentifier {
 
 type Base64String = String;
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct RpcPriceFeedMetadata {
     #[schema(value_type = Option<u64>, example=85480034)]
     pub slot:                       Option<Slot>,
@@ -64,7 +72,7 @@ pub struct RpcPriceFeedMetadata {
     pub prev_publish_time:          Option<UnixTimestamp>,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct RpcPriceFeedMetadataV2 {
     #[schema(value_type = Option<u64>, example=85480034)]
     pub slot:                 Option<Slot>,
@@ -74,7 +82,7 @@ pub struct RpcPriceFeedMetadataV2 {
     pub prev_publish_time:    Option<UnixTimestamp>,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct RpcPriceFeed {
     pub id:        RpcPriceIdentifier,
     pub price:     RpcPrice,
@@ -142,8 +150,8 @@ impl RpcPriceFeed {
     Eq,
     BorshSerialize,
     BorshDeserialize,
-    serde::Serialize,
-    serde::Deserialize,
+    Serialize,
+    Deserialize,
     ToSchema,
 )]
 pub struct RpcPrice {
@@ -178,8 +186,8 @@ pub struct RpcPrice {
     Hash,
     BorshSerialize,
     BorshDeserialize,
-    serde::Serialize,
-    serde::Deserialize,
+    Serialize,
+    Deserialize,
     ToSchema,
 )]
 #[repr(C)]
@@ -204,7 +212,7 @@ impl From<PriceIdentifier> for RpcPriceIdentifier {
     }
 }
 
-#[derive(Clone, Copy, Debug, Default, serde::Deserialize, serde::Serialize, ToSchema)]
+#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, ToSchema)]
 pub enum EncodingType {
     #[default]
     #[serde(rename = "hex")]
@@ -222,13 +230,13 @@ impl EncodingType {
     }
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct BinaryPriceUpdate {
     pub encoding: EncodingType,
     pub data:     Vec<String>,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct ParsedPriceUpdate {
     pub id:        RpcPriceIdentifier,
     pub price:     RpcPrice,
@@ -263,7 +271,7 @@ impl From<PriceFeedUpdate> for ParsedPriceUpdate {
     }
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema)]
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct PriceUpdate {
     pub binary: BinaryPriceUpdate,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -316,3 +324,27 @@ impl TryFrom<PriceUpdate> for PriceFeedsWithUpdateData {
         })
     }
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct PriceFeedMetadata {
+    pub id:         PriceIdentifier,
+    // BTreeMap is used to automatically sort the keys to ensure consistent ordering of attributes in the JSON response.
+    // This enhances user experience by providing a predictable structure, avoiding confusion from varying orders in different responses.
+    pub attributes: BTreeMap<String, String>,
+}
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, ToSchema)]
+#[serde(rename_all = "lowercase")]
+pub enum AssetType {
+    Crypto,
+    FX,
+    Equity,
+    Metals,
+    Rates,
+}
+
+impl Display for AssetType {
+    fn fmt(&self, f: &mut Formatter) -> FmtResult {
+        write!(f, "{:?}", self)
+    }
+}

hermes/src/api/types.rs

@@ -1,4 +1,9 @@
-use clap::Args;
+use {
+    clap::Args,
+    solana_sdk::pubkey::Pubkey,
+};
+
+const DEFAULT_PYTHNET_MAPPING_ADDR: &str = "AHtgzX45WTKfkPG53L6WYhGEXwQkN1BVknET3sVsLL8J";
 
 #[derive(Args, Clone, Debug)]
 #[command(next_help_heading = "Pythnet Options")]
@@ -13,4 +18,10 @@ pub struct Options {
     #[arg(long = "pythnet-http-addr")]
     #[arg(env = "PYTHNET_HTTP_ADDR")]
     pub http_addr: String,
+
+    /// Pyth mapping account address.
+    #[arg(long = "mapping-address")]
+    #[arg(default_value = DEFAULT_PYTHNET_MAPPING_ADDR)]
+    #[arg(env = "MAPPING_ADDRESS")]
+    pub mapping_addr: Pubkey,
 }