chore: address comments

Author: ali-behjati

hermes/src/aggregate.rs

@@ -251,7 +251,7 @@ pub async fn store_update(state: &State, update: Update) -> Result<()> {
 
     // Once the accumulator reaches a complete state for a specific slot
     // we can build the message states
-    let message_states = build_message_states(accumulator_messages, wormhole_merkle_state).await?;
+    let message_states = build_message_states(accumulator_messages, wormhole_merkle_state)?;
 
     let message_state_keys = message_states
         .iter()
@@ -306,7 +306,7 @@ pub async fn store_update(state: &State, update: Update) -> Result<()> {
 }
 
 #[tracing::instrument(skip(accumulator_messages, wormhole_merkle_state))]
-async fn build_message_states(
+fn build_message_states(
     accumulator_messages: AccumulatorMessages,
     wormhole_merkle_state: WormholeMerkleState,
 ) -> Result<Vec<MessageState>> {

hermes/src/api/ws.rs

@@ -362,20 +362,27 @@ impl Subscriber {
             Err(_) => {
                 // The error can only happen when a price feed was available
                 // and is no longer there as we check the price feed ids upon
-                // subscription. In this case we send an error message and
-                // close the connection.
-                self.sender
-                    .send(
-                        serde_json::to_string(&ServerResponseMessage::Err {
-                            error: "Some of the subscribed price feeds have been removed. Closing connection."
-                                .to_string(),
-                        })?
-                        .into(),
-                    )
-                    .await?;
-                self.sender.close().await?;
-                self.closed = true;
-                return Ok(());
+                // subscription. In this case we just remove the non-existing
+                // price feed from the list and will keep sending updates for
+                // the rest.
+                let available_price_feed_ids =
+                    crate::aggregate::get_price_feed_ids(&*self.store).await;
+
+                self.price_feeds_with_config
+                    .retain(|price_feed_id, _| available_price_feed_ids.contains(price_feed_id));
+
+                let price_feed_ids = self
+                    .price_feeds_with_config
+                    .keys()
+                    .cloned()
+                    .collect::<Vec<_>>();
+
+                crate::aggregate::get_price_feeds_with_update_data(
+                    &*self.store,
+                    &price_feed_ids,
+                    RequestTime::AtSlot(event.slot()),
+                )
+                .await?
             }
         };
 

hermes/src/state/cache.rs

@@ -228,7 +228,7 @@ impl AggregateCache for crate::state::State {
     async fn prune_removed_keys(&self, current_keys: HashSet<MessageStateKey>) {
         let mut message_cache = self.cache.message_cache.write().await;
 
-        // Sometiems, some keys are removed from the accumulator. We track which keys are not
+        // Sometimes, some keys are removed from the accumulator. We track which keys are not
         // present in the message states and remove them from the cache.
         let keys_in_cache = message_cache
             .iter()
@@ -237,6 +237,7 @@ impl AggregateCache for crate::state::State {
 
         for key in keys_in_cache {
             if !current_keys.contains(&key) {
+                tracing::info!("Feed {:?} seems to be removed. Removing it from cache", key);
                 message_cache.remove(&key);
             }
         }