feat(websocket): implement automatic reconnect and subscription replay with enhanced tracking

Author: kzhdev

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ---
 
+## [0.2.1] - 2026-06-23
+
+### Added
+- Add automatic `WebsocketManager` reconnect attempts with bounded exponential backoff after disconnects or transport errors.
+- Replay active WebSocket subscriptions after reconnect so `Info::subscribe()` callbacks resume without manual re-subscription.
+
+### Fixed
+- Keep WebSocket subscription replay lock-free and synchronized with unsubscribe by tracking per-identifier subscription state, in-flight subscribe/replay sends, and final unsubscribe phases with atomics.
+- Avoid stale server-side subscriptions when a reconnect replay overlaps with the last local `unsubscribe()` for a channel.
+- Coalesce multiple callbacks on the same channel into one wire subscription until the last callback unsubscribes.
+
+### Tests
+- Add offline `SubscriptionTracking` tests covering subscribe while disconnected, callback routing after unsubscribe, concurrent subscribe/unsubscribe, wire subscription coalescing, and unsubscribe-before-resubscribe ordering using a local in-process WebSocket server.
+
 ## [0.2.0] -2026-06-19
 
 ### Added

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.21)
-project(hyperliquid-cpp VERSION 0.2.0 LANGUAGES CXX)
+project(hyperliquid-cpp VERSION 0.2.1 LANGUAGES CXX)
 
 set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)

README.md

@@ -25,7 +25,7 @@ All authenticated actions are signed locally using EIP-712 / secp256k1; your pri
 
 - Read-only market data (`Info`): universe metadata, mid prices, order books, candles, funding history, account state
 - Authenticated trading (`Exchange`): limit/market/trigger orders, bulk ops, cancel, modify, leverage, margin, USD/spot transfers, bridge withdrawal
-- Real-time WebSocket subscriptions via `Info::subscribe` with multi-callback fan-out per channel
+- Real-time WebSocket subscriptions via `Info::subscribe` with multi-callback fan-out, automatic reconnect, and active-subscription replay
 - Optional caller-thread WebSocket dispatch via `Info::dispatch()` for applications that own their event loop
 - Optional shared-memory WebSocket buffers backed by `slick::stream_buffer_multiplexer`
 - Native EIP-712 signing for both L1 actions (orders, leverage) and user-signed actions (transfers) using OpenSSL secp256k1
@@ -191,8 +191,10 @@ info->unsubscribe({{"type", "l2Book"}, {"coin", "ETH"}}, sid);
 info->unsubscribe({{"type", "l2Book"}, {"coin", "ETH"}}, sid2);
 ```
 
-The WebSocket manager sends periodic pings to keep connections alive and uses
-bounded atomic shutdown checks so teardown does not wait for the full ping interval.
+The WebSocket manager sends periodic pings, reconnects with bounded exponential
+backoff after disconnects or transport errors, and replays active subscriptions
+after reconnect. Multiple callbacks on the same channel share one server-side
+subscription until the last callback unsubscribes.
 
 For applications that need callbacks on their own thread, enable caller-thread dispatch
 and poll queued WebSocket records:
@@ -507,7 +509,7 @@ cmake --build build --config Debug --target hyperliquid_tests
 ctest --test-dir build -C Debug -R hyperliquid_tests -V
 ```
 
-Covers: Keccak-256 vectors, EIP-712 signing round-trips, type serialisation (`float_to_wire`, `Cloid`, `Tif`), WebSocket URL conversion, channel identifier generation, caller-thread WebSocket dispatch routing, and shared-memory stream-buffer attachment.
+Covers: Keccak-256 vectors, EIP-712 signing round-trips, type serialisation (`float_to_wire`, `Cloid`, `Tif`), WebSocket URL conversion, channel identifier generation, caller-thread WebSocket dispatch routing, lock-free reconnect subscription tracking, and shared-memory stream-buffer attachment.
 
 ### Integration tests (requires testnet)
 

