Check for staleness in each Hyperliquid websocket channel (#3289)

Author: merolish

apps/hip-3-pusher/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "hip-3-pusher"
-version = "0.2.3"
+version = "0.2.4"
 description = "Hyperliquid HIP-3 market oracle pusher"
 readme = "README.md"
 requires-python = "==3.13.*"

apps/hip-3-pusher/src/pusher/config.py

@@ -3,7 +3,14 @@
 from typing import Optional
 from typing import Literal
 
+# Interval of time after which we'll cycle websocket connections
 STALE_TIMEOUT_SECONDS = 5
+# This is the interval to call userRateLimit. Low-frequency as it's just for long-term metrics.
+USER_LIMIT_INTERVAL_SECONDS = 1800
+# HL has an application-level ping-pong that should be handled on the order of a minute.
+HYPERLIQUID_WS_PING_INTERVAL_SECONDS = 20
+# Number of websocket failures before we crash/restart the app.
+DEFAULT_STOP_AFTER_ATTEMPT = 20
 
 
 class KMSConfig(BaseModel):
@@ -20,11 +27,13 @@ class LazerConfig(BaseModel):
     lazer_urls: list[str]
     lazer_api_key: str
     feed_ids: list[int]
+    stop_after_attempt: int = DEFAULT_STOP_AFTER_ATTEMPT
 
 
 class HermesConfig(BaseModel):
     hermes_urls: list[str]
     feed_ids: list[str]
+    stop_after_attempt: int = DEFAULT_STOP_AFTER_ATTEMPT
 
 
 class HyperliquidConfig(BaseModel):
@@ -37,6 +46,9 @@ class HyperliquidConfig(BaseModel):
     publish_interval: float
     publish_timeout: float
     enable_publish: bool
+    user_limit_interval: int = USER_LIMIT_INTERVAL_SECONDS
+    ws_ping_interval: int = HYPERLIQUID_WS_PING_INTERVAL_SECONDS
+    stop_after_attempt: int = DEFAULT_STOP_AFTER_ATTEMPT
 
     @model_validator(mode="after")
     def set_default_urls(self):

apps/hip-3-pusher/src/pusher/hermes_listener.py

@@ -1,9 +1,8 @@
 import asyncio
 import json
 from loguru import logger
-import time
 import websockets
-from tenacity import retry, retry_if_exception_type, wait_exponential
+from tenacity import retry, retry_if_exception_type, wait_fixed, stop_after_attempt
 
 from pusher.config import Config, STALE_TIMEOUT_SECONDS
 from pusher.exception import StaleConnectionError
@@ -18,6 +17,7 @@ def __init__(self, config: Config, hermes_state: PriceSourceState):
         self.hermes_urls = config.hermes.hermes_urls
         self.feed_ids = config.hermes.feed_ids
         self.hermes_state = hermes_state
+        self.stop_after_attempt = config.hermes.stop_after_attempt
 
     def get_subscribe_request(self):
         return {
@@ -36,13 +36,19 @@ async def subscribe_all(self):
 
         await asyncio.gather(*(self.subscribe_single(url) for url in self.hermes_urls))
 
-    @retry(
-        retry=retry_if_exception_type((StaleConnectionError, websockets.ConnectionClosed)),
-        wait=wait_exponential(multiplier=1, min=1, max=10),
-        reraise=True,
-    )
     async def subscribe_single(self, url):
-        return await self.subscribe_single_inner(url)
+        logger.info("Starting Hermes listener loop: {}", url)
+
+        @retry(
+            retry=retry_if_exception_type(Exception),
+            wait=wait_fixed(1),
+            stop=stop_after_attempt(self.stop_after_attempt),
+            reraise=True,
+        )
+        async def _run():
+            return await self.subscribe_single_inner(url)
+
+        return await _run()
 
     async def subscribe_single_inner(self, url):
         async with websockets.connect(url) as ws:
@@ -58,13 +64,15 @@ async def subscribe_single_inner(self, url):
                     data = json.loads(message)
                     self.parse_hermes_message(data)
                 except asyncio.TimeoutError:
+                    logger.warning("HermesListener: No messages in {} seconds, reconnecting...", STALE_TIMEOUT_SECONDS)
                     raise StaleConnectionError(f"No messages in {STALE_TIMEOUT_SECONDS} seconds, reconnecting")
                 except websockets.ConnectionClosed:
+                    logger.warning("HermesListener: Connection closed, reconnecting...")
                     raise
                 except json.JSONDecodeError as e:
-                    logger.error("Failed to decode JSON message: {}", e)
+                    logger.exception("Failed to decode JSON message: {}", repr(e))
                 except Exception as e:
-                    logger.error("Unexpected exception: {}", e)
+                    logger.exception("Unexpected exception: {}", repr(e))
 
     def parse_hermes_message(self, data):
         """
@@ -83,7 +91,6 @@ def parse_hermes_message(self, data):
             expo = price_object["expo"]
             publish_time = price_object["publish_time"]
             logger.debug("Hermes update: {} {} {} {}", id, price, expo, publish_time)
-            now = time.time()
-            self.hermes_state.put(id, PriceUpdate(price, now))
+            self.hermes_state.put(id, PriceUpdate(price, publish_time))
         except Exception as e:
-            logger.error("parse_hermes_message error: {}", e)
+            logger.exception("parse_hermes_message error: {}", repr(e))

apps/hip-3-pusher/src/pusher/hyperliquid_listener.py

@@ -1,8 +1,10 @@
 import asyncio
 import json
+from enum import StrEnum
+
 import websockets
 from loguru import logger
-from tenacity import retry, retry_if_exception_type, wait_exponential
+from tenacity import retry, retry_if_exception_type, wait_fixed, stop_after_attempt
 import time
 
 from pusher.config import Config, STALE_TIMEOUT_SECONDS
@@ -14,6 +16,22 @@
 HYPERLIQUID_MAINNET_WS_URL = "wss://api.hyperliquid.xyz/ws"
 HYPERLIQUID_TESTNET_WS_URL = "wss://api.hyperliquid-testnet.xyz/ws"
 
+class HLChannel(StrEnum):
+    """ Hyperliquid websocket subscription channels. See https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions """
+
+    # activeAssetCtx includes oracle and mark price for perps (either main HyperCore or HIP-3)
+    CHANNEL_ACTIVE_ASSET_CTX = "activeAssetCtx"
+    # HL market mid price
+    CHANNEL_ALL_MIDS = "allMids"
+    # either subscription ack or error
+    CHANNEL_SUBSCRIPTION_RESPONSE = "subscriptionResponse"
+    # application-level ping response
+    CHANNEL_PONG = "pong"
+    # error response
+    CHANNEL_ERROR = "error"
+
+DATA_CHANNELS = [HLChannel.CHANNEL_ACTIVE_ASSET_CTX, HLChannel.CHANNEL_ALL_MIDS]
+
 
 class HyperliquidListener:
     """
@@ -27,6 +45,8 @@ def __init__(self, config: Config, hl_oracle_state: PriceSourceState, hl_mark_st
         self.hl_oracle_state = hl_oracle_state
         self.hl_mark_state = hl_mark_state
         self.hl_mid_state = hl_mid_state
+        self.ws_ping_interval = config.hyperliquid.ws_ping_interval
+        self.stop_after_attempt = config.hyperliquid.stop_after_attempt
 
     def get_subscribe_request(self, asset):
         return {
@@ -37,13 +57,19 @@ def get_subscribe_request(self, asset):
     async def subscribe_all(self):
         await asyncio.gather(*(self.subscribe_single(hyperliquid_ws_url) for hyperliquid_ws_url in self.hyperliquid_ws_urls))
 
-    @retry(
-        retry=retry_if_exception_type((StaleConnectionError, websockets.ConnectionClosed)),
-        wait=wait_exponential(multiplier=1, min=1, max=10),
-        reraise=True,
-    )
     async def subscribe_single(self, url):
-        return await self.subscribe_single_inner(url)
+        logger.info("Starting Hyperliquid listener loop: {}", url)
+
+        @retry(
+            retry=retry_if_exception_type(Exception),
+            wait=wait_fixed(1),
+            stop=stop_after_attempt(self.stop_after_attempt),
+            reraise=True,
+        )
+        async def _run():
+            return await self.subscribe_single_inner(url)
+
+        return await _run()
 
     async def subscribe_single_inner(self, url):
         async with websockets.connect(url) as ws:
@@ -59,48 +85,69 @@ async def subscribe_single_inner(self, url):
             await ws.send(json.dumps(subscribe_all_mids_request))
             logger.info("Sent subscribe request for allMids for dex: {} to {}", self.market_name, url)
 
+            now = time.time()
+            channel_last_message_timestamp = {channel: now for channel in HLChannel}
+            last_ping_timestamp = now
+
             # listen for updates
             while True:
                 try:
                     message = await asyncio.wait_for(ws.recv(), timeout=STALE_TIMEOUT_SECONDS)
                     data = json.loads(message)
                     channel = data.get("channel", None)
+                    now = time.time()
                     if not channel:
                         logger.error("No channel in message: {}", data)
-                    elif channel == "subscriptionResponse":
-                        logger.debug("Received subscription response: {}", data)
-                    elif channel == "error":
+                    elif channel == HLChannel.CHANNEL_SUBSCRIPTION_RESPONSE:
+                        logger.info("Received subscription response: {}", data)
+                    elif channel == HLChannel.CHANNEL_ERROR:
                         logger.error("Received Hyperliquid error response: {}", data)
-                    elif channel == "activeAssetCtx":
-                        self.parse_hyperliquid_active_asset_ctx_update(data)
-                    elif channel == "allMids":
-                        self.parse_hyperliquid_all_mids_update(data)
+                    elif channel == HLChannel.CHANNEL_ACTIVE_ASSET_CTX:
+                        self.parse_hyperliquid_active_asset_ctx_update(data, now)
+                        channel_last_message_timestamp[channel] = now
+                    elif channel == HLChannel.CHANNEL_ALL_MIDS:
+                        self.parse_hyperliquid_all_mids_update(data, now)
+                        channel_last_message_timestamp[channel] = now
+                    elif channel == HLChannel.CHANNEL_PONG:
+                        logger.debug("Received pong")
                     else:
                         logger.error("Received unknown channel: {}", channel)
+
+                    # check for stale channels
+                    for channel in DATA_CHANNELS:
+                        if now - channel_last_message_timestamp[channel] > STALE_TIMEOUT_SECONDS:
+                            logger.warning("HyperliquidLister: no messages in channel {} stale in {} seconds; reconnecting...", channel, STALE_TIMEOUT_SECONDS)
+                            raise StaleConnectionError(f"No messages in channel {channel} in {STALE_TIMEOUT_SECONDS} seconds, reconnecting...")
+
+                    # ping if we need to
+                    if now - last_ping_timestamp > self.ws_ping_interval:
+                        await ws.send(json.dumps({"method": "ping"}))
+                        last_ping_timestamp = now
                 except asyncio.TimeoutError:
-                    raise StaleConnectionError(f"No messages in {STALE_TIMEOUT_SECONDS} seconds, reconnecting...")
-                except websockets.ConnectionClosed:
+                    logger.warning("HyperliquidListener: No messages overall in {} seconds, reconnecting...", STALE_TIMEOUT_SECONDS)
+                    raise StaleConnectionError(f"No messages overall in {STALE_TIMEOUT_SECONDS} seconds, reconnecting...")
+                except websockets.ConnectionClosed as e:
+                    rc, rr = e.rcvd.code if e.rcvd else None, e.rcvd.reason if e.rcvd else None
+                    logger.warning("HyperliquidListener: Websocket connection closed (code={} reason={}); reconnecting...", rc, rr)
                     raise
                 except json.JSONDecodeError as e:
-                    logger.error("Failed to decode JSON message: {} error: {}", message, e)
+                    logger.exception("Failed to decode JSON message: {} error: {}", message, repr(e))
                 except Exception as e:
-                    logger.error("Unexpected exception: {}", e)
+                    logger.exception("Unexpected exception: {}", repr(e))
 
-    def parse_hyperliquid_active_asset_ctx_update(self, message):
+    def parse_hyperliquid_active_asset_ctx_update(self, message, now):
         try:
             ctx = message["data"]["ctx"]
             symbol = message["data"]["coin"]
-            now = time.time()
             self.hl_oracle_state.put(symbol, PriceUpdate(ctx["oraclePx"], now))
             self.hl_mark_state.put(symbol, PriceUpdate(ctx["markPx"], now))
             logger.debug("activeAssetCtx symbol: {} oraclePx: {} markPx: {}", symbol, ctx["oraclePx"], ctx["markPx"])
         except Exception as e:
-            logger.error("parse_hyperliquid_active_asset_ctx_update error: message: {} e: {}", message, e)
+            logger.exception("parse_hyperliquid_active_asset_ctx_update error: message: {} e: {}", message, repr(e))
 
-    def parse_hyperliquid_all_mids_update(self, message):
+    def parse_hyperliquid_all_mids_update(self, message, now):
         try:
             mids = message["data"]["mids"]
-            now = time.time()
             for mid in mids:
                 self.hl_mid_state.put(mid, PriceUpdate(mids[mid], now))
             logger.debug("allMids: {}", mids)

apps/hip-3-pusher/src/pusher/lazer_listener.py

@@ -1,9 +1,8 @@
 import asyncio
 import json
 from loguru import logger
-import time
 import websockets
-from tenacity import retry, retry_if_exception_type, wait_exponential
+from tenacity import retry, retry_if_exception_type, wait_fixed, stop_after_attempt
 
 from pusher.config import Config, STALE_TIMEOUT_SECONDS
 from pusher.exception import StaleConnectionError
@@ -19,6 +18,7 @@ def __init__(self, config: Config, lazer_state: PriceSourceState):
         self.api_key = config.lazer.lazer_api_key
         self.feed_ids = config.lazer.feed_ids
         self.lazer_state = lazer_state
+        self.stop_after_attempt = config.lazer.stop_after_attempt
 
     def get_subscribe_request(self, subscription_id: int):
         return {
@@ -40,13 +40,19 @@ async def subscribe_all(self):
 
         await asyncio.gather(*(self.subscribe_single(router_url) for router_url in self.lazer_urls))
 
-    @retry(
-        retry=retry_if_exception_type((StaleConnectionError, websockets.ConnectionClosed)),
-        wait=wait_exponential(multiplier=1, min=1, max=10),
-        reraise=True,
-    )
     async def subscribe_single(self, router_url):
-        return await self.subscribe_single_inner(router_url)
+        logger.info("Starting Lazer listener loop: {}", router_url)
+
+        @retry(
+            retry=retry_if_exception_type(Exception),
+            wait=wait_fixed(1),
+            stop=stop_after_attempt(self.stop_after_attempt),
+            reraise=True,
+        )
+        async def _run():
+            return await self.subscribe_single_inner(router_url)
+
+        return await _run()
 
     async def subscribe_single_inner(self, router_url):
         headers = {
@@ -66,13 +72,15 @@ async def subscribe_single_inner(self, router_url):
                     data = json.loads(message)
                     self.parse_lazer_message(data)
                 except asyncio.TimeoutError:
+                    logger.warning("LazerListener: No messages in {} seconds, reconnecting...", STALE_TIMEOUT_SECONDS)
                     raise StaleConnectionError(f"No messages in {STALE_TIMEOUT_SECONDS} seconds, reconnecting")
                 except websockets.ConnectionClosed:
+                    logger.warning("LazerListener: Connection closed, reconnecting...")
                     raise
                 except json.JSONDecodeError as e:
-                    logger.error("Failed to decode JSON message: {}", e)
+                    logger.exception("Failed to decode JSON message: {}", repr(e))
                 except Exception as e:
-                    logger.error("Unexpected exception: {}", e)
+                    logger.exception("Unexpected exception: {}", repr(e))
 
     def parse_lazer_message(self, data):
         """
@@ -85,14 +93,15 @@ def parse_lazer_message(self, data):
             if data.get("type", "") != "streamUpdated":
                 return
             price_feeds = data["parsed"]["priceFeeds"]
-            logger.debug("price_feeds: {}", price_feeds)
-            now = time.time()
+            # timestampUs is in micros, this is scaled to unix seconds the same as time.time()
+            timestamp_seconds = int(data["parsed"]["timestampUs"]) / 1_000_000.0
+            logger.debug("price_feeds: {} timestamp: {}", price_feeds, timestamp_seconds)
             for feed_update in price_feeds:
                 feed_id = feed_update.get("priceFeedId", None)
                 price = feed_update.get("price", None)
                 if feed_id is None or price is None:
                     continue
                 else:
-                    self.lazer_state.put(feed_id, PriceUpdate(price, now))
+                    self.lazer_state.put(feed_id, PriceUpdate(price, timestamp_seconds))
         except Exception as e:
-            logger.error("parse_lazer_message error: {}", e)
+            logger.exception("parse_lazer_message error: {}", repr(e))

apps/hip-3-pusher/src/pusher/main.py

@@ -13,6 +13,7 @@
 from pusher.price_state import PriceState
 from pusher.publisher import Publisher
 from pusher.metrics import Metrics
+from pusher.user_limit_listener import UserLimitListener
 
 
 def load_config():
@@ -52,13 +53,15 @@ async def main():
     lazer_listener = LazerListener(config, price_state.lazer_state)
     hermes_listener = HermesListener(config, price_state.hermes_state)
     seda_listener = SedaListener(config, price_state.seda_state)
+    user_limit_listener = UserLimitListener(config, metrics, publisher.user_limit_address)
 
     await asyncio.gather(
         publisher.run(),
         hyperliquid_listener.subscribe_all(),
         lazer_listener.subscribe_all(),
         hermes_listener.subscribe_all(),
         seda_listener.run(),
+        user_limit_listener.run(),
     )
     logger.info("Exiting hip-3-pusher..")
 

apps/hip-3-pusher/src/pusher/metrics.py

@@ -48,6 +48,11 @@ def _init_metrics(self):
             name="hip_3_relayer_price_config",
             description="Price source config",
         )
+        # labels: dex, user
+        self.user_request_balance = self.meter.create_gauge(
+            name="hip_3_relayer_user_request_balance",
+            description="Number of update requests left before rate limit",
+        )
 
     def set_price_configs(self, dex: str, price_config: PriceConfig):
         self._set_price_config_type(dex, price_config.oracle, "oracle")