fixes

Author: merolish

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

@@ -5,6 +5,7 @@
 
 STALE_TIMEOUT_SECONDS = 5
 USER_LIMIT_INTERVAL_SECONDS = 1800
+HYPERLIQUID_WS_PING_INTERVAL_SECONDS = 20
 
 
 class KMSConfig(BaseModel):
@@ -39,6 +40,7 @@ class HyperliquidConfig(BaseModel):
     publish_timeout: float
     enable_publish: bool
     user_limit_interval: int = USER_LIMIT_INTERVAL_SECONDS
+    ws_ping_interval: int = HYPERLIQUID_WS_PING_INTERVAL_SECONDS
 
     @model_validator(mode="after")
     def set_default_urls(self):

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

@@ -19,6 +19,11 @@
 class HLChannel(StrEnum):
     CHANNEL_ACTIVE_ASSET_CTX = "activeAssetCtx"
     CHANNEL_ALL_MIDS = "allMids"
+    CHANNEL_SUBSCRIPTION_RESPONSE = "subscriptionResponse"
+    CHANNEL_PONG = "pong"
+    CHANNEL_ERROR = "error"
+
+DATA_CHANNELS = [HLChannel.CHANNEL_ACTIVE_ASSET_CTX, HLChannel.CHANNEL_ALL_MIDS]
 
 
 class HyperliquidListener:
@@ -33,6 +38,7 @@ 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
 
     def get_subscribe_request(self, asset):
         return {
@@ -66,7 +72,10 @@ 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)
 
-            channel_last_message_timestamp = {channel: time.time() for channel in HLChannel}
+            now = time.time()
+            channel_last_message_timestamp = {channel: now for channel in HLChannel}
+            last_ping_timestamp = now
+
             # listen for updates
             while True:
                 try:
@@ -76,29 +85,37 @@ async def subscribe_single_inner(self, url):
                     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 == 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 HLChannel:
+                    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:
                     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:
-                    logger.warning("HyperliquidListener: Connection closed, 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)

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

@@ -48,7 +48,7 @@ def _init_metrics(self):
             name="hip_3_relayer_price_config",
             description="Price source config",
         )
-        # labels: user
+        # 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",

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

@@ -15,6 +15,7 @@ def __init__(self, config: Config, metrics: Metrics, address: str):
         self.address = address.lower()
         self.metrics = metrics
         self.interval = config.hyperliquid.user_limit_interval
+        self.dex = config.hyperliquid.market_name
 
         base_url = constants.TESTNET_API_URL if config.hyperliquid.use_testnet else constants.MAINNET_API_URL
         self.info = Info(base_url=base_url, skip_ws=True, meta=Meta(universe=[]), spot_meta=SpotMeta(universe=[], tokens=[]))
@@ -27,7 +28,7 @@ async def run(self):
                 logger.debug("userRateLimit response: {}", response)
                 balance = response["nRequestsSurplus"] - response["nRequestsCap"] - response["nRequestsUsed"]
                 logger.debug("userRateLimit user: {} balance: {}", self.address, balance)
-                self.metrics.user_request_balance.set(balance, {"user": self.address})
+                self.metrics.user_request_balance.set(balance, {"dex": self.dex, "user": self.address})
             except Exception as e:
                 logger.error("userRateLimit query failed: {}", e)
             await asyncio.sleep(self.interval)