include/hyperliquid/websocket_manager.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <atomic>
+#include <cstdint>
 #include <functional>
 #include <limits>
 #include <memory>
@@ -101,6 +102,9 @@ class WebsocketManager {
     bool dispatch(uint32_t producer_id, const char* data, std::size_t length);
 
 private:
+    struct SubscriptionState;
+    using SubscriptionStatePtr = std::shared_ptr<SubscriptionState>;
+
     void init(
         uint32_t read_buffer_size,
         uint32_t read_control_size,
@@ -112,33 +116,73 @@ class WebsocketManager {
     void on_disconnected();
     void on_message(const char* data, std::size_t len);
     void on_error(std::string err);
+    void resubscribe_all();  // re-sends all subscribe messages; called from on_connected()
+    void send_subscribe_if_active(const SubscriptionStatePtr& state);
+    void activate_subscription(const SubscriptionStatePtr& state);
+    void deactivate_subscription(const SubscriptionStatePtr& state,
+                                 const nlohmann::json& subscription);
+    SubscriptionStatePtr subscription_state_for(
+        const std::string& identifier,
+        const nlohmann::json& subscription);
 
-    void ping_loop();        // runs in ping_thread_
+    void ping_loop();        // runs in ping_thread_; also manages reconnection
 
     std::string ws_url_;
     std::unique_ptr<Websocket> ws_;
 
     std::unique_ptr<slick::stream_buffer_multiplexer> owning_mux_;
     slick::stream_buffer_multiplexer &mux_;
 
+    enum class SubscriptionPhase : std::uint8_t {
+        Inactive,
+        Active,
+        Unsubscribing,
+    };
+
+    // Per-identifier state is shared by all callbacks for that subscription.
+    // It serializes reconnect replay and final unsubscribe with atomics only:
+    // final unsubscribe waits for in-flight subscribe/replay sends before
+    // sending the server-side unsubscribe.
+    struct SubscriptionState {
+        std::shared_ptr<const nlohmann::json> subscription;
+        std::atomic<unsigned int> handler_count{0};
+        std::atomic<unsigned int> send_in_flight{0};
+        std::atomic<SubscriptionPhase> phase{SubscriptionPhase::Inactive};
+
+        explicit SubscriptionState(const nlohmann::json& sub)
+            : subscription(std::make_shared<const nlohmann::json>(sub)) {}
+    };
+
+    using SubStateMap = std::unordered_map<std::string, SubscriptionStatePtr>;
+
     // Handler snapshots are published with copy-on-write. Each snapshot holds
     // shared per-callback state so unsubscribe() can deactivate exactly one
     // callback and wait for only that callback's in-flight invocation to drain.
     struct Handler {
         int subscription_id;
         std::function<void(const nlohmann::json&)> callback;
+        SubscriptionStatePtr subscription_state;
         std::atomic<bool> active{true};
         std::atomic<unsigned int> in_flight{0};
 
-        Handler(int id, std::function<void(const nlohmann::json&)> cb)
-            : subscription_id(id), callback(std::move(cb)) {}
+        Handler(int id,
+                std::function<void(const nlohmann::json&)> cb,
+                SubscriptionStatePtr state)
+            : subscription_id(id)
+            , callback(std::move(cb))
+            , subscription_state(std::move(state)) {}
     };
 
     using HandlerPtr = std::shared_ptr<Handler>;
     using HandlerMap = std::unordered_map<std::string, std::vector<HandlerPtr>>;
 
     std::shared_ptr<const HandlerMap> handlers_;
 
+    // Subscription states keyed by identifier, used to re-subscribe after reconnect.
+    // States may remain as inactive tombstones so a new subscribe can safely
+    // synchronize with a concurrent final unsubscribe for the same identifier.
+    std::shared_ptr<const SubStateMap> sub_state_map_{std::make_shared<SubStateMap>()};
+
     std::atomic<bool> connected_{false};
     std::atomic<bool> running_{true};
     std::atomic<int>  next_id_{1};