Skip to content

Commit 52c2def

Browse files
cctdanielcctdaniel
authored
fix(apps/price_pusher): fix bug causing price_pusher to hand when invalid price feed ids are passed in to hermes ws (#2297)
* fix * add cleanup
1 parent 4e10ebe commit 52c2def

File tree

3 files changed

+208
-14
lines changed

3 files changed

+208
-14
lines changed

‎apps/price_pusher/package.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@pythnetwork/price-pusher",
3-
"version": "8.3.2",
3+
"version": "8.3.3",
44
"description": "Pyth Price Pusher",
55
"homepage": "https://pyth.network",
66
"main": "lib/index.js",
@@ -24,6 +24,7 @@
2424
"format": "prettier --write \"src/**/*.ts\"",
2525
"test:lint": "eslint src/",
2626
"start": "node lib/index.js",
27+
"test": "jest",
2728
"dev": "ts-node src/index.ts",
2829
"prepublishOnly": "pnpm run build && pnpm run test:lint",
2930
"preversion": "pnpm run test:lint",

‎apps/price_pusher/src/__tests__/pyth-price-listener.test.ts‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
import { PythPriceListener } from "../pyth-price-listener";
2+
import { PriceServiceConnection } from "@pythnetwork/price-service-client";
3+
import { Logger } from "pino";
4+
5+
describe("PythPriceListener", () => {
6+
let logger: Logger;
7+
let connection: PriceServiceConnection;
8+
let listener: PythPriceListener;
9+
let originalConsoleError: typeof console.error;
10+
11+
beforeEach(() => {
12+
// Save original console.error and mock it
13+
originalConsoleError = console.error;
14+
console.error = jest.fn();
15+
16+
logger = {
17+
debug: jest.fn(),
18+
error: jest.fn(),
19+
info: jest.fn(),
20+
} as unknown as Logger;
21+
22+
// Use real Hermes beta endpoint for testing
23+
connection = new PriceServiceConnection("https://hermes.pyth.network");
24+
});
25+
26+
afterEach(() => {
27+
// Clean up websocket connection
28+
connection.closeWebSocket();
29+
// Clean up health check interval
30+
if (listener) {
31+
listener.cleanup();
32+
}
33+
// Restore original console.error
34+
console.error = originalConsoleError;
35+
});
36+
37+
it("should handle invalid price feeds gracefully", async () => {
38+
const validFeedId =
39+
"e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43"; // BTC/USD
40+
const invalidFeedId =
41+
"0000000000000000000000000000000000000000000000000000000000000000";
42+
43+
const priceItems = [
44+
{ id: validFeedId, alias: "BTC/USD" },
45+
{ id: invalidFeedId, alias: "INVALID/PRICE" },
46+
];
47+
48+
listener = new PythPriceListener(connection, priceItems, logger);
49+
50+
await listener.start();
51+
52+
// Wait for both error handlers to complete
53+
await new Promise((resolve) => {
54+
const checkInterval = setInterval(() => {
55+
const errorCalls = (logger.error as jest.Mock).mock.calls;
56+
57+
// Check for both HTTP and websocket error logs
58+
const hasHttpError = errorCalls.some(
59+
(call) => call[0] === "Failed to get latest price feeds:"
60+
);
61+
const hasGetLatestError = errorCalls.some((call) =>
62+
call[0].includes("not found for getLatestPriceFeeds")
63+
);
64+
const hasWsError = errorCalls.some((call) =>
65+
call[0].includes("not found for subscribePriceFeedUpdates")
66+
);
67+
68+
if (hasHttpError && hasGetLatestError && hasWsError) {
69+
clearInterval(checkInterval);
70+
resolve(true);
71+
}
72+
}, 100);
73+
});
74+
75+
// Verify HTTP error was logged
76+
expect(logger.error).toHaveBeenCalledWith(
77+
"Failed to get latest price feeds:",
78+
expect.objectContaining({
79+
message: "Request failed with status code 404",
80+
})
81+
);
82+
83+
// Verify invalid feed error was logged
84+
expect(logger.error).toHaveBeenCalledWith(
85+
`Price feed ${invalidFeedId} (INVALID/PRICE) not found for getLatestPriceFeeds`
86+
);
87+
88+
// Verify invalid feed error was logged
89+
expect(logger.error).toHaveBeenCalledWith(
90+
`Price feed ${invalidFeedId} (INVALID/PRICE) not found for subscribePriceFeedUpdates`
91+
);
92+
93+
// Verify resubscription message was logged
94+
expect(logger.info).toHaveBeenCalledWith(
95+
"Resubscribing with valid feeds only"
96+
);
97+
98+
// Verify priceIds was updated to only include valid feeds
99+
expect(listener["priceIds"]).toEqual([validFeedId]);
100+
});
101+
});

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

Lines changed: 105 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ export class PythPriceListener implements IPriceListener {
1515
private latestPriceInfo: Map<HexString, PriceInfo>;
1616
private logger: Logger;
1717
private lastUpdated: TimestampInMs | undefined;
18+
private healthCheckInterval?: NodeJS.Timeout;
1819

1920
constructor(
2021
connection: PriceServiceConnection,
@@ -33,26 +34,111 @@ export class PythPriceListener implements IPriceListener {
3334
// This method should be awaited on and once it finishes it has the latest value
3435
// for the given price feeds (if they exist).
3536
async start() {
37+
// Set custom error handler for websocket errors
38+
this.connection.onWsError = (error: Error) => {
39+
if (error.message.includes("not found")) {
40+
// Extract invalid feed IDs from error message
41+
const match = error.message.match(/\[(.*?)\]/);
42+
if (match) {
43+
const invalidFeedIds = match[1].split(",").map((id) => {
44+
// Remove '0x' prefix if present to match our stored IDs
45+
return id.trim().replace(/^0x/, "");
46+
});
47+
48+
// Log invalid feeds with their aliases
49+
invalidFeedIds.forEach((id) => {
50+
this.logger.error(
51+
`Price feed ${id} (${this.priceIdToAlias.get(
52+
id
53+
)}) not found for subscribePriceFeedUpdates`
54+
);
55+
});
56+
57+
// Filter out invalid feeds and resubscribe with valid ones
58+
const validFeeds = this.priceIds.filter(
59+
(id) => !invalidFeedIds.includes(id)
60+
);
61+
62+
this.priceIds = validFeeds;
63+
64+
if (validFeeds.length > 0) {
65+
this.logger.info("Resubscribing with valid feeds only");
66+
this.connection.subscribePriceFeedUpdates(
67+
validFeeds,
68+
this.onNewPriceFeed.bind(this)
69+
);
70+
}
71+
}
72+
} else {
73+
this.logger.error("Websocket error occurred:", error);
74+
}
75+
};
76+
3677
this.connection.subscribePriceFeedUpdates(
3778
this.priceIds,
3879
this.onNewPriceFeed.bind(this)
3980
);
4081

41-
const priceFeeds = await this.connection.getLatestPriceFeeds(this.priceIds);
42-
priceFeeds?.forEach((priceFeed) => {
43-
// Getting unchecked because although it might be old
44-
// but might not be there on the target chain.
45-
const latestAvailablePrice = priceFeed.getPriceUnchecked();
46-
this.latestPriceInfo.set(priceFeed.id, {
47-
price: latestAvailablePrice.price,
48-
conf: latestAvailablePrice.conf,
49-
publishTime: latestAvailablePrice.publishTime,
82+
try {
83+
const priceFeeds = await this.connection.getLatestPriceFeeds(
84+
this.priceIds
85+
);
86+
priceFeeds?.forEach((priceFeed) => {
87+
const latestAvailablePrice = priceFeed.getPriceUnchecked();
88+
this.latestPriceInfo.set(priceFeed.id, {
89+
price: latestAvailablePrice.price,
90+
conf: latestAvailablePrice.conf,
91+
publishTime: latestAvailablePrice.publishTime,
92+
});
5093
});
51-
});
94+
} catch (error: any) {
95+
// Always log the HTTP error first
96+
this.logger.error("Failed to get latest price feeds:", error);
97+
98+
if (error.response.data.includes("Price ids not found:")) {
99+
// Extract invalid feed IDs from error message
100+
const invalidFeedIds = error.response.data
101+
.split("Price ids not found:")[1]
102+
.split(",")
103+
.map((id: string) => id.trim().replace(/^0x/, ""));
104+
105+
// Log invalid feeds with their aliases
106+
invalidFeedIds.forEach((id: string) => {
107+
this.logger.error(
108+
`Price feed ${id} (${this.priceIdToAlias.get(
109+
id
110+
)}) not found for getLatestPriceFeeds`
111+
);
112+
});
52113

53-
// Check health of the price feeds 5 second. If the price feeds are not updating
54-
// for more than 30s, throw an error.
55-
setInterval(() => {
114+
// Filter out invalid feeds and retry
115+
const validFeeds = this.priceIds.filter(
116+
(id) => !invalidFeedIds.includes(id)
117+
);
118+
119+
this.priceIds = validFeeds;
120+
121+
if (validFeeds.length > 0) {
122+
this.logger.info(
123+
"Retrying getLatestPriceFeeds with valid feeds only"
124+
);
125+
const validPriceFeeds = await this.connection.getLatestPriceFeeds(
126+
validFeeds
127+
);
128+
validPriceFeeds?.forEach((priceFeed) => {
129+
const latestAvailablePrice = priceFeed.getPriceUnchecked();
130+
this.latestPriceInfo.set(priceFeed.id, {
131+
price: latestAvailablePrice.price,
132+
conf: latestAvailablePrice.conf,
133+
publishTime: latestAvailablePrice.publishTime,
134+
});
135+
});
136+
}
137+
}
138+
}
139+
140+
// Store health check interval reference
141+
this.healthCheckInterval = setInterval(() => {
56142
if (
57143
this.lastUpdated === undefined ||
58144
this.lastUpdated < Date.now() - 30 * 1000
@@ -88,4 +174,10 @@ export class PythPriceListener implements IPriceListener {
88174
getLatestPriceInfo(priceId: string): PriceInfo | undefined {
89175
return this.latestPriceInfo.get(priceId);
90176
}
177+
178+
cleanup() {
179+
if (this.healthCheckInterval) {
180+
clearInterval(this.healthCheckInterval);
181+
}
182+
}
91183
}

0 commit comments

Comments
 (0)