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:

1. Restore subscriptions - No need to re-subscribe

2. Receive replay - Automatically get missed data (up to 30 seconds)

3. 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",
  "channel": "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 is your position in the data stream. It is an opaque string: store the latest one you received and send it back unchanged when you reconnect. The server handles all parsing.

Every live message and every replay chunk carries a cursor naming the last event in that message, precise to the individual event, so resuming from it neither misses nor duplicates events. Live-only channels with no replay (bbo, l2Book, allActiveAssetCtx, setOracleUpdates) send the sentinel "0".

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",
  "channel": "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 client subscription type in both type and channel. The values are always equal:

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

Field	Type	Description
type	string	Subscription type (e.g., "allFills")
channel	string	Same value as type; either field can be used for routing
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,
  "hasMore": true,
  "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
hasMore	boolean	Always present. true while more chunks follow; false on the terminal chunk. Prefer it over chunk == totalChunks to detect the end of a replay.
hasGap	boolean	Only present when true. Set on the first chunk when the cache didn't cover the entire disconnect period, and repeated on the terminal chunk when the replay had to stop early. 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 hasMore (or 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

• hasMore: true means another chunk follows; the terminal chunk has hasMore: false and chunk == totalChunks

• hasGap is set on the first chunk (chunk: 1), and repeated on the terminal chunk if the replay stopped early

• Process chunks in order as they arrive

Important: Replay messages do NOT have a seq field. Only live event messages carry the per-subscription sequence. Distinguish replay from live by "type": "replay".

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",
  "requestId": "fills-1",
  "subscription": { "type": "allFills" }
}

Server confirms with subscriptionUpdate:

{
  "type": "subscriptionUpdate",
  "requestId": "fills-1",
  "operation": "subscribe",
  "subscription": { "type": "allFills" },
  "subscribed": ["allFills"],
  "failed": []
}

requestId is optional, but a unique value is recommended for every subscribe or unsubscribe request. When present it must be at most 128 characters and must not contain control characters; invalid values are rejected with code: "invalid_request_id". The server echoes valid ids on both confirmations and errors, allowing multiplexed clients to match feedback to the request that caused it. Unsubscribe confirmations additionally contain unsubscribed. These correlation fields apply to the request/response exchange only; live data frames do not carry requestId.

Then live data begins flowing with seq starting at 1.

Reconnection Flow

Step-by-Step Process

1. Client disconnects (network issue, restart, etc.)

2. Server marks session disconnected - starts 30-second grace period

3. Client reconnects with sessionId and cursor:

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

4. Server validates session - checks if still within grace period

5. Server automatically restores subscriptions - no need to re-subscribe

6. Server sends reconnected:

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

7. Server sends replay - events since your cursor (up to 30 seconds of data)

8. 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	No	Session ID to resume (omit for new session)
cursor	No	Last cursor you received (opaque string, e.g. "500:1706123456789:3")
liveFormat	No	Required on every reconnect if you are using chunked-v1 live message format

Cursor priority:

1. Client-provided cursor (recommended) - from the cursor query parameter

2. 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. Route using the additive channels array, which contains canonical client subscription types. The legacy subscriptions array remains available and may contain internal all-market policy keys such as allFillsAll or l2BookAll.

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 replay coverage retained by the server - some events aged out before you reconnected. hasGap is determined from block coverage, not from individual events, so an item-time ordering violation and hasGap are separate signals and should be diagnosed independently.

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	0	true	[]	Cache empty and cursor is old - all events expired
No data, no gap	0	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,
  "hasMore": false,
  "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,
  "hasMore": false,
  "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,
  "hasMore": false,
  "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,
  "hasMore": false,
  "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

The server holds live frames on a resumed connection until replay and restored snapshots finish. Client-side buffering remains safe. A live batch that straddles the last replayed item can still overlap replay; deduplicate as below.

Why Overlap Can Still Happen

The server adds you to live broadcast immediately, then holds those frames until recovery completes. Held live that the recovery watermark already covers is dropped. A live batch that straddles the last replayed item is delivered whole, so earlier items in that batch can duplicate replay.

allMids is snapshot-only and does not advance the shared session cursor. Per-reconnecting-client memory is bounded at the client channel plus the hold queue (2× WS_CLIENT_BUFFER_SIZE).

Recommended Client Pattern

class ReconnectionHandler {
  constructor() {
    this.replayComplete = false;
    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);
      }

      // The terminal chunk carries hasMore: false
      if (msg.hasMore === false) {
        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:

1. Subscribes you to live broadcasts immediately on reconnect

2. 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
builderApprovedFundings	(time, user, coin)	entry.time, entry.user, entry.delta.coin
builderApprovedNonFundingLedgerEvents	(time, txIndex, user, role)	entry.time, entry.txIndex, entry.user, entry.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

JavaScript

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.hasMore ? '' : ' (final)'}: ${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.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') {
      // 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);
      }

      // The terminal chunk carries hasMore: false
      if (msg.hasMore === false) {
        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();

Python

import websocket
import json
import os
import time

class ReconnectionHandler:
    def __init__(self, on_cursor_update=None):
        self.replay_complete = False
        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.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':
            # 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'])

            # The terminal chunk carries hasMore: false
            if msg.get('hasMore') is False:
                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

1. Always save the sessionId from connected messages

2. Always pass sessionId when reconnecting

3. Track the cursor from each successfully processed message

4. Always pass your cursor when reconnecting for accurate replay

5. Distinguish replay from live - check msg.type === 'replay'

6. Buffer live messages during replay processing

7. Deduplicate events - replay and live may overlap (see Deduplication)

8. Handle hasGap: true by fetching missing data from REST API if critical

9. Track sequences within a connection to detect gaps

10. Implement exponential backoff for reconnection attempts

11. 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 (opaque string)	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.