feat: update hermes to support publisher stake caps

apps/hermes/server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hermes"
-version = "0.5.17"
+version = "0.6.0"
 description = "Hermes is an agent that provides Verified Prices from the Pythnet Pyth Oracle."
 edition = "2021"
 

apps/hermes/server/src/api.rs

@@ -111,6 +111,7 @@ where
  rest::latest_vaas,
  rest::price_feed_ids,
  rest::latest_price_updates,
+ rest::latest_publisher_stake_caps,
  rest::timestamp_price_updates,
  rest::price_feeds_metadata,
  rest::price_stream_sse_handler,
@@ -127,7 +128,7 @@ where
  types::RpcPriceIdentifier,
  types::EncodingType,
  types::PriceUpdate,
- types::BinaryPriceUpdate,
+ types::BinaryUpdate,
  types::ParsedPriceUpdate,
  types::RpcPriceFeedMetadataV2,
  types::PriceFeedMetadata,
@@ -158,6 +159,10 @@ where
  get(rest::price_stream_sse_handler),
  )
  .route("/v2/updates/price/latest", get(rest::latest_price_updates))
+ .route(
+ "/v2/updates/publisher_stake_caps/latest",
+ get(rest::latest_publisher_stake_caps),
+ )
  .route(
  "/v2/updates/price/:publish_time",
  get(rest::timestamp_price_updates),

apps/hermes/server/src/api/rest.rs

@@ -35,6 +35,7 @@ pub use {
  ready::*,
  v2::{
  latest_price_updates::*,
+ latest_publisher_stake_caps::*,
  price_feeds_metadata::*,
  sse::*,
  timestamp_price_updates::*,

apps/hermes/server/src/api/rest/v2/latest_price_updates.rs

@@ -6,7 +6,7 @@ use {
  RestError,
  },
  types::{
- BinaryPriceUpdate,
+ BinaryUpdate,
  EncodingType,
  ParsedPriceUpdate,
  PriceIdInput,
@@ -108,7 +108,7 @@ where
  EncodingType::Hex => hex::encode(data),
  })
  .collect();
- let binary_price_update = BinaryPriceUpdate {
+ let binary_price_update = BinaryUpdate {
  encoding: params.encoding,
  data: encoded_data,
  };

apps/hermes/server/src/api/rest/v2/latest_publisher_stake_caps.rs

@@ -0,0 +1,114 @@
+use {
+ crate::{
+ api::{
+ rest::RestError,
+ types::{
+ BinaryUpdate,
+ EncodingType,
+ ParsedPublisherStakeCapsUpdate,
+ },
+ ApiState,
+ },
+ state::Aggregates,
+ },
+ anyhow::Result,
+ axum::{
+ extract::State,
+ Json,
+ },
+ base64::{
+ engine::general_purpose::STANDARD as base64_standard_engine,
+ Engine as _,
+ },
+ serde::{
+ Deserialize,
+ Serialize,
+ },
+ serde_qs::axum::QsQuery,
+ utoipa::{
+ IntoParams,
+ ToSchema,
+ },
+};
+
+
+#[derive(Debug, Deserialize, IntoParams)]
+#[into_params(parameter_in=Query)]
+pub struct LatestPublisherStakeCapsUpdateData {
+ /// Get the most recent publisher stake caps update data.
+
+ /// Optional encoding type. If true, return the message in the encoding specified by the encoding parameter. Default is `hex`.
+ #[serde(default)]
+ encoding: EncodingType,
+
+ /// If true, include the parsed update in the `parsed` field of each returned feed. Default is `true`.
+ #[serde(default = "default_true")]
+ parsed: bool,
+}
+
+fn default_true() -> bool {
+ true
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct LatestPublisherStakeCapsUpdateDataResponse {
+ pub binary: BinaryUpdate,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub parsed: Option<Vec<ParsedPublisherStakeCapsUpdate>>,
+}
+
+/// Get the most recent publisher stake caps update data.
+#[utoipa::path(
+ get,
+ path = "/v2/updates/publisher_stake_caps/latest",
+ responses(
+ (status = 200, description = "Publisher stake caps update data retrieved succesfully", body = Vec<PriceFeedMetadata>)
+ ),
+ params(
+ LatestPublisherStakeCapsUpdateData
+ )
+)]
+pub async fn latest_publisher_stake_caps<S>(
+ State(state): State<ApiState<S>>,
+ QsQuery(params): QsQuery<LatestPublisherStakeCapsUpdateData>,
+) -> Result<Json<LatestPublisherStakeCapsUpdateDataResponse>, RestError>
+where
+ S: Aggregates,
+{
+ let state = &*state.state;
+ let publisher_stake_caps_with_update_data =
+ Aggregates::get_latest_publisher_stake_caps_with_update_data(state)
+ .await
+ .map_err(|e| {
+ tracing::warn!(
+ "Error getting publisher stake caps with update data: {:?}",
+ e
+ );
+ RestError::UpdateDataNotFound
+ })?;
+
+ let encoded_data: Vec<String> = publisher_stake_caps_with_update_data
+ .update_data
+ .into_iter()
+ .map(|data| match params.encoding {
+ EncodingType::Base64 => base64_standard_engine.encode(data),
+ EncodingType::Hex => hex::encode(data),
+ })
+ .collect();
+
+ let binary = BinaryUpdate {
+ encoding: params.encoding,
+ data: encoded_data,
+ };
+
+ let parsed: Option<Vec<ParsedPublisherStakeCapsUpdate>> = if params.parsed {
+ Some(publisher_stake_caps_with_update_data.publisher_stake_caps)
+ } else {
+ None
+ };
+
+ Ok(Json(LatestPublisherStakeCapsUpdateDataResponse {
+ binary,
+ parsed,
+ }))
+}

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

@@ -1,4 +1,5 @@
 pub mod latest_price_updates;
+pub mod latest_publisher_stake_caps;
 pub mod price_feeds_metadata;
 pub mod sse;
 pub mod timestamp_price_updates;

apps/hermes/server/src/api/rest/v2/sse.rs

@@ -6,7 +6,7 @@ use {
  RestError,
  },
  types::{
- BinaryPriceUpdate,
+ BinaryUpdate,
  EncodingType,
  ParsedPriceUpdate,
  PriceIdInput,
@@ -211,7 +211,7 @@ where
  .into_iter()
  .map(|data| encoding.encode_str(&data))
  .collect();
- let binary_price_update = BinaryPriceUpdate {
+ let binary_price_update = BinaryUpdate {
  encoding,
  data: encoded_data,
  };

apps/hermes/server/src/api/rest/v2/timestamp_price_updates.rs

@@ -7,7 +7,7 @@ use {
  RestError,
  },
  types::{
- BinaryPriceUpdate,
+ BinaryUpdate,
  EncodingType,
  ParsedPriceUpdate,
  PriceIdInput,
@@ -123,7 +123,7 @@ where
  .into_iter()
  .map(|data| query_params.encoding.encode_str(&data))
  .collect();
- let binary_price_update = BinaryPriceUpdate {
+ let binary_price_update = BinaryUpdate {
  encoding: query_params.encoding,
  data: encoded_data,
  };

apps/hermes/server/src/api/types.rs

@@ -28,6 +28,7 @@ use {
  Deserialize,
  Serialize,
  },
+ solana_sdk::pubkey::Pubkey,
  std::{
  collections::BTreeMap,
  fmt::{
@@ -40,6 +41,7 @@ use {
  wormhole_sdk::Chain,
 };
 
+
 /// A price id is a 32-byte hex string, optionally prefixed with "0x".
 /// Price ids are case insensitive.
 ///
@@ -231,7 +233,7 @@ impl EncodingType {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
-pub struct BinaryPriceUpdate {
+pub struct BinaryUpdate {
  pub encoding: EncodingType,
  pub data: Vec<String>,
 }
@@ -271,9 +273,21 @@ impl From<PriceFeedUpdate> for ParsedPriceUpdate {
  }
 }
 
+#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize, Clone)]
+pub struct ParsedPublisherStakeCapsUpdate {
+ pub publisher_stake_caps: Vec<ParsedPublisherStakeCap>,
+}
+
+#[derive(Debug, PartialEq, serde::Serialize, serde::Deserialize, Clone)]
+pub struct ParsedPublisherStakeCap {
+ #[serde(with = "pyth_sdk::utils::as_string")]
+ pub publisher: Pubkey,
+ pub cap: u64,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct PriceUpdate {
- pub binary: BinaryPriceUpdate,
+ pub binary: BinaryUpdate,
  #[serde(skip_serializing_if = "Option::is_none")]
  pub parsed: Option<Vec<ParsedPriceUpdate>>,
 }

apps/hermes/server/src/state/aggregate.rs

@@ -17,6 +17,10 @@ use {
  WormholeMerkleState,
  },
  crate::{
+ api::types::{
+ ParsedPublisherStakeCap,
+ ParsedPublisherStakeCapsUpdate,
+ },
  network::wormhole::VaaBytes,
  state::{
  benchmarks::Benchmarks,
@@ -45,6 +49,7 @@ use {
  messages::{
  Message,
  MessageType,
+ PUBLISHER_STAKE_CAPS_MESSAGE_FEED_ID,
  },
  wire::{
  from_slice,
@@ -55,6 +60,7 @@ use {
  },
  },
  serde::Serialize,
+ solana_sdk::pubkey::Pubkey,
  std::{
  collections::HashSet,
  time::Duration,
@@ -68,7 +74,6 @@ use {
  },
  wormhole_sdk::Vaa,
 };
-
 pub mod metrics;
 pub mod wormhole_merkle;
 
@@ -215,6 +220,12 @@ pub struct PriceFeedsWithUpdateData {
  pub update_data: Vec<Vec<u8>>,
 }
 
+#[derive(Debug, PartialEq)]
+pub struct PublisherStakeCapsWithUpdateData {
+ pub publisher_stake_caps: Vec<ParsedPublisherStakeCapsUpdate>,
+ pub update_data: Vec<Vec<u8>>,
+}
+
 #[derive(Debug, Serialize)]
 pub struct ReadinessMetadata {
  pub has_completed_recently: bool,
@@ -242,6 +253,9 @@ where
  price_ids: &[PriceIdentifier],
  request_time: RequestTime,
  ) -> Result<PriceFeedsWithUpdateData>;
+ async fn get_latest_publisher_stake_caps_with_update_data(
+ &self,
+ ) -> Result<PublisherStakeCapsWithUpdateData>;
 }
 
 /// Allow downcasting State into CacheState for functions that depend on the `Cache` service.
@@ -403,10 +417,46 @@ where
  }
  }
 
+ async fn get_latest_publisher_stake_caps_with_update_data(
+ &self,
+ ) -> Result<PublisherStakeCapsWithUpdateData> {
+ let messages = self
+ .fetch_message_states(
+ vec![PUBLISHER_STAKE_CAPS_MESSAGE_FEED_ID],
+ RequestTime::Latest,
+ MessageStateFilter::Only(MessageType::PublisherStakeCapsMessage),
+ )
+ .await?;
+
+ let publisher_stake_caps = messages
+ .iter()
+ .map(|message_state| match message_state.message.clone() {
+ Message::PublisherStakeCapsMessage(message) => Ok(ParsedPublisherStakeCapsUpdate {
+ publisher_stake_caps: message
+ .caps
+ .iter()
+ .map(|cap| ParsedPublisherStakeCap {
+ publisher: Pubkey::from(cap.publisher),
+ cap: cap.cap,
+ })
+ .collect(),
+ }),
+ _ => Err(anyhow!("Invalid message state type")),
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ let update_data = construct_update_data(messages.into_iter().map(|m| m.into()).collect())?;
+ Ok(PublisherStakeCapsWithUpdateData {
+ publisher_stake_caps,
+ update_data,
+ })
+ }
+
  async fn get_price_feed_ids(&self) -> HashSet<PriceIdentifier> {
  Cache::message_state_keys(self)
  .await
  .iter()
+ .filter(|key| key.feed_id != PUBLISHER_STAKE_CAPS_MESSAGE_FEED_ID)
  .map(|key| PriceIdentifier::new(key.feed_id))
  .collect()
  }