From a7bb9160c48420a4e1857cfac0251e37ddecb19f Mon Sep 17 00:00:00 2001 From: Daniel Chew Date: Mon, 15 Apr 2024 21:38:46 +0900 Subject: [PATCH] feat(hermes): add additional sse features (#1443) * add allow_unordered query param * add benchmarks_only query params * update docs * bump * address comments * address comments * address comments --- hermes/Cargo.lock | 2 +- hermes/Cargo.toml | 2 +- hermes/src/api.rs | 70 +++++++------- hermes/src/api/rest.rs | 1 + hermes/src/api/rest/index.rs | 1 + .../src/api/rest/v2/latest_price_updates.rs | 4 +- hermes/src/api/rest/v2/mod.rs | 1 + hermes/src/api/{ => rest/v2}/sse.rs | 92 +++++++++++++++---- .../api/rest/v2/timestamp_price_updates.rs | 4 +- 9 files changed, 116 insertions(+), 61 deletions(-) rename hermes/src/api/{ => rest/v2}/sse.rs (65%) diff --git a/hermes/Cargo.lock b/hermes/Cargo.lock index c3332c8b6c..23eeeaf642 100644 --- a/hermes/Cargo.lock +++ b/hermes/Cargo.lock @@ -1796,7 +1796,7 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hermes" -version = "0.5.4" +version = "0.5.5" dependencies = [ "anyhow", "async-trait", diff --git a/hermes/Cargo.toml b/hermes/Cargo.toml index a62616921d..a67dabbfb2 100644 --- a/hermes/Cargo.toml +++ b/hermes/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hermes" -version = "0.5.4" +version = "0.5.5" description = "Hermes is an agent that provides Verified Prices from the Pythnet Pyth Oracle." edition = "2021" diff --git a/hermes/src/api.rs b/hermes/src/api.rs index 107c65fb21..1dc4c70a98 100644 --- a/hermes/src/api.rs +++ b/hermes/src/api.rs @@ -23,7 +23,6 @@ use { mod doc_examples; mod metrics_middleware; mod rest; -mod sse; pub mod types; mod ws; @@ -96,39 +95,40 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> { #[derive(OpenApi)] #[openapi( - paths( - rest::get_price_feed, - rest::get_vaa, - rest::get_vaa_ccip, - rest::latest_price_feeds, - rest::latest_vaas, - rest::price_feed_ids, - rest::latest_price_updates, - rest::timestamp_price_updates, - rest::price_feeds_metadata, - ), - components( - schemas( - rest::GetVaaCcipInput, - rest::GetVaaCcipResponse, - rest::GetVaaResponse, - types::PriceIdInput, - types::RpcPrice, - types::RpcPriceFeed, - types::RpcPriceFeedMetadata, - types::RpcPriceIdentifier, - types::EncodingType, - types::PriceUpdate, - types::BinaryPriceUpdate, - types::ParsedPriceUpdate, - types::RpcPriceFeedMetadataV2, - types::PriceFeedMetadata, - types::AssetType - ) - ), - tags( - (name = "hermes", description = "Pyth Real-Time Pricing API") - ) + paths( + rest::get_price_feed, + rest::get_vaa, + rest::get_vaa_ccip, + rest::latest_price_feeds, + rest::latest_vaas, + rest::price_feed_ids, + rest::latest_price_updates, + rest::timestamp_price_updates, + rest::price_feeds_metadata, + rest::price_stream_sse_handler, + ), + components( + schemas( + rest::GetVaaCcipInput, + rest::GetVaaCcipResponse, + rest::GetVaaResponse, + types::PriceIdInput, + types::RpcPrice, + types::RpcPriceFeed, + types::RpcPriceFeedMetadata, + types::RpcPriceIdentifier, + types::EncodingType, + types::PriceUpdate, + types::BinaryPriceUpdate, + types::ParsedPriceUpdate, + types::RpcPriceFeedMetadataV2, + types::PriceFeedMetadata, + types::AssetType + ) + ), + tags( + (name = "hermes", description = "Pyth Real-Time Pricing API") + ) )] struct ApiDoc; @@ -146,7 +146,7 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> { .route("/api/price_feed_ids", get(rest::price_feed_ids)) .route( "/v2/updates/price/stream", - get(sse::price_stream_sse_handler), + get(rest::price_stream_sse_handler), ) .route("/v2/updates/price/latest", get(rest::latest_price_updates)) .route( diff --git a/hermes/src/api/rest.rs b/hermes/src/api/rest.rs index 7cee21a9f8..9d29cc9448 100644 --- a/hermes/src/api/rest.rs +++ b/hermes/src/api/rest.rs @@ -35,6 +35,7 @@ pub use { v2::{ latest_price_updates::*, price_feeds_metadata::*, + sse::*, timestamp_price_updates::*, }, }; diff --git a/hermes/src/api/rest/index.rs b/hermes/src/api/rest/index.rs index f9dcdf1bc7..99611f821e 100644 --- a/hermes/src/api/rest/index.rs +++ b/hermes/src/api/rest/index.rs @@ -17,6 +17,7 @@ pub async fn index() -> impl IntoResponse { "/api/get_vaa?id=&publish_time=", "/api/get_vaa_ccip?data=<0x+>", "/v2/updates/price/latest?ids[]=&ids[]=&..(&encoding=hex|base64)(&parsed=false)", + "/v2/updates/price/stream?ids[]=&ids[]=&..(&encoding=hex|base64)(&parsed=false)(&allow_unordered=false)(&benchmarks_only=false)", "/v2/updates/price/?ids[]=&ids[]=&..(&encoding=hex|base64)(&parsed=false)", "/v2/price_feeds?(query=btc)(&asset_type=crypto|equity|fx|metal|rates)", ]) diff --git a/hermes/src/api/rest/v2/latest_price_updates.rs b/hermes/src/api/rest/v2/latest_price_updates.rs index c5a4b3d5b3..df2f40116d 100644 --- a/hermes/src/api/rest/v2/latest_price_updates.rs +++ b/hermes/src/api/rest/v2/latest_price_updates.rs @@ -46,11 +46,11 @@ pub struct LatestPriceUpdatesQueryParams { #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")] ids: Vec, - /// If true, include the parsed price update in the `parsed` field of each returned feed. + /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `hex`. #[serde(default)] encoding: EncodingType, - /// If true, include the parsed price update in the `parsed` field of each returned feed. + /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `true`. #[serde(default = "default_true")] parsed: bool, } diff --git a/hermes/src/api/rest/v2/mod.rs b/hermes/src/api/rest/v2/mod.rs index b8102741fa..a02ddbc482 100644 --- a/hermes/src/api/rest/v2/mod.rs +++ b/hermes/src/api/rest/v2/mod.rs @@ -1,3 +1,4 @@ pub mod latest_price_updates; pub mod price_feeds_metadata; +pub mod sse; pub mod timestamp_price_updates; diff --git a/hermes/src/api/sse.rs b/hermes/src/api/rest/v2/sse.rs similarity index 65% rename from hermes/src/api/sse.rs rename to hermes/src/api/rest/v2/sse.rs index 1b690836ad..b5094b3098 100644 --- a/hermes/src/api/sse.rs +++ b/hermes/src/api/rest/v2/sse.rs @@ -15,6 +15,7 @@ use { ParsedPriceUpdate, PriceIdInput, PriceUpdate, + RpcPriceIdentifier, }, ApiState, }, @@ -56,13 +57,21 @@ pub struct StreamPriceUpdatesQueryParams { #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")] ids: Vec, - /// If true, include the parsed price update in the `parsed` field of each returned feed. + /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `hex`. #[serde(default)] encoding: EncodingType, - /// If true, include the parsed price update in the `parsed` field of each returned feed. + /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `true`. #[serde(default = "default_true")] parsed: bool, + + /// If true, allows unordered price updates to be included in the stream. + #[serde(default)] + allow_unordered: bool, + + /// If true, only include benchmark prices that are the initial price updates at a given timestamp (i.e., prevPubTime != pubTime). + #[serde(default)] + benchmarks_only: bool, } fn default_true() -> bool { @@ -105,10 +114,15 @@ pub async fn price_stream_sse_handler( price_ids_clone, params.encoding, params.parsed, + params.benchmarks_only, + params.allow_unordered, ) .await { - Ok(price_update) => Ok(Event::default().json_data(price_update).unwrap()), + Ok(Some(update)) => Ok(Event::default() + .json_data(update) + .unwrap_or_else(|e| error_event(e))), + Ok(None) => Ok(Event::default().comment("No update available")), Err(e) => Ok(error_event(e)), } } @@ -126,18 +140,64 @@ async fn handle_aggregation_event( mut price_ids: Vec, encoding: EncodingType, parsed: bool, -) -> Result { + benchmarks_only: bool, + allow_unordered: bool, +) -> Result> { + // Handle out-of-order events + if let AggregationEvent::OutOfOrder { .. } = event { + if !allow_unordered { + return Ok(None); + } + } + // We check for available price feed ids to ensure that the price feed ids provided exists since price feeds can be removed. let available_price_feed_ids = crate::aggregate::get_price_feed_ids(&*state.state).await; price_ids.retain(|price_feed_id| available_price_feed_ids.contains(price_feed_id)); - let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data( + let mut price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data( &*state.state, &price_ids, RequestTime::AtSlot(event.slot()), ) .await?; + + let mut parsed_price_updates: Vec = price_feeds_with_update_data + .price_feeds + .into_iter() + .map(|price_feed| price_feed.into()) + .collect(); + + + if benchmarks_only { + // Remove those with metadata.prev_publish_time != price.publish_time from parsed_price_updates + parsed_price_updates.retain(|price_feed| { + price_feed + .metadata + .prev_publish_time + .map_or(false, |prev_time| { + prev_time != price_feed.price.publish_time + }) + }); + // Retain price id in price_ids that are in parsed_price_updates + price_ids.retain(|price_id| { + parsed_price_updates + .iter() + .any(|price_feed| price_feed.id == RpcPriceIdentifier::from(*price_id)) + }); + price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data( + &*state.state, + &price_ids, + RequestTime::AtSlot(event.slot()), + ) + .await?; + } + + // Check if price_ids is empty after filtering and return None if it is + if price_ids.is_empty() { + return Ok(None); + } + let price_update_data = price_feeds_with_update_data.update_data; let encoded_data: Vec = price_update_data .into_iter() @@ -147,23 +207,15 @@ async fn handle_aggregation_event( encoding, data: encoded_data, }; - let parsed_price_updates: Option> = if parsed { - Some( - price_feeds_with_update_data - .price_feeds - .into_iter() - .map(|price_feed| price_feed.into()) - .collect(), - ) - } else { - None - }; - - Ok(PriceUpdate { + Ok(Some(PriceUpdate { binary: binary_price_update, - parsed: parsed_price_updates, - }) + parsed: if parsed { + Some(parsed_price_updates) + } else { + None + }, + })) } fn error_event(e: E) -> Event { diff --git a/hermes/src/api/rest/v2/timestamp_price_updates.rs b/hermes/src/api/rest/v2/timestamp_price_updates.rs index 64ebc1cc34..f47d2e7180 100644 --- a/hermes/src/api/rest/v2/timestamp_price_updates.rs +++ b/hermes/src/api/rest/v2/timestamp_price_updates.rs @@ -58,11 +58,11 @@ pub struct TimestampPriceUpdatesQueryParams { #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")] ids: Vec, - /// If true, include the parsed price update in the `parsed` field of each returned feed. + /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `hex`. #[serde(default)] encoding: EncodingType, - /// If true, include the parsed price update in the `parsed` field of each returned feed. + /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `true`. #[serde(default = "default_true")] parsed: bool, }