Session Management and Reconnection

Session persistence, reconnection, and automatic replay for reliable WebSocket connections

Sessions enable reliable WebSocket connections with automatic replay of missed data on reconnect.

Overview

When you connect to the WebSocket API, you receive a sessionId that persists your subscriptions and tracks your position in the data stream. On reconnect, provide this session ID to:

Restore subscriptions - No need to re-subscribe
Receive replay - Automatically get missed data (up to 30 seconds)
Resume seamlessly - Continue from where you left off

Key Concepts

Sequence Numbers (`seq`)

Every live data message includes a seq field - a monotonically increasing integer for client use only:

{
  "type": "allFills",
  "seq": 42,
  "fills": [...]
}

Purpose: Detect gaps within a single connection. If you receive seq: 5 then seq: 7, you know seq: 6 was lost.

Key behaviors:

Resets to 1 on every new connection (including reconnects)
Per-subscription channel (each subscription has its own sequence)
Client-side only - server doesn't track your last received seq

Cursor

The cursor tracks your position in the data stream, tied to Hyperliquid's block production. The server supports two cursor formats:

Cursor Format

Cursors use a colon-separated format with up to 3 components:

Format

Example

Description

block:timestamp:tx_index

"500:1706123456789:3"

Full precision — most subscription types

block:timestamp

"500:1706123456789"

Block-level — used by l4BookUpdates, fundingRates, and setOracleUpdates

timestamp (number)

1706123456789

Numeric — used by leverage and isolated margin update subscriptions

The tx_index component enables precise sub-block positioning. Within a single block (~70ms), there can be hundreds of events. Without tx_index, resuming from a cursor could miss events or produce duplicates when boundaries fall mid-block.

Treat cursors as opaque strings — always store and return the full cursor exactly as received. The server handles all parsing.

Cursor format: The server always sends cursors in block:timestamp:tx_index format (or block:timestamp for block-level subscriptions). Clients should treat cursors as opaque strings.

Client-side tracking (recommended):

Every live event message includes a cursor field
Save this cursor as a string after successfully processing each message
Provide your cursor on reconnect for accurate replay

{
  "type": "allFills",
  "seq": 42,
  "cursor": "500:1706123456789:3",
  "fills": [...]
}

Server-side tracking (fallback):

The server also tracks a cursor for your session
Persists across disconnects
Used as fallback if you don't provide your own cursor

Important: The server cursor is based on messages sent, not messages received. Due to network issues, OS socket buffering, or client-side delays, you may not have actually received all messages the server sent. For best accuracy, always track and provide your own cursor based on messages you successfully processed.

Manual cursor provision: When subscribing, you can optionally provide a cursor to request replay from a specific point:

{
  "method": "subscribe",
  "subscription": {
    "type": "allFills",
    "cursor": "500:1706123400000:0"
  }
}

This is useful for:

Resuming from a known checkpoint after application restart
Requesting specific historical data within the cache window (30 seconds)

Message Types

Control Messages

Type

Direction

Description

connected

Server→Client

New session created

reconnected

Server→Client

Existing session resumed

subscriptionUpdate

Server→Client

Subscription confirmed

ping

Server→Client

Heartbeat (respond with pong)

pong

Client→Server

Heartbeat response

error

Server→Client

Error occurred

Data Messages (Live Events)

Live event messages have the subscription type as their type field:

{
  "type": "allFills",
  "seq": 42,
  "cursor": "500:1706123456789:3",
  "fills": [
    ["0x742d35...", { "coin": "ETH", "px": "2150.50", ... }]
  ]
}

Field

Type

Description

type

string

Subscription type (e.g., "allFills")

seq

number

Sequence number for gap detection (resets on reconnect)

cursor

string

Position cursor (opaque string, e.g. "500:1706123456789:3")

fills/updates/events/trades/liquidations/data

array

Event data (field name varies by subscription type)

Common data message types: allFills, userFills, userOrderUpdates, liquidationFills, l4BookUpdates, etc.

Replay Messages

Replay messages are distinct from live events - they have "type": "replay" and contain historical data:

{
  "type": "replay",
  "channel": "allFills",
  "replayTimeMs": "500:1706123456789:3",
  "cursor": "500:1706123456789:3",
  "count": 150,
  "chunk": 1,
  "totalChunks": 2,
  "data": [
    ["0x742d35...", { "coin": "ETH", "px": "2150.50", ... }]
  ]
}

