blockworks-foundation
/
pyth-crosschain
mirror of https://github.com/blockworks-foundation/pyth-crosschain.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Daniel Chew 2024-04-15 21:38:46 +09:00 committed by GitHub
parent 392a3df7eb
commit a7bb9160c4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 116 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hermes/Cargo.lock generated
View File

@ -1796,7 +1796,7 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
[[package]]
name = "hermes"
version = "0.5.4"
version = "0.5.5"
dependencies = [
"anyhow",
"async-trait",

2
hermes/Cargo.toml
View File

@ -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"

70
hermes/src/api.rs
View File

@ -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(

1
hermes/src/api/rest.rs
View File

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

1
hermes/src/api/rest/index.rs
View File

@ -17,6 +17,7 @@ pub async fn index() -> impl IntoResponse {
"/api/get_vaa?id=<price_feed_id>&publish_time=<publish_time_in_unix_timestamp>",
"/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/stream?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..(&encoding=hex|base64)(&parsed=false)(&allow_unordered=false)(&benchmarks_only=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)",
])

4
hermes/src/api/rest/v2/latest_price_updates.rs
View File

@ -46,11 +46,11 @@ pub struct LatestPriceUpdatesQueryParams {
#[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
ids: Vec<PriceIdInput>,
/// 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,
}

1
hermes/src/api/rest/v2/mod.rs
View File

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

92
hermes/src/api/sse.rs → hermes/src/api/rest/v2/sse.rs
View File

@ -15,6 +15,7 @@ use {
ParsedPriceUpdate,
PriceIdInput,
PriceUpdate,
RpcPriceIdentifier,
},
ApiState,
},
@ -56,13 +57,21 @@ pub struct StreamPriceUpdatesQueryParams {
#[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
ids: Vec<PriceIdInput>,
/// 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<PriceIdentifier>,
encoding: EncodingType,
parsed: bool,
) -> Result<PriceUpdate> {
benchmarks_only: bool,
allow_unordered: bool,
) -> Result<Option<PriceUpdate>> {
// 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<ParsedPriceUpdate> = 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<String> = price_update_data
.into_iter()
@ -147,23 +207,15 @@ async fn handle_aggregation_event(
encoding,
data: encoded_data,
};
let parsed_price_updates: Option<Vec<ParsedPriceUpdate>> = 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: std::fmt::Debug>(e: E) -> Event {

4
hermes/src/api/rest/v2/timestamp_price_updates.rs
View File

@ -58,11 +58,11 @@ pub struct TimestampPriceUpdatesQueryParams {
#[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
ids: Vec<PriceIdInput>,
/// 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,
}