fix(apps/price_pusher) handle price feed removal more gracefully

apps/price_pusher/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pythnetwork/price-pusher",
-  "version": "9.3.3",
+  "version": "9.3.4",
   "description": "Pyth Price Pusher",
   "homepage": "https://pyth.network",
   "main": "lib/index.js",

apps/price_pusher/src/aptos/aptos.ts

@@ -109,6 +109,7 @@ export class AptosPricePusher implements IPricePusher {
   async getPriceFeedsUpdateData(priceIds: string[]): Promise<number[][]> {
     const response = await this.hermesClient.getLatestPriceUpdates(priceIds, {
       encoding: "base64",
+      ignoreInvalidPriceIds: true,
     });
     return response.binary.data.map((data) =>
       Array.from(Buffer.from(data, "base64")),

apps/price_pusher/src/fuel/fuel.ts

@@ -101,6 +101,7 @@ export class FuelPricePusher implements IPricePusher {
     try {
       const response = await this.hermesClient.getLatestPriceUpdates(priceIds, {
         encoding: "base64",
+        ignoreInvalidPriceIds: true,
       });
       priceFeedUpdateData = response.binary.data;
     } catch (err: any) {

apps/price_pusher/src/injective/injective.ts

@@ -301,6 +301,7 @@ export class InjectivePricePusher implements IPricePusher {
     try {
       const response = await this.hermesClient.getLatestPriceUpdates(priceIds, {
         encoding: "base64",
+        ignoreInvalidPriceIds: true,
       });
       const vaas = response.binary.data;
 

apps/price_pusher/src/near/near.ts

@@ -131,6 +131,7 @@ export class NearPricePusher implements IPricePusher {
   ): Promise<string[]> {
     const response = await this.hermesClient.getLatestPriceUpdates(priceIds, {
       encoding: "base64",
+      ignoreInvalidPriceIds: true,
     });
     return response.binary.data;
   }

apps/price_pusher/src/price-config.ts

@@ -96,9 +96,12 @@ export function shouldUpdate(
 ): UpdateCondition {
   const priceId = priceConfig.id;
 
-  // There is no price to update the target with.
+  // There is no price to update the target with. So we should not update it.
   if (sourceLatestPrice === undefined) {
-    return UpdateCondition.YES;
+    logger.info(
+      `${priceConfig.alias} (${priceId}) is not available on the source network. Ignoring it.`,
+    );
+    return UpdateCondition.NO;
   }
 
   // It means that price never existed there. So we should push the latest price feed.
@@ -140,7 +143,7 @@ export function shouldUpdate(
       }%? / early: < ${priceConfig.earlyUpdatePriceDeviation}%?) OR ` +
       `Confidence ratio: ${confidenceRatioPct.toFixed(5)}% (< ${
         priceConfig.confidenceRatio
-      }%? / early: < ${priceConfig.earlyUpdatePriceDeviation}%?)`,
+      }%? / early: < ${priceConfig.earlyUpdateConfidenceRatio}%?)`,
   );
 
   if (

apps/price_pusher/src/pyth-price-listener.ts

@@ -5,6 +5,7 @@ import {
 } from "@pythnetwork/hermes-client";
 import { PriceInfo, IPriceListener, PriceItem } from "./interface";
 import { Logger } from "pino";
+import { sleep } from "./utils";
 
 type TimestampInMs = number & { readonly _: unique symbol };
 
@@ -34,6 +35,24 @@ export class PythPriceListener implements IPriceListener {
   // This method should be awaited on and once it finishes it has the latest value
   // for the given price feeds (if they exist).
   async start() {
+    this.startListening();
+
+    // Store health check interval reference
+    this.healthCheckInterval = setInterval(() => {
+      if (
+        this.lastUpdated === undefined ||
+        this.lastUpdated < Date.now() - 30 * 1000
+      ) {
+        throw new Error("Hermes Price feeds are not updating.");
+      }
+    }, 5000);
+  }
+
+  async startListening() {
+    this.logger.info(
+      `Starting to listen for price updates from Hermes for ${this.priceIds.length} price feeds.`,
+    );
+
     const eventSource = await this.hermesClient.getPriceUpdatesStream(
       this.priceIds,
       {
@@ -71,20 +90,12 @@ export class PythPriceListener implements IPriceListener {
       });
     };
 
-    eventSource.onerror = (error: Event) => {
+    eventSource.onerror = async (error: Event) => {
       console.error("Error receiving updates from Hermes:", error);
       eventSource.close();
+      await sleep(5000); // Wait a bit before trying to reconnect
+      this.startListening(); // Attempt to restart the listener
     };
-
-    // Store health check interval reference
-    this.healthCheckInterval = setInterval(() => {
-      if (
-        this.lastUpdated === undefined ||
-        this.lastUpdated < Date.now() - 30 * 1000
-      ) {
-        throw new Error("Hermes Price feeds are not updating.");
-      }
-    }, 5000);
   }
 
   getLatestPriceInfo(priceId: HexString): PriceInfo | undefined {

apps/price_pusher/src/solana/solana.ts

@@ -118,6 +118,7 @@ export class SolanaPricePusher implements IPricePusher {
         shuffledPriceIds,
         {
           encoding: "base64",
+          ignoreInvalidPriceIds: true,
         },
       );
       priceFeedUpdateData = response.binary.data;

apps/price_pusher/src/sui/sui.ts

@@ -225,6 +225,7 @@ export class SuiPricePusher implements IPricePusher {
           priceIdChunk,
           {
             encoding: "base64",
+            ignoreInvalidPriceIds: true,
           },
         );
         if (response.binary.data.length !== 1) {

apps/price_pusher/src/ton/ton.ts

@@ -98,6 +98,7 @@ export class TonPricePusher implements IPricePusher {
     try {
       const response = await this.hermesClient.getLatestPriceUpdates(priceIds, {
         encoding: "base64",
+        ignoreInvalidPriceIds: true,
       });
       priceFeedUpdateData = response.binary.data;
     } catch (err: any) {