Field

Type

Description

type

string

Always "replay"

channel

string

The subscription type (e.g., "allFills")

cursor

string

Per-chunk cursor derived from the last event in this chunk (same format as live cursor). Use this for reconnection.

replayTimeMs

string

Same as cursor. Deprecated — use cursor instead.

count

number

Number of items in this chunk

chunk

number

Current chunk number (1-indexed)

totalChunks

number

Total chunks in this replay

hasGap

boolean

Only present when true. Indicates cache didn't cover entire disconnect period (only meaningful on first chunk). Absent when there is no gap.

data

array

Array of missed events (same format as live data)

Chunking

Large replays are split into multiple chunks to keep message sizes manageable. Chunking occurs when either limit is reached:

Max items: 2,000 items per chunk
Max size: 1 MB per chunk

Use chunk and totalChunks to track replay progress:

Update your cursor from each chunk's cursor as it arrives — this ensures progress is saved even if the connection drops mid-replay
hasGap is only set on the first chunk (chunk: 1)
Process chunks in order as they arrive

Important: Replay messages do NOT have a seq field. Only live event messages have sequences.

Connection Flow

Initial Connection

wss://api.hydromancer.xyz/ws?token=YOUR_API_KEY

Server responds with connected:

{
  "type": "connected",
  "clientId": "abc123-def456-...",
  "sessionId": "789xyz-..."
}

Save the sessionId - you'll need it for reconnection.

Subscribing

{
  "method": "subscribe",
  "subscription": { "type": "allFills" }
}

Server confirms with subscriptionUpdate:

{
  "type": "subscriptionUpdate",
  "subscribed": ["allFills"],
  "failed": []
}

Then live data begins flowing with seq starting at 1.

Reconnection Flow

Step-by-Step Process

Client disconnects (network issue, restart, etc.)
Server marks session disconnected - starts 30-second grace period

Client reconnects with sessionId and cursor:

wss://api.hydromancer.xyz/ws?token=YOUR_API_KEY&sessionId=YOUR_SESSION_ID&cursor=500:1706123456789:3

Server validates session - checks if still within grace period
Server automatically restores subscriptions - no need to re-subscribe

Server sends reconnected:

{
  "type": "reconnected",
  "clientId": "new-client-id-...",
  "sessionId": "789xyz-...",
  "subscriptions": ["allFills"],
  "lastSeq": 0
}

Server sends replay - events since your cursor (up to 30 seconds of data)
Live events resume - with seq starting at 1

If your original connection used liveFormat=chunked-v1, you must include liveFormat=chunked-v1 again on reconnect. liveFormat is connection-level and is not persisted on the session.

Providing Your Cursor on Reconnect

When reconnecting, you can provide your last successfully processed cursor as a query parameter:

wss://api.hydromancer.xyz/ws?token=YOUR_API_KEY&sessionId=YOUR_SESSION_ID&cursor=500:1706123456789:3

If you are using live chunking, include liveFormat=chunked-v1 as well:

wss://api.hydromancer.xyz/ws?token=YOUR_API_KEY&sessionId=YOUR_SESSION_ID&cursor=500:1706123456789:3&liveFormat=chunked-v1

Parameter

Required

Description

token

Yes

Your API key

sessionId

Session ID to resume (omit for new session)

cursor

Last cursor you received (opaque string, e.g. "500:1706123456789:3")

liveFormat

Required on every reconnect if you are using chunked-v1 live message format

Cursor priority:

Client-provided cursor (recommended) - from the cursor query parameter
Server-stored cursor (fallback) - used if you don't provide one

Best Practice: Always track the cursor field from each message you process successfully, and provide it on reconnect. This ensures you receive exactly the data you missed, even if some messages were lost in transit before disconnect.

Automatic Subscription Restore: You do not need to re-subscribe after reconnecting. The server automatically restores all your previous subscriptions and begins sending data immediately. The subscriptions array in the reconnected message confirms which feeds were restored.

Note: lastSeq is always 0 because sequences reset on each connection.

What Triggers Replay

Replay is sent when ALL of these conditions are met:

You reconnect with a valid, non-expired sessionId
The session had active subscriptions before disconnect
There is cached data newer than your cursor
The cache TTL (30 seconds) hasn't fully expired

When Replay is NOT Sent

New connection (no sessionId provided)
Expired session (disconnected > 30 seconds ago)
No cached data for your subscriptions (quiet market)
Your cursor is already current (instant reconnect)

