feat(hermes): Add ignore_invalid_price_ids flag to Hermes v2 REST APIs

apps/hermes/server/README.md

@@ -21,9 +21,10 @@ To set up and run a Hermes node, follow the steps below:
    using [Beacon](https://github.com/pyth-network/beacon), a highly available rewrite for spy, for production purposes.
 1. **Install Rust nightly-2023-07-23**: If you haven't already, you'll need to install Rust. You can
    do so by following the official instructions. Then, run the following command to install the required
-   nightly version of Rust:
+   nightly version of Rust and set it as the default:
    ```bash
     rustup toolchain install nightly-2023-07-23
+    rustup default nightly-2023-07-23
    ```
 2. **Install Go**: If you haven't already, you'll also need to install Go. You can
    do so by following the official instructions. If you are on a Mac with M series
@@ -52,8 +53,10 @@ To set up and run a Hermes node, follow the steps below:
    can interact with the node using the REST and Websocket APIs on port 33999.
 
    For local development, you can also run the node with [cargo watch](https://crates.io/crates/cargo-watch) to restart
-   it automatically when the code changes:
+   it automatically when the code changes.
+   Note that cargo-watch v8 and above is incompatible with our `nightly-2023-07-23` rust toolchain, so you'll need to install version 7.x:
 
    ```bash
+   cargo install cargo-watch@^7.x
    cargo watch -w src -x "run -- run --pythnet-http-addr https://pythnet-rpc/ --pythnet-ws-addr wss://pythnet-rpc/ --wormhole-spy-rpc-addr https://wormhole-spy-rpc/
    ```

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

@@ -94,25 +94,45 @@ impl IntoResponse for RestError {
     }
 }
 
-/// Verify that the price ids exist in the aggregate state.
-pub async fn verify_price_ids_exist<S>(
+/// Validate that the passed in price_ids exist in the aggregate state. Return a Vec of valid price ids.
+/// # Returns
+/// If `remove_invalid` is true, invalid price ids are filtered out and only valid price ids are returned.
+/// If `remove_invalid` is false and any passed in IDs are invalid, an error is returned.
+pub async fn validate_price_ids<S>(
     state: &ApiState<S>,
     price_ids: &[PriceIdentifier],
-) -> Result<(), RestError>
+    remove_invalid: bool,
+) -> Result<Vec<PriceIdentifier>, RestError>
 where
     S: Aggregates,
 {
     let state = &*state.state;
-    let all_ids = Aggregates::get_price_feed_ids(state).await;
-    let missing_ids = price_ids
+    let available_ids = Aggregates::get_price_feed_ids(state).await;
+
+    // Find any price IDs that don't exist in the valid set
+    let not_found_ids: Vec<PriceIdentifier> = price_ids
         .iter()
-        .filter(|id| !all_ids.contains(id))
-        .cloned()
-        .collect::<Vec<_>>();
+        .filter(|id| !available_ids.contains(id))
+        .copied()
+        .collect();
 
-    if !missing_ids.is_empty() {
-        return Err(RestError::PriceIdsNotFound { missing_ids });
+    if !not_found_ids.is_empty() {
+        // Some of the passed in IDs are invalid
+        if remove_invalid {
+            // Filter out invalid IDs and return only the valid ones
+            Ok(price_ids
+                .iter()
+                .filter(|id| available_ids.contains(id))
+                .copied()
+                .collect())
+        } else {
+            // Return error with list of missing IDs
+            Err(RestError::PriceIdsNotFound {
+                missing_ids: not_found_ids,
+            })
+        }
+    } else {
+        // All IDs are valid, return them unchanged
+        Ok(price_ids.to_vec())
     }
-
-    Ok(())
 }

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

@@ -1,5 +1,5 @@
 use {
-    super::verify_price_ids_exist,
+    super::validate_price_ids,
     crate::{
         api::{
             doc_examples,
@@ -73,7 +73,7 @@ where
     S: Aggregates,
 {
     let price_id: PriceIdentifier = params.id.into();
-    verify_price_ids_exist(&state, &[price_id]).await?;
+    validate_price_ids(&state, &[price_id], false).await?;
 
     let state = &*state.state;
     let price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(

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

@@ -1,5 +1,5 @@
 use {
-    super::verify_price_ids_exist,
+    super::validate_price_ids,
     crate::{
         api::{
             doc_examples,
@@ -80,7 +80,7 @@ where
     S: Aggregates,
 {
     let price_id: PriceIdentifier = params.id.into();
-    verify_price_ids_exist(&state, &[price_id]).await?;
+    validate_price_ids(&state, &[price_id], false).await?;
 
     let state = &*state.state;
     let price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(

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

@@ -1,5 +1,5 @@
 use {
-    super::verify_price_ids_exist,
+    super::validate_price_ids,
     crate::{
         api::{
             rest::RestError,
@@ -75,7 +75,7 @@ where
             .try_into()
             .map_err(|_| RestError::InvalidCCIPInput)?,
     );
-    verify_price_ids_exist(&state, &[price_id]).await?;
+    validate_price_ids(&state, &[price_id], false).await?;
 
     let publish_time = UnixTimestamp::from_be_bytes(
         params.data[32..40]

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

@@ -1,5 +1,5 @@
 use {
-    super::verify_price_ids_exist,
+    super::validate_price_ids,
     crate::{
         api::{
             rest::RestError,
@@ -74,7 +74,7 @@ where
     S: Aggregates,
 {
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
-    verify_price_ids_exist(&state, &price_ids).await?;
+    validate_price_ids(&state, &price_ids, false).await?;
 
     let state = &*state.state;
     let price_feeds_with_update_data =

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

@@ -1,5 +1,5 @@
 use {
-    super::verify_price_ids_exist,
+    super::validate_price_ids,
     crate::{
         api::{
             doc_examples,
@@ -69,7 +69,7 @@ where
     S: Aggregates,
 {
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
-    verify_price_ids_exist(&state, &price_ids).await?;
+    validate_price_ids(&state, &price_ids, false).await?;
 
     let state = &*state.state;
     let price_feeds_with_update_data =

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

@@ -2,7 +2,7 @@ use {
     crate::{
         api::{
             rest::{
-                verify_price_ids_exist,
+                validate_price_ids,
                 RestError,
             },
             types::{
@@ -57,6 +57,10 @@ pub struct LatestPriceUpdatesQueryParams {
     /// 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, invalid price IDs in the `ids` parameter are ignored. Default is `false`.
+    #[serde(default)]
+    ignore_invalid_price_ids: bool,
 }
 
 fn default_true() -> bool {
@@ -84,8 +88,10 @@ pub async fn latest_price_updates<S>(
 where
     S: Aggregates,
 {
-    let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
-    verify_price_ids_exist(&state, &price_ids).await?;
+    let price_id_inputs: Vec<PriceIdentifier> =
+        params.ids.into_iter().map(|id| id.into()).collect();
+    let price_ids: Vec<PriceIdentifier> =
+        validate_price_ids(&state, &price_id_inputs, params.ignore_invalid_price_ids).await?;
 
     let state = &*state.state;
     let price_feeds_with_update_data =

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

@@ -2,7 +2,7 @@ use {
     crate::{
         api::{
             rest::{
-                verify_price_ids_exist,
+                validate_price_ids,
                 RestError,
             },
             types::{
@@ -73,6 +73,10 @@ pub struct StreamPriceUpdatesQueryParams {
     /// 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,
+
+    /// If true, invalid price IDs in the `ids` parameter are ignored. Default is `false`.
+    #[serde(default)]
+    ignore_invalid_price_ids: bool,
 }
 
 fn default_true() -> bool {
@@ -97,9 +101,9 @@ where
     S: Aggregates,
     S: Send + Sync + 'static,
 {
-    let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(Into::into).collect();
-
-    verify_price_ids_exist(&state, &price_ids).await?;
+    let price_id_inputs: Vec<PriceIdentifier> = params.ids.into_iter().map(Into::into).collect();
+    let price_ids: Vec<PriceIdentifier> =
+        validate_price_ids(&state, &price_id_inputs, params.ignore_invalid_price_ids).await?;
 
     // Clone the update_tx receiver to listen for new price updates
     let update_rx: broadcast::Receiver<AggregationEvent> = Aggregates::subscribe(&*state.state);

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

@@ -3,7 +3,7 @@ use {
         api::{
             doc_examples,
             rest::{
-                verify_price_ids_exist,
+                validate_price_ids,
                 RestError,
             },
             types::{
@@ -67,6 +67,10 @@ pub struct TimestampPriceUpdatesQueryParams {
     /// 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, invalid price IDs in the `ids` parameter are ignored. Default is `false`.
+    #[serde(default)]
+    ignore_invalid_price_ids: bool,
 }
 
 
@@ -97,10 +101,14 @@ pub async fn timestamp_price_updates<S>(
 where
     S: Aggregates,
 {
-    let price_ids: Vec<PriceIdentifier> =
+    let price_id_inputs: Vec<PriceIdentifier> =
         query_params.ids.into_iter().map(|id| id.into()).collect();
-
-    verify_price_ids_exist(&state, &price_ids).await?;
+    let price_ids: Vec<PriceIdentifier> = validate_price_ids(
+        &state,
+        &price_id_inputs,
+        query_params.ignore_invalid_price_ids,
+    )
+    .await?;
 
     let state = &*state.state;
     let price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(