refactor: moves price producer creation into hermes client from update command

Author: stalniy

package.json

@@ -2,8 +2,6 @@
     "name": "akash-hermes-client",
     "version": "1.0.0",
     "description": "Hermes client for updating Akash oracle with Pyth price data",
-    "main": "dist/hermes-client.js",
-    "types": "dist/hermes-client.d.ts",
     "type": "module",
     "bin": {
         "hermes-cli": "dist/cli.js"

src/cli-commands/update-command.test.ts

@@ -1,23 +1,9 @@
 import { describe, expect, it, vi } from "vitest";
 import { mock } from "vitest-mock-extended";
 import type { HermesClient } from "../hermes-client.ts";
-import type { PriceUpdate } from "../types.ts";
 import type { CommandConfig } from "./command-config.ts";
 import { updateCommand } from "./update-command.ts";
 
-const fakePriceUpdate: PriceUpdate = {
-    priceData: {
-        id: "test-feed-id",
-        price: { price: "100", conf: "1", expo: -8, publish_time: 1000 },
-        ema_price: { price: "100", conf: "1", expo: -8, publish_time: 1000 },
-    },
-    vaa: "dGVzdC12YWE=",
-};
-
-async function* fakePriceProducer(): AsyncGenerator<PriceUpdate, void, unknown> {
-    yield fakePriceUpdate;
-}
-
 function setup() {
     const client = mock<HermesClient>();
     client.queryConfig.mockResolvedValue({
@@ -39,7 +25,7 @@ function setup() {
         healthcheckPort: 3000,
         rawConfig: {} as CommandConfig["rawConfig"],
         smartContractConfigCacheTTLMs: 60000,
-        priceProducerFactory: vi.fn(() => fakePriceProducer()),
+        priceProducerFactory: vi.fn(),
         createHermesClient: vi.fn(() => Promise.resolve(client)),
     };
     return { config, client, logger };
@@ -51,15 +37,14 @@ describe("updateCommand", () => {
         await updateCommand(config);
 
         expect(logger.log).toHaveBeenCalledWith("Updating oracle price...\n");
-        expect(logger.log).toHaveBeenCalledWith("\nUpdate completed successfully!");
     });
 
     it("creates client and calls updatePrice", async () => {
         const { config, client } = setup();
         await updateCommand(config);
 
         expect(config.createHermesClient).toHaveBeenCalledWith(config);
-        expect(client.updatePrice).toHaveBeenCalledWith(fakePriceUpdate);
+        expect(client.updatePrice).toHaveBeenCalledWith({ signal: config.signal });
     });
 
     it("propagates errors from updatePrice", async () => {

src/cli-commands/update-command.ts

@@ -3,17 +3,5 @@ import type { CommandConfig } from "./command-config.ts";
 export async function updateCommand(config: CommandConfig): Promise<void> {
     config.logger?.log("Updating oracle price...\n");
     const client = await config.createHermesClient(config);
-    const smartCotractConfig = await client.queryConfig();
-    const priceStream = config.priceProducerFactory({
-        priceFeedId: smartCotractConfig.price_feed_id,
-        logger: config.logger,
-        signal: config.signal,
-    });
-    const priceUpdate = await priceStream.next();
-    if (priceUpdate.value) {
-        await client.updatePrice(priceUpdate.value);
-        config.logger?.log("\nUpdate completed successfully!");
-    } else {
-        config.logger?.log("\nUpdate skipped because no new price was available.");
-    }
+    await client.updatePrice({ signal: config.signal });
 }

src/hermes-client.ts

@@ -14,7 +14,7 @@
 import { SigningCosmWasmClient } from "@cosmjs/cosmwasm-stargate";
 import { DirectSecp256k1HdWallet, DirectSecp256k1Wallet, type OfflineDirectSigner } from "@cosmjs/proto-signing";
 import { GasPrice } from "@cosmjs/stargate";
-import { priceUpdateCounter, priceStaleness } from "./metrics.ts";
+import { priceUpdateCounter, blockchainPriceStaleness } from "./metrics.ts";
 import { latestValue } from "./price-stream/latest-value/latest-value.ts";
 import { PriceUpdateConfirmed } from "./price-update/price-update-confirmed/price-update-confirmed.ts";
 import type { Logger, PriceProducerFactory, PriceUpdate, PriceUpdater, PythPriceData } from "./types.ts";
@@ -153,23 +153,6 @@ interface OracleParamsResponse {
 
 const DEFAULT_PRICE_DEVIATION_TOLERANCE: Required<HermesConfig>["priceDeviationTolerance"] = { type: "absolute", value: 0 };
 
-export type ErrorCode = "insufficient_balance" | "timeout" | "connection_issue" | "unknown";
-
-export function classifyError(error: unknown): ErrorCode {
-    const message = error instanceof Error ? error.message : "";
-
-    if (/insufficient funds|insufficient fee/i.test(message)) {
-        return "insufficient_balance";
-    }
-    if (/timeout|ETIMEDOUT/i.test(message)) {
-        return "timeout";
-    }
-    if (/ECONNREFUSED|ECONNRESET|ENOTFOUND/i.test(message)) {
-        return "connection_issue";
-    }
-    return "unknown";
-}
-
 export class HermesClient {
     #cosmClient?: SigningCosmWasmClient;
     #wallet?: OfflineDirectSigner;
@@ -434,14 +417,21 @@ export class HermesClient {
             throw new Error("Client not initialized");
         }
 
+        if (this.#insufficientBalanceCooldownUntil !== null) {
+            if (Date.now() < this.#insufficientBalanceCooldownUntil) {
+                this.#logger.warn("Skipping price update: insufficient balance cooldown active");
+                return;
+            }
+            this.#logger.log("Insufficient balance cooldown expired, retrying...");
+        }
+
         const startTime = performance.now();
 
         try {
             const currentPrice = await this.queryCurrentPrice();
 
-            // Record staleness: how far behind on-chain is from Pyth
             const staleness = priceUpdate.priceData.price.publish_time - currentPrice.publish_time;
-            priceStaleness.record(staleness);
+            blockchainPriceStaleness.record(staleness);
 
             if (this.#canIgnorePriceUpdate(priceUpdate.priceData, currentPrice)) {
                 priceUpdateCounter.add(1, { result: "skipped" });
@@ -450,7 +440,6 @@ export class HermesClient {
 
             const config = await this.queryConfig();
 
-            // Execute update
             this.#logger.log("Submitting VAA to Pyth contract...");
             this.#logger.log(`  Wormhole contract: ${config.wormhole_contract}`);
             this.#priceUpdater ??= new PriceUpdateConfirmed(this.#getCosmClient());
@@ -469,6 +458,7 @@ export class HermesClient {
             this.#logger.log(`  New price: ${price.price} (expo: ${price.expo})`);
             priceUpdateCounter.add(1, { result: "success" });
             this.#lastPriceUpdateAt = new Date().toISOString();
+            this.#insufficientBalanceCooldownUntil = null;
         } catch (error) {
             // SEC-04: Sanitize error messages to prevent information leakage
             const errorCode = classifyError(error);
@@ -585,15 +575,7 @@ export class HermesClient {
             const updatePrices = async () => {
                 for await (const priceUpdate of priceUpdates) {
                     try {
-                        if (this.#insufficientBalanceCooldownUntil !== null) {
-                            if (Date.now() < this.#insufficientBalanceCooldownUntil) {
-                                this.#logger.warn("Skipping price update: insufficient balance cooldown active");
-                                continue;
-                            }
-                            this.#logger.log("Insufficient balance cooldown expired, retrying...");
-                        }
                         await this.#updatePrice(priceUpdate);
-                        this.#insufficientBalanceCooldownUntil = null;
                     } catch (error) {
                         this.#logger.error("Error in scheduled update:", error);
                     }
@@ -637,7 +619,23 @@ export class HermesClient {
     }
 }
 
-// Export types for external use
 export type {
     ConfigResponse, DataSourceResponse, OracleParamsResponse, PriceFeedIdResponse, PriceFeedResponse, PriceResponse, RefreshOracleParamsMsg, TransferAdminMsg, UpdateFeeMsg,
 };
+
+export type ErrorCode = "insufficient_balance" | "timeout" | "connection_issue" | "unknown";
+
+export function classifyError(error: unknown): ErrorCode {
+    const message = error instanceof Error ? error.message : "";
+
+    if (/insufficient funds|insufficient fee/i.test(message)) {
+        return "insufficient_balance";
+    }
+    if (/timeout|ETIMEDOUT/i.test(message)) {
+        return "timeout";
+    }
+    if (/ECONNREFUSED|ECONNRESET|ENOTFOUND/i.test(message)) {
+        return "connection_issue";
+    }
+    return "unknown";
+}

src/metrics.ts

@@ -8,7 +8,7 @@ export const hermesFetchDuration = meter.createHistogram("hermes.price_fetch_dur
     description: "Duration of Hermes API price fetch in milliseconds",
     unit: "ms",
 });
-export const priceStaleness = meter.createGauge("hermes.price_staleness", {
+export const blockchainPriceStaleness = meter.createGauge("hermes.chain_price_staleness", {
     description: "How far behind the on-chain price is from the latest Pyth price",
     unit: "s",
 });

src/price-stream/polling-price-stream/polling-price-stream.ts

@@ -27,8 +27,12 @@ export async function *pollPriceStream(options: PollPriceStreamOptions): AsyncGe
     while (!options.signal?.aborted) {
         const fetchStart = performance.now();
         response = undefined;
+        status = 0;
         try {
-            response = await fetch(`${options.baseUrl}/v2/updates/price/latest?${params.toString()}`);
+            const timeoutSignal = AbortSignal.timeout(10_000);
+            response = await fetch(`${options.baseUrl}/v2/updates/price/latest?${params.toString()}`, {
+                signal: options.signal ? AbortSignal.any([options.signal, timeoutSignal]) : timeoutSignal,
+            });
             status = response.status;
         } catch (error) {
             if (error instanceof Error && (error.name === "AbortError" || error.message === "AbortError")) {

src/price-stream/price-sse-stream/price-sse-stream.test.ts

@@ -75,9 +75,9 @@ describe(priceSSEStream.name, () => {
         const gen = priceSSEStream(options);
         await gen.next();
 
-        expect(fetchMock.mock.calls[0][1]).toEqual(
-            expect.objectContaining({ signal: controller.signal }),
-        );
+        const fetchOptions = fetchMock.mock.calls[0][1] as { signal: AbortSignal };
+        expect(fetchOptions.signal).toBeInstanceOf(AbortSignal);
+        expect(fetchOptions.signal.aborted).toBe(false);
     });
 
     it("throws on non-ok HTTP response after max retries", async () => {

src/price-stream/price-sse-stream/price-sse-stream.ts

@@ -31,9 +31,10 @@ export async function *priceSSEStream(options: PriceSSEStreamOptions): AsyncGene
             }
 
             options.logger?.log(`Connecting to Hermes price stream at ${options.baseUrl}${lastEventId ? ` (Last-Event-ID: ${lastEventId})` : ""}...`);
+            const timeoutSignal = AbortSignal.timeout(10_000);
             const response = await fetch(`${options.baseUrl}/v2/updates/price/stream?${params.toString()}`, {
                 headers,
-                signal: options.signal,
+                signal: options.signal ? AbortSignal.any([options.signal, timeoutSignal]) : timeoutSignal,
             });
             if (!response.ok) {
                 const statusText = response.status ? ` (HTTP ${response.status})` : "";