Gap Detection

The hasGap field in replay messages indicates data completeness:

hasGap

Meaning

Action

absent

Complete replay - no data lost

Process normally

true

Some data may be missing

Consider fetching from REST API

A gap occurs when your cursor is older than the oldest cached event - some events aged out before you reconnected.

Replay Response Scenarios

When you reconnect with a session, you'll receive one of these replay response types depending on market activity and your disconnect duration:

Scenario

count

hasGap

data

Meaning

Data, no gap

> 0

absent

Events array

Normal replay - you have all missed events

Data, with gap

> 0

true

Events array

Partial replay - some older events expired from cache

No data, with gap

true

[]

Cache empty and cursor is old - all events expired

No data, no gap

absent

[]

Quiet market - no events occurred during disconnect

Example Responses

Data with no gap (complete replay):

{
  "type": "replay",
  "channel": "allFills",
  "replayTimeMs": "500:1706123456789:3",
  "cursor": "500:1706123456789:3",
  "count": 42,
  "chunk": 1,
  "totalChunks": 1,
  "data": [/* 42 fill events */]
}

Note: hasGap is absent here because there is no gap. The field is only included when true.

Data with gap (partial replay - some events expired):

{
  "type": "replay",
  "channel": "allFills",
  "replayTimeMs": "500:1706123456789:3",
  "cursor": "500:1706123456789:3",
  "count": 15,
  "chunk": 1,
  "totalChunks": 1,
  "hasGap": true,
  "data": [/* 15 fill events - older events expired */]
}

No data with gap (all cached events expired):

{
  "type": "replay",
  "channel": "allFills",
  "replayTimeMs": "500:1706123456789:3",
  "cursor": "500:1706123456789:3",
  "count": 0,
  "chunk": 1,
  "totalChunks": 1,
  "hasGap": true,
  "data": []
}

No data, no gap (quiet market - nothing happened):

{
  "type": "replay",
  "channel": "allFills",
  "replayTimeMs": "500:1706123456789:3",
  "cursor": "500:1706123456789:3",
  "count": 0,
  "chunk": 1,
  "totalChunks": 1,
  "data": []
}

Quiet Markets: A replay with count: 0 and hasGap: false confirms you're fully up-to-date - there simply were no events during your disconnect period. This is common for user-specific subscriptions (e.g., userFills) where activity is infrequent.

Handling Gaps: When hasGap: true, consider fetching historical data from the REST API to fill in missing events. The replay still contains all available cached data - it just doesn't cover your entire disconnect period.

Message Ordering

Important: Live messages may begin arriving before replay is complete. Your client must handle this correctly.

Why This Happens

For minimal latency, the server:

Adds you to live broadcast immediately on reconnect
Sends replay data in parallel

In high-throughput scenarios, a live event may arrive before the final replay chunk.

Recommended Client Pattern

class ReconnectionHandler {
  constructor() {
    this.replayComplete = false;
    this.replayChunksReceived = 0;
    this.replayTotalChunks = 0;
    this.bufferedLive = [];
    this.lastSeen = { time: 0, txIndex: 0 };  // For deduplication
  }

  onMessage(msg) {
    if (msg.type === 'replay') {
      // Process replay data (historical)
      this.processReplayData(msg.data);

      // Update cursor from each chunk for incremental progress
      if (msg.cursor && this.onCursorUpdate) {
        this.onCursorUpdate(msg.cursor);
      }

      this.replayTotalChunks = msg.totalChunks;
      this.replayChunksReceived++;

      // Check if all replay chunks received
      if (this.replayChunksReceived === this.replayTotalChunks) {
        this.replayComplete = true;
        // Now process buffered live messages in order
        this.bufferedLive.forEach(m => this.processLiveData(m));
        this.bufferedLive = [];
      }
    } else if (msg.seq !== undefined) {
      // Live data message (has seq field)
      if (!this.replayComplete) {
        // Buffer until replay is done
        this.bufferedLive.push(msg);
      } else {
        this.processLiveData(msg);
      }
    }
  }
}

Deduplication

Important: Replay and live events may overlap. The same event can appear in both the replay data and the first live messages. Your client must deduplicate events to avoid processing them twice.

Why Overlap Occurs

To ensure no events are lost, the server:

Subscribes you to live broadcasts immediately on reconnect
Generates and sends replay data in parallel

