hyperliquid: unWatchOHLCV ack rejects a subscription that was re-armed in the meantime (watch -> unwatch -> watch raises UnsubscribeError)

#!/usr/bin/env python3
"""Minimal reproducer for the ccxt.pro Hyperliquid unsubscribe/re-subscribe race.

The sequence reproduced here is the one freqtrade's websocket layer produces in
normal operation (freqtrade/exchange/exchange_ws.py):

    watch_ohlcv(...)            # task A is running
    -> task A ends              # e.g. cleanup_expired() dropped the pair
    -> un_watch_ohlcv(...)      # unsubscribe is sent (asynchronous)
    -> watch_ohlcv(...)         # same pair/timeframe is re-scheduled right away

Observed on current ccxt (4.5.76 and 4.5.78): the re-subscription dies with
`UnsubscribeError` and never delivers a candle, because the unsubscribe ack
rejects the future that was armed by the re-subscription.

Steps:
  1. control: fresh subscription                       -> expect candles
  2. control: unsubscribe fully acked, then re-subscribe -> expect candles
  3. bug:     unsubscribe in flight + immediate re-subscribe -> expect UnsubscribeError
  4. recovery: subscribe once more                       -> expect candles

Exit code 0 = bug reproduced, 1 = not reproduced.
"""

import asyncio
import sys
import time

import ccxt
import ccxt.pro as ccxtpro

TIMEOUT = 30.0
CONTROL_PAIR = "BTC/USDC:USDC"
CONTROL2_PAIR = "ETH/USDC:USDC"
BUG_PAIR = "XRP/USDC:USDC"
TIMEFRAME = "5m"


async def watch_once(ex, pair, timeframe, label, results, timeout=TIMEOUT):
    """Await one watch_ohlcv call and record the outcome."""
    start = time.time()
    try:
        data = await asyncio.wait_for(ex.watch_ohlcv(pair, timeframe), timeout)
        results.append((label, "OK", f"{len(data)} candle(s)", time.time() - start))
        return True
    except asyncio.TimeoutError:
        results.append((label, "TIMEOUT", f"no data within {timeout:.0f}s", time.time() - start))
    except asyncio.CancelledError:
        results.append((label, "CANCELLED", "", time.time() - start))
    except Exception as e:  # noqa: BLE001 - report whatever comes back
        results.append((label, type(e).__name__, str(e)[:60], time.time() - start))
    return False


async def main():
    print(f"python : {sys.version.split()[0]}")
    print(f"ccxt   : {ccxt.__version__}")
    print(f"exchange: hyperliquid (USDC perps), timeframe {TIMEFRAME}")
    print()

    ex = ccxtpro.hyperliquid({"options": {"defaultType": "swap"}, "enableRateLimit": True})
    await ex.load_markets()
    print(f"markets: {len(ex.markets)} loaded")
    print()

    results = []
    try:
        # 1 - control: a fresh subscription works
        await watch_once(ex, CONTROL_PAIR, TIMEFRAME, "1 control: fresh subscribe", results)

        # 2 - control: unsubscribe fully acked, then re-subscribe
        await watch_once(ex, CONTROL2_PAIR, TIMEFRAME, "2 control: watch", results)
        acked = False
        try:
            await asyncio.wait_for(ex.un_watch_ohlcv(CONTROL2_PAIR, TIMEFRAME), TIMEOUT)
            acked = True
        except Exception as e:  # noqa: BLE001
            results.append(("2 control: unwatch", type(e).__name__, str(e)[:60], 0.0))
        if acked:
            results.append(("2 control: unwatch", "ACKED", "", 0.0))
            await watch_once(
                ex, CONTROL2_PAIR, TIMEFRAME, "2 control: re-subscribe", results
            )

        # 3 - bug: unsubscribe in flight + immediate re-subscribe
        await watch_once(ex, BUG_PAIR, TIMEFRAME, "3 bug: first watch", results)
        unwatch_task = asyncio.create_task(ex.un_watch_ohlcv(BUG_PAIR, TIMEFRAME))
        await asyncio.sleep(0.1)  # the ack has not arrived yet - we re-subscribe now
        await watch_once(ex, BUG_PAIR, TIMEFRAME, "3 bug: re-subscribe", results)
        unwatch_task.cancel()

        # 4 - recovery: subscribe once more
        await watch_once(ex, BUG_PAIR, TIMEFRAME, "4 recovery: subscribe again", results)
    finally:
        try:
            await ex.close()
        except Exception:  # noqa: BLE001
            pass

    print(f"{'step':<32} {'result':<18} {'detail':<24} {'secs':>5}")
    print("-" * 84)
    for label, outcome, detail, secs in results:
        print(f"{label:<32} {outcome:<18} {detail:<24} {secs:>5.1f}")
    print()

    bug_step = next((r for r in results if r[0] == "3 bug: re-subscribe"), None)
    reproduced = bool(bug_step and bug_step[1] == "UnsubscribeError")
    if reproduced:
        print("BUG REPRODUCED: the re-subscription was rejected with UnsubscribeError.")
        return 0
    print(f"NOT REPRODUCED: '3 bug: re-subscribe' returned {bug_step[1] if bug_step else 'n/a'}.")
    return 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

python : 3.14.7
ccxt   : 4.5.76
exchange: hyperliquid (USDC perps), timeframe 5m

markets: 812 loaded

step                             result             detail                    secs
------------------------------------------------------------------------------------
1 control: fresh subscribe       OK                 1 candle(s)                0.9
2 control: watch                 OK                 1 candle(s)                2.7
2 control: unwatch               ACKED                                         0.0
2 control: re-subscribe          OK                 1 candle(s)                0.8
3 bug: first watch               OK                 1 candle(s)                9.0
3 bug: re-subscribe              UnsubscribeError   hyperliquid candles:5m:XRP/USDC:USDC   0.1
4 recovery: subscribe again      OK                 1 candle(s)                1.3

BUG REPRODUCED: the re-subscription was rejected with UnsubscribeError.

handleOHLCVUnsubscription (client: Client, subscription: Dict) {
    const coin = this.safeString (subscription, 'coin');
    const marketId = this.coinToMarketId (coin);
    const symbol = this.safeSymbol (marketId);
    const interval = this.safeString (subscription, 'interval');
    const timeframe = this.findTimeframe (interval);
    const subMessageHash = 'candles:' + timeframe + ':' + symbol;
    const messageHash = 'unsubscribe:' + subMessageHash;
    const rearmed = (subMessageHash in client.futures);  // (re)subscribed while unsubscribe was in flight
    if (rearmed) {
        // clean bookkeeping without rejecting the (new) waiter
        if (messageHash in client.subscriptions) { delete client.subscriptions[messageHash]; }
        if (subMessageHash in client.subscriptions) { delete client.subscriptions[subMessageHash]; }
        client.resolve (true, messageHash);
        this.spawn (this.watchOHLCV, symbol, timeframe);  // re-send subscribe (subscription marker was cleared above)
    } else {
        this.cleanUnsubscription (client, subMessageHash, messageHash);
    }
    // ... ohlcvs cleanup as before
}