This means events occurring during the brief window between these steps may appear in both streams. This is intentional - it's better to receive a duplicate than to miss an event.

Deduplication Keys

Each event type has a monotonically increasing key composed of (time, txIndex) (or similar). Since these are ordered, you only need to track the last seen key and skip any events at or before it.

Subscription Type

Dedup Key

Fields

allFills, userFills, builderFills, liquidationFills

(time, txIndex)

fill.time, fill.txIndex

userOrderUpdates, builderOrderUpdates

(time, txIndex)

update.time, update.txIndex

allTwapStatusUpdates

(time, txIndex)

update.time, update.txIndex

builderLiquidations, allBuilderLiquidations

(time, txIndex)

fill.time, fill.txIndex

allCompletedTrades, userCompletedTrades, builderCompletedTrades

(closeTime, txIndex)

trade.closeTime, trade.txIndex

userNonFundingLedgerEvents, allUserNonFundingLedgerEvents

(time, txIndex, role)

event.time, event.txIndex, user role

Deduplication Example (Fills)

Since keys are monotonically increasing, track the last processed (time, txIndex) and skip events at or before it:

class DeduplicatingHandler {
  constructor() {
    this.lastSeen = { time: 0, txIndex: 0 };
    // ... other fields
  }

  // Compare (time, txIndex) tuples
  isNewEvent(time, txIndex) {
    if (time > this.lastSeen.time) return true;
    if (time === this.lastSeen.time && txIndex > this.lastSeen.txIndex) return true;
    return false;
  }

  processFill(fill) {
    const time = fill.time;
    const txIndex = fill.txIndex ?? 0;

    // Skip if already seen (duplicate from replay/live overlap)
    if (!this.isNewEvent(time, txIndex)) {
      return;
    }

    // Update last seen
    this.lastSeen = { time, txIndex };

    // Process the fill...
    console.log('New fill:', fill.coin, fill.sz, '@', fill.px);
  }
}

No Memory Cleanup Needed: Since you only track the last seen key (not a set of all seen IDs), there's no memory growth over time.

Session Lifetime

Parameter

Value

Description

Grace period

30 seconds

Time to reconnect before session expires

Redis TTL

60 seconds

Total time session persists in Redis (grace + buffer)

Replay cache

30 seconds

Maximum lookback for replay data

Cross-instance

Supported

Sessions persist across server restarts

Timeline Example

T+0s:   Client connected, subscribed to allFills
T+10s:  Client disconnects (network issue)
        └─ Session marked disconnected, 30s grace period starts
T+15s:  Client reconnects with sessionId
        └─ Session resumed, replay sent for T+10s to T+15s
T+40s:  Grace period expires (T+10s + 30s)
        └─ Session would be deleted if not reconnected

Examples

const WebSocket = require('ws');

class HydromancerClient {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.liveFormat = null;  // Set to 'chunked-v1' if you use live chunking
    this.sessionId = null;
    this.lastCursor = null;  // Track last successfully processed cursor
    this.ws = null;
    this.replayHandler = new ReconnectionHandler((cursor) => {
      this.lastCursor = cursor;  // Update cursor after each successfully processed message
    });
  }

  connect() {
    let url = `wss://api.hydromancer.xyz/ws?token=${this.apiKey}`;
    if (this.sessionId) {
      url += `&sessionId=${this.sessionId}`;
      // Always provide cursor for accurate replay
      if (this.lastCursor) {
        url += `&cursor=${this.lastCursor}`;
      }
      console.log(`Reconnecting with session ${this.sessionId.slice(0, 8)}, cursor ${this.lastCursor}...`);
    }
    if (this.liveFormat) {
      url += `&liveFormat=${this.liveFormat}`;
    }

    this.ws = new WebSocket(url);

    this.ws.on('message', (data) => {
      const msg = JSON.parse(data);
      this.handleMessage(msg);
    });

    this.ws.on('close', () => {
      console.log('Disconnected, reconnecting in 1s...');
      setTimeout(() => this.connect(), 1000);
    });

    this.ws.on('error', (err) => {
      console.error('WebSocket error:', err.message);
    });
  }

  handleMessage(msg) {
    switch (msg.type) {
      case 'connected':
        // New session - save ID and subscribe
        this.sessionId = msg.sessionId;
        console.log(`New session: ${this.sessionId}`);
        this.subscribe({ type: 'allFills' });
        break;

      case 'reconnected':
        // Session resumed - subscriptions already restored
        console.log(`Resumed session: ${msg.sessionId}`);
        console.log(`Restored subscriptions: ${msg.subscriptions.join(', ')}`);
        // Reset replay handler for new connection (keep cursor callback)
        this.replayHandler = new ReconnectionHandler((cursor) => {
          this.lastCursor = cursor;
        });
        // Don't subscribe - already restored!
        break;

      case 'replay':
        // Historical data - different from live events
        console.log(`Replay chunk ${msg.chunk}/${msg.totalChunks}: ${msg.data.length} items`);
        if (msg.hasGap) {
          console.warn('Gap detected - some data may be missing');
          // Consider fetching from REST API
        }
        this.replayHandler.onMessage(msg);
        break;

      case 'subscriptionUpdate':
        console.log(`Subscribed: ${msg.subscribed.join(', ')}`);
        break;

      case 'ping':
        this.ws.send(JSON.stringify({ type: 'pong' }));
        break;

      case 'error':
        console.error(`Server error: ${msg.message}`);
        break;

      default:
        // Live data message (has seq field)
        if (msg.seq !== undefined) {
          this.replayHandler.onMessage(msg);
        }
    }
  }

  subscribe(subscription) {
    this.ws.send(JSON.stringify({
      method: 'subscribe',
      subscription
    }));
  }
}

// Replay handler from earlier (with deduplication)
class ReconnectionHandler {
  constructor(onCursorUpdate) {
    this.replayComplete = false;
    this.replayChunksReceived = 0;
    this.replayTotalChunks = 0;
    this.bufferedLive = [];
    this.lastSeq = 0;
    this.onCursorUpdate = onCursorUpdate;  // Callback to update client's cursor
    // Deduplication: track last seen (time, txIndex) per channel
    this.lastSeen = { time: 0, txIndex: 0 };
  }

  // Check if event is new (for deduplication)
  isNewEvent(time, txIndex) {
    if (time > this.lastSeen.time) return true;
    if (time === this.lastSeen.time && txIndex > this.lastSeen.txIndex) return true;
    return false;
  }

  onMessage(msg) {
    if (msg.type === 'replay') {
      this.replayTotalChunks = msg.totalChunks;
      this.replayChunksReceived++;

      // Process replay data
      msg.data.forEach(item => this.processEvent(item, 'replay'));

      // Update cursor from each chunk for incremental progress
      // If disconnected mid-replay, reconnect resumes from the last completed chunk
      if (msg.cursor && this.onCursorUpdate) {
        this.onCursorUpdate(msg.cursor);
      }

      if (this.replayChunksReceived === this.replayTotalChunks) {
        this.replayComplete = true;
        console.log('Replay complete, processing buffered live events...');
        this.bufferedLive.forEach(m => this.processLiveMessage(m));
        this.bufferedLive = [];
      }
    } else if (msg.seq !== undefined) {
      if (!this.replayComplete) {
        this.bufferedLive.push(msg);
      } else {
        this.processLiveMessage(msg);
      }
    }
  }

  processLiveMessage(msg) {
    // Check for sequence gaps
    if (this.lastSeq > 0 && msg.seq !== this.lastSeq + 1) {
      console.warn(`Sequence gap: expected ${this.lastSeq + 1}, got ${msg.seq}`);
    }
    this.lastSeq = msg.seq;

    // Process the data
    const items = msg.fills || msg.updates || msg.events || msg.trades || msg.liquidations || msg.data || [];
    items.forEach(item => this.processEvent(item, 'live'));

    // Update cursor after successful processing
    if (msg.cursor && this.onCursorUpdate) {
      this.onCursorUpdate(msg.cursor);
    }
  }

  processEvent(item, source) {
    // Extract event data (fills are [address, fillData], others may differ)
    const eventData = Array.isArray(item) ? item[1] : item;
    const time = eventData.time || eventData.closeTime || 0;
    const txIndex = eventData.txIndex ?? 0;

    // Deduplicate: skip if already seen
    if (!this.isNewEvent(time, txIndex)) {
      return;  // Duplicate from replay/live overlap
    }
    this.lastSeen = { time, txIndex };

    console.log(`[${source}] Event:`, JSON.stringify(item).slice(0, 100));
  }
}

const client = new HydromancerClient(process.env.HYDROMANCER_API_KEY);
client.connect();

import websocket
import json
import os
import time

class ReconnectionHandler:
    def __init__(self, on_cursor_update=None):
        self.replay_complete = False
        self.replay_chunks_received = 0
        self.replay_total_chunks = 0
        self.buffered_live = []
        self.last_seq = 0
        self.on_cursor_update = on_cursor_update  # Callback to update client's cursor
        # Deduplication: track last seen (time, tx_index)
        self.last_seen = {'time': 0, 'tx_index': 0}

    def reset(self):
        """Reset state for new connection."""
        self.replay_complete = False
        self.replay_chunks_received = 0
        self.replay_total_chunks = 0
        self.buffered_live = []
        self.last_seq = 0
        self.last_seen = {'time': 0, 'tx_index': 0}

    def is_new_event(self, time, tx_index):
        """Check if event is new (for deduplication)."""
        if time > self.last_seen['time']:
            return True
        if time == self.last_seen['time'] and tx_index > self.last_seen['tx_index']:
            return True
        return False

    def on_message(self, msg):
        if msg.get('type') == 'replay':
            self.replay_total_chunks = msg['totalChunks']
            self.replay_chunks_received += 1

            # Process replay data
            for item in msg.get('data', []):
                self.process_event(item, 'replay')

            # Update cursor from each chunk for incremental progress
            if msg.get('cursor') and self.on_cursor_update:
                self.on_cursor_update(msg['cursor'])

            if self.replay_chunks_received == self.replay_total_chunks:
                self.replay_complete = True
                print('Replay complete, processing buffered live events...')
                for buffered in self.buffered_live:
                    self.process_live_message(buffered)
                self.buffered_live = []

        elif 'seq' in msg:
            if not self.replay_complete:
                self.buffered_live.append(msg)
            else:
                self.process_live_message(msg)

    def process_live_message(self, msg):
        seq = msg['seq']
        # Check for sequence gaps
        if self.last_seq > 0 and seq != self.last_seq + 1:
            print(f'Sequence gap: expected {self.last_seq + 1}, got {seq}')
        self.last_seq = seq

        # Process the data
        items = msg.get('fills') or msg.get('updates') or msg.get('events') or msg.get('trades') or msg.get('liquidations') or msg.get('data') or []
        for item in items:
            self.process_event(item, 'live')

        # Update cursor after successful processing
        if msg.get('cursor') and self.on_cursor_update:
            self.on_cursor_update(msg['cursor'])

    def process_event(self, item, source):
        # Extract event data (fills are [address, fill_data], others may differ)
        event_data = item[1] if isinstance(item, list) else item
        time = event_data.get('time') or event_data.get('closeTime') or 0
        tx_index = event_data.get('txIndex') or 0

        # Deduplicate: skip if already seen
        if not self.is_new_event(time, tx_index):
            return  # Duplicate from replay/live overlap
        self.last_seen = {'time': time, 'tx_index': tx_index}

        print(f'[{source}] Event: {str(item)[:100]}')


class HydromancerClient:
    def __init__(self, api_key):
        self.api_key = api_key
        self.live_format = None  # Set to "chunked-v1" if you use live chunking
        self.session_id = None
        self.last_cursor = None  # Track last successfully processed cursor
        self.ws = None
        self.replay_handler = ReconnectionHandler(self.on_cursor_update)

    def on_cursor_update(self, cursor):
        """Called after each successfully processed message."""
        self.last_cursor = cursor

    def connect(self):
        url = f"wss://api.hydromancer.xyz/ws?token={self.api_key}"
        if self.session_id:
            url += f"&sessionId={self.session_id}"
            # Always provide cursor for accurate replay
            if self.last_cursor:
                url += f"&cursor={self.last_cursor}"
            print(f"Reconnecting with session {self.session_id[:8]}, cursor {self.last_cursor}...")
        if self.live_format:
            url += f"&liveFormat={self.live_format}"

        self.ws = websocket.WebSocketApp(
            url,
            on_message=self.on_message,
            on_close=self.on_close,
            on_open=self.on_open,
            on_error=self.on_error
        )
        self.ws.run_forever()

    def on_open(self, ws):
        print("WebSocket connected")

    def on_error(self, ws, error):
        print(f"WebSocket error: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        print("Disconnected, reconnecting in 1s...")
        time.sleep(1)
        self.connect()

    def on_message(self, ws, message):
        msg = json.loads(message)
        msg_type = msg.get('type')

        if msg_type == 'connected':
            # New session
            self.session_id = msg['sessionId']
            print(f"New session: {self.session_id}")
            self.subscribe({'type': 'allFills'})

        elif msg_type == 'reconnected':
            # Session resumed - subscriptions restored
            print(f"Resumed session: {msg['sessionId']}")
            print(f"Restored subscriptions: {msg.get('subscriptions', [])}")
            # Reset replay handler for new connection (keeps cursor callback)
            self.replay_handler.reset()
            # Don't subscribe - already restored!

        elif msg_type == 'replay':
            # Historical data
            chunk = msg['chunk']
            total = msg['totalChunks']
            items = len(msg.get('data', []))
            print(f"Replay chunk {chunk}/{total}: {items} items")
            if msg.get('hasGap'):
                print("WARNING: Gap detected - some data may be missing")
            self.replay_handler.on_message(msg)

        elif msg_type == 'subscriptionUpdate':
            print(f"Subscribed: {msg.get('subscribed', [])}")

        elif msg_type == 'ping':
            ws.send(json.dumps({'type': 'pong'}))

        elif msg_type == 'error':
            print(f"Server error: {msg.get('message')}")

        elif 'seq' in msg:
            # Live data message
            self.replay_handler.on_message(msg)

    def subscribe(self, subscription):
        self.ws.send(json.dumps({
            'method': 'subscribe',
            'subscription': subscription
        }))


if __name__ == '__main__':
    api_key = os.environ.get('HYDROMANCER_API_KEY')
    if not api_key:
        print("Set HYDROMANCER_API_KEY environment variable")
        exit(1)

    client = HydromancerClient(api_key)
    client.connect()

Best Practices

Always save the sessionId from connected messages
Always pass sessionId when reconnecting
Track the cursor from each successfully processed message
Always pass your cursor when reconnecting for accurate replay
Distinguish replay from live - check msg.type === 'replay'
Buffer live messages during replay processing
Deduplicate events - replay and live may overlap (see Deduplication)
Handle hasGap: true by fetching missing data from REST API if critical
Track sequences within a connection to detect gaps
Implement exponential backoff for reconnection attempts
Respond to pings within 150 seconds to keep connection alive

Error Handling

Session Expired

If your session expired, you'll receive connected instead of reconnected:

{
  "type": "connected",
  "clientId": "new-client-...",
  "sessionId": "new-session-..."
}

This means:

A new session was created
You must re-subscribe to your feeds
No replay will be sent

Detection: Check msg.type === 'reconnected' to confirm successful resume.

Common Errors

Error

Cause

Solution

"Session not found"

Invalid or expired sessionId

Create new session, re-subscribe

"Invalid API key"

Bad token

Check API key

"Rate limit exceeded"

Too many connections/messages

Implement backoff

"Connection timeout"

Missed ping/pong

Respond to pings within 150s

Summary

Concept

Purpose

Client Responsibility

sessionId

Persist subscriptions across reconnects

Save and reuse

seq

Detect gaps within a connection

Track and validate

cursor

Stream position (block:ts:tx_index)

Save from each message and replay chunk, provide on reconnect

Replay

Historical data on reconnect

Buffer live until complete

Deduplication

Replay/live events may overlap

Track (time, txIndex), skip events at or before it

hasGap

Only present when true - incomplete replay

Fetch from REST if critical

Auto-restore

Subscriptions restored automatically

None - just reconnect with sessionId

Remember: The server's cursor tracks what was sent, not what you received. For guaranteed data continuity, always track and provide your own cursor based on messages you successfully processed.

PreviousWebsocket NextRate limits, user limits and heartbeats

Last updated 1 month ago

Overview

Key Concepts

Sequence Numbers (seq)

Cursor

Cursor Format

Message Types

Control Messages

Data Messages (Live Events)

Replay Messages

Chunking

Connection Flow

Initial Connection

Subscribing

Reconnection Flow

Step-by-Step Process

Providing Your Cursor on Reconnect

What Triggers Replay

When Replay is NOT Sent

Gap Detection

Replay Response Scenarios

Example Responses

Message Ordering

Why This Happens

Recommended Client Pattern

Deduplication

Why Overlap Occurs

Deduplication Keys

Deduplication Example (Fills)

Session Lifetime

Timeline Example

Examples

Best Practices

Error Handling

Session Expired

Common Errors

Summary

Sequence Numbers (`seq`)