diff --git a/CHANGELOG.md b/CHANGELOG.md index fbd6d5a..b7a7cab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,41 @@ All notable changes to the `oxarchive` Rust SDK are tracked in this file. The format is loosely based on [Keep a Changelog](https://keepachangelog.com). +## [1.13.0] - 2026-10-08 + +### Added +- The `mempool` WebSocket channel (`ws::MEMPOOL_CHANNEL`): signed Hyperliquid + transactions as our Hyperliquid node receives them from its peers, before + they are included in a block, for every Hyperliquid product. It is live + only, served only at `wss://stream.0xarchive.io/ws`, and included with the + Pro, Scale and Enterprise plans. Every other channel stays on every plan. + `subscribe(MEMPOOL_CHANNEL, None)` streams every pending transaction our + Hyperliquid node receives; with a symbol, only the actions that reference + that market. `mempool` is + the only channel whose symbol is optional. +- `ServerMsg::Mempool`, with one `MempoolItem` (and its `MempoolSignature`) + per signed action. `ServerMsg::from_text()`, which the client uses to read + every message, builds it straight from the frame text, so + `MempoolItem::action` (a `Box`) keeps the + exact bytes the server sent, key order included, for signature recovery. + `MempoolItem::action_value()` parses the action into a `serde_json::Value` + for inspection, which may not keep the original key order. `ServerMsg::mempool_items()` returns + a message's items. Unknown item fields are ignored. +- serde_json's `raw_value` feature is enabled. +- `ws::STREAM_WS_URL` (`wss://stream.0xarchive.io/ws`) and + `ws::DEFAULT_WS_URL` (`wss://api.0xarchive.io/ws`, the default `ws_url`). + Pass `STREAM_WS_URL` to `WsOptions::ws_url()` for a connection that + subscribes to `mempool`. +- `Capability::ws_endpoint` and `Capability::plans`, the fields + `/v1/capabilities` sets on the `mempool` row only. For `ws_endpoint`, + `None` means the default endpoint, `wss://api.0xarchive.io/ws`. For + `plans`, `None` means every plan. They are left out when serializing a row that + does not have them. + +### Changed +- `ServerMsg` gained the `Mempool` variant; an exhaustive match on it needs + a new arm. + ## [1.12.0] - 2026-10-05 ### Upgrading from 1.8 diff --git a/Cargo.toml b/Cargo.toml index e7d263f..9bfcb2a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "oxarchive" -version = "1.12.0" +version = "1.13.0" edition = "2021" rust-version = "1.85" description = "Rust SDK for async 0xArchive market data clients" @@ -63,7 +63,7 @@ include = [ [dependencies] reqwest = { version = "0.12", features = ["json"] } serde = { version = "1", features = ["derive"] } -serde_json = "1" +serde_json = { version = "1", features = ["raw_value"] } thiserror = "2" chrono = { version = "0.4", features = ["serde"] } tokio = { version = "1", features = ["rt", "macros"] } diff --git a/README.md b/README.md index 0bc2a44..4ff8df0 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Or add directly to your `Cargo.toml`: ```toml [dependencies] -oxarchive = "1.12" +oxarchive = "1.13" tokio = { version = "1", features = ["rt-multi-thread", "macros"] } chrono = "0.4" # the examples below use it for time windows ``` @@ -26,7 +26,7 @@ chrono = "0.4" # the examples below use it for time windows For WebSocket support (real-time streaming and replay): ```toml -oxarchive = { version = "1.12", features = ["websocket"] } +oxarchive = { version = "1.13", features = ["websocket"] } ``` ## Quick Start @@ -196,8 +196,10 @@ println!("{} symbols, {} on HIP-3", symbols.len(), hip3.len()); one `Capability` row per venue and datatype: its REST routes, its WebSocket channels, whether those stream live (`live`) and replay history (`replay`), the first instant served (`available_from`), the cadence, the largest page -(`page_limit`), and the accepted `interval` values. The route is public and -uses no credits. Read replay availability from it rather than from a fixed +(`page_limit`), and the accepted `interval` values. The `mempool` row is the one +that sets `ws_endpoint` (the endpoint that serves it) and `plans` (the plans that +include it); rows without them use the default endpoint, `wss://api.0xarchive.io/ws`, +and are on every plan. The route is public and uses no credits. Read replay availability from it rather than from a fixed list: ```rust @@ -1850,6 +1852,8 @@ For large historical downloads, use the S3 Parquet bulk export in the [Data Cata > Lighter live subscriptions are available for `lighter_orderbook`, `lighter_trades`, `lighter_open_interest`, and `lighter_funding` on mainnet, and for `rh_lighter_orderbook`, `rh_lighter_trades`, `rh_lighter_open_interest`, and `rh_lighter_funding` on Robinhood Chain. `lighter_candles`, `lighter_l3_orderbook`, and `rh_lighter_candles` remain replay-only. Every Lighter channel on both deployments supports historical replay. +> The `mempool` channel (pending Hyperliquid transactions) is served only at `wss://stream.0xarchive.io/ws` and is included with the Pro, Scale and Enterprise plans. See [Pending Transactions (Mempool)](#pending-transactions-mempool). + The client connects with `version=2026-10-01`. Under that version, error messages carry an `error_code` (see [Error Handling](#error-handling)), and Lighter replay rows have the same shapes as the live payloads. `client.capabilities()` lists which channels stream live and which replay. ### Real-time Streaming @@ -1957,6 +1961,94 @@ If the lag persists, a notice such as `Stopped the lighter_trades stream for BTC ends that subscription; subscribe again to resume. `lighter_orderbook` never sends an older book in place of a newer one. +### Pending Transactions (Mempool) + +The `mempool` channel streams signed Hyperliquid transactions (orders, cancels, +modifies, TWAPs, leverage changes, transfers and every other action type) as our +Hyperliquid node receives them from its peers, before they are included in a block. +It covers every Hyperliquid product: perps, HIP-3, HIP-4 and spot. + +- **Live only.** Pending transactions are never stored. There is no replay, history, + REST route or export. +- **One endpoint.** It is served only at `wss://stream.0xarchive.io/ws` + (`oxarchive::ws::STREAM_WS_URL`), with the same API key and protocol. At + `wss://api.0xarchive.io/ws`, the default `ws_url`, a subscribe is answered with + `ErrorCode::EndpointUnsupported`. Use a separate connection for the channels served + at the default endpoint. +- **Plans.** It is included with the Pro, Scale and Enterprise plans; on other plans a + subscribe is answered with `ErrorCode::Forbidden`. Every other channel stays on every + plan, Free included. Each data message is metered like any other WebSocket message. + +```rust +use oxarchive::ws::{OxArchiveWs, ServerMsg, WsOptions, MEMPOOL_CHANNEL, STREAM_WS_URL}; + +let mut ws = OxArchiveWs::new(WsOptions::new("your-api-key").ws_url(STREAM_WS_URL)); +ws.connect().await?; +let mut rx = ws.rx.take().expect("receiver"); + +ws.subscribe(MEMPOOL_CHANNEL, Some("BTC")).await?; // actions that reference BTC +ws.subscribe(MEMPOOL_CHANNEL, Some("xyz:TSLA")).await?; // HIP-3; spot "HYPE-USDC", HIP-4 "#49720" +// ws.subscribe(MEMPOOL_CHANNEL, None).await?; // pending transactions our node receives (unfiltered) + +while let Some(msg) = rx.recv().await { + match msg { + ServerMsg::Mempool { symbol, items, .. } => { + for item in items { + // item.action.get() is the action's exact bytes, for signature recovery. + let action = item.action_value()?; // parsed, for inspection only + println!("{symbol:?} {:?} {} {:?}", item.received_at, action["type"], item.symbols); + } + } + ServerMsg::Error { message, error_code } => eprintln!("{error_code:?}: {message}"), + _ => {} + } +} +``` + +The symbol is optional on this channel only. Without it you receive every pending +transaction our Hyperliquid node receives; with it, every action whose asset ids include that market, whole (an +order batch that touches `BTC` and `ETH` reaches both subscriptions). Symbols are +spelled as everywhere else: perps `BTC`, HIP-3 `xyz:TSLA`, spot `HYPE-USDC` +(`HYPE/USDC` is also accepted) and HIP-4 `#49720`. An unknown symbol is answered with +`ErrorCode::InvalidSymbol`. The `subscribed` acknowledgement carries the canonical +symbol, or `None` for the unfiltered stream. + +The server sends one message per batch of transactions as it arrives. The client +delivers it as `ServerMsg::Mempool`, with `coin` and `symbol` set to the +subscription's symbol (`None` when unfiltered) and one `MempoolItem` per signed +action in `items`: + +| Field | Description | +|-------|-------------| +| `received_at` | When our node received the transaction: an RFC 3339 UTC string with nanosecond precision. Not a block time. | +| `received_at_ms` | The same time in Unix milliseconds. | +| `symbols` | Markets the action's asset ids reference, in first-seen order without repeats. Empty for actions with no market, such as transfers, `noop`, `scheduleCancel` and validator actions. | +| `action` | The action exactly as signed, in Hyperliquid's exchange-action format: asset ids (`a` or `asset`) rather than symbols, prices and sizes as strings. It is a `Box` that keeps the exact bytes the server sent, key order included, for signature recovery (`item.action.get()`). `item.action_value()` parses it into a `serde_json::Value` for inspection, which may not keep the original key order. | +| `nonce` | The action's nonce. | +| `vault_address` | The vault or subaccount the action acts for, or `None`. | +| `expires_after_ms` | The action's `expiresAfter` in Unix milliseconds, or `None`. | +| `signature` | `MempoolSignature` with `r`, `s` and `v`. The signer's address is not included. | + +The action's `type` names it, for example `order`, `cancel`, `cancelByCloid`, +`modify`, `batchModify`, `scheduleCancel`, `twapOrder`, `twapCancel`, +`updateLeverage`, `updateIsolatedMargin`, `noop`, `evmRawTx`, or a transfer such as +`usdSend`, `spotSend`, `usdClassTransfer` or `sendAsset`. Hyperliquid adds action +types, so handle types you do not recognise. + +A pending transaction is not an executed one: it can still be rejected, expire or +never land in a block. The same signed action can occasionally arrive twice; +deduplicate on `signature` if that matters to you. + +**Volume and limits.** The unfiltered stream is several megabytes per second before +compression, and this client does not negotiate permessage-deflate compression, so +subscribe with a symbol where you can. Unfiltered subscriptions are limited +server-wide, and when they are at capacity a subscribe without a symbol is answered +with `ErrorCode::RateLimited`; symbol subscriptions are not capped this way. A +connection that reads too slowly is disconnected, as on any channel, and the usual +per-connection limits apply (subscriptions per plan, 10 subscribe operations per +second). If the feed is temporarily unavailable, a subscribe is answered with +`ErrorCode::UpstreamUnavailable`. + ### Historical Replay Replay a bounded historical window with original timing preserved. Every replay @@ -2037,6 +2129,7 @@ ws.replay_stop().await?; | `funding` | Funding rate snapshots | Yes | Yes | | `ticker` | Price and 24h volume | Yes | No | | `all_tickers` | All market tickers | Yes | No | +| `mempool` | Pending transactions on every Hyperliquid product, before they are in a block. Symbol optional. Pro, Scale and Enterprise plans; served only at `wss://stream.0xarchive.io/ws`. | Yes | No | | `orderbook_full` | Hyperliquid core full-depth L2 order book, aggregated from order-level data (every price level, no user attribution): an `l4_snapshot` frame (`ServerMsg::L4Snapshot`) with the whole book, then `l4_batch` frames (`ServerMsg::L4Batch`) of price-level changes (`side`, `px`, `sz`, `n`, `bn`; `sz` and `n` are `0` when a level is removed). | Yes | Yes, bulk and single-channel, from 2026-03-11 01:03 UTC | | `lighter_orderbook` | Lighter L2 order book | Yes | Yes | | `lighter_trades` | Lighter trades | Yes | Yes | @@ -2095,7 +2188,7 @@ while let Some(msg) = rx.recv().await { ### Tier Limits -All self-serve tiers reach the published route families; Free covers the most recent rolling 30 days of history (30-day span per request or replay), and Build and above keep the retained archive. Schema availability remains family-specific; plans gate capacity and Free's 30-day history window, not route families, schemas, or served depth. +All self-serve tiers reach the published route families; Free covers the most recent rolling 30 days of history (30-day span per request or replay), and Build and above keep the retained archive. Schema availability remains family-specific; plans gate capacity and Free's 30-day history window, not route families, schemas, or served depth. The one exception is the live `mempool` channel, included with the Pro, Scale and Enterprise plans. | Tier | Max Subscriptions | Max Connections | Max Replay Speed | |------|------------------|-----------------|------------------| diff --git a/src/error.rs b/src/error.rs index 926dffa..27b33d9 100644 --- a/src/error.rs +++ b/src/error.rs @@ -52,7 +52,9 @@ pub enum ErrorCode { NotFound, /// Missing or invalid credentials. Unauthorized, - /// The credentials do not grant access to this resource. + /// The credentials do not grant access to this resource, or the plan does + /// not include it (the `mempool` WebSocket channel on plans other than + /// Pro, Scale and Enterprise). Forbidden, /// The monthly credits are spent. InsufficientCredits, diff --git a/src/lib.rs b/src/lib.rs index 21cc82b..e1fdff7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -151,7 +151,7 @@ //! replay: //! //! ```toml -//! oxarchive = { version = "1.12", features = ["websocket"] } +//! oxarchive = { version = "1.13", features = ["websocket"] } //! ``` //! //! Live subscriptions cover the supported Hyperliquid channels and four @@ -167,6 +167,11 @@ //! replay as bulk, single-channel streams. `client.capabilities()` lists //! which channels stream live and which replay. //! +//! The live-only `mempool` channel carries pending Hyperliquid transactions, +//! before they are in a block; its items decode into [`MempoolItem`]. It is +//! served only at `wss://stream.0xarchive.io/ws` (`ws::STREAM_WS_URL`) and +//! included with the Pro, Scale and Enterprise plans. +//! //! Bulk streaming over WebSocket has been discontinued, so //! `OxArchiveWs::stream` is deprecated. For large historical downloads, use //! the S3 Parquet bulk export at . @@ -197,7 +202,7 @@ pub use types::{ LiquidationLevelBucket, LiquidationLevels, LiquidationLevelsHistoryItem, TriggerLevelBucket, TriggerLevels, TriggerLevelsHistoryItem, LighterLiveAssetCtx, LighterLiveData, LighterLiveLevel, LighterLiveMarketStats, - LighterLiveOrderBook, LighterLiveTrade, + LighterLiveOrderBook, LighterLiveTrade, MempoolItem, MempoolSignature, AccountSummary, CumulativeFunding, LighterL1Account, LighterL1Accounts, LighterLiquidation, LighterLiquidationVolume, MarketPosition, MarketPositionsSummary, MetaResponse, Position, PositionChange, PositionLeverage, PositionsFreshness, ResponseMeta, WalletPositions, diff --git a/src/types.rs b/src/types.rs index 4efea09..cafc349 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1266,6 +1266,93 @@ impl LighterLiveData { } } +// --------------------------------------------------------------------------- +// Mempool (pending Hyperliquid transactions, live only) +// --------------------------------------------------------------------------- + +/// One signed action from a live `mempool` message. +/// +/// The `mempool` channel streams signed Hyperliquid transactions as our +/// Hyperliquid node receives them from its peers, before they are included in +/// a block, for every Hyperliquid product (perps, HIP-3, HIP-4 and spot). It is +/// live only, served only at `wss://stream.0xarchive.io/ws`, and included +/// with the Pro, Scale and Enterprise plans. +/// +/// A pending transaction is not an executed one: it can still be rejected, +/// expire or never land in a block. The same signed action can occasionally +/// arrive twice; deduplicate on [`signature`](Self::signature) if that matters +/// to you. +/// +/// With the `websocket` feature, `OxArchiveWs` delivers each message's items +/// in `ServerMsg::Mempool`. Elsewhere, deserialize them with serde_json from +/// the message text, not from a `serde_json::Value`, which has already sorted +/// the action's keys. Fields this version does not know are ignored, and new +/// ones may be added in minor releases, so the struct cannot be built with a +/// literal outside this crate. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[non_exhaustive] +pub struct MempoolItem { + /// When our node received the transaction, as an RFC 3339 UTC string with + /// the node's nanosecond precision, for example + /// `2026-10-08T01:57:23.548737209Z`. Not a block time. + pub received_at: Option, + /// When our node received the transaction, in Unix milliseconds. + pub received_at_ms: Option, + /// The markets the action's asset ids reference, spelled as everywhere + /// else (`BTC`, `xyz:TSLA`, `HYPE-USDC`, `#49720`), in first-seen order + /// without repeats. Empty for actions with no market, such as transfers, + /// `noop`, `scheduleCancel` and validator actions. + #[serde(default)] + pub symbols: Vec, + /// The action exactly as signed, in Hyperliquid's exchange-action format: + /// asset ids (`a` or `asset`) rather than symbols, and prices and sizes as + /// strings. Its `type` names it, for example `order`, `cancel`, + /// `cancelByCloid`, `modify`, `batchModify`, `scheduleCancel`, + /// `twapOrder`, `twapCancel`, `updateLeverage`, `updateIsolatedMargin`, + /// `noop`, `evmRawTx`, or a transfer such as `usdSend`, `spotSend`, + /// `usdClassTransfer` or `sendAsset`. Hyperliquid adds action types, so + /// handle types you do not recognise. + /// + /// This raw form keeps the exact bytes the server sent, key order + /// included, as signature recovery needs them; `action.get()` returns + /// them as a `&str`. To inspect the action, + /// [`action_value`](Self::action_value) parses it into a + /// `serde_json::Value`, which may not keep the original key order. + pub action: Box, + /// The action's nonce. + pub nonce: Option, + /// The vault or subaccount the action acts for, or `None`. + pub vault_address: Option, + /// The action's `expiresAfter` in Unix milliseconds, or `None`. + pub expires_after_ms: Option, + /// The signature over the action. The signer's address is not included; + /// it can be recovered from the signature and the action. + pub signature: Option, +} + +/// The `{r, s, v}` signature of a pending action, as signed. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct MempoolSignature { + /// `r` as a hex string. + pub r: String, + /// `s` as a hex string. + pub s: String, + /// Recovery id. + pub v: u64, +} + +impl MempoolItem { + /// The action parsed into a `serde_json::Value`, for inspection, for + /// example `item.action_value()?["type"]`. + /// + /// The parsed form may not keep the original key order. Use the raw + /// [`action`](Self::action) for signature recovery. + pub fn action_value(&self) -> crate::Result { + serde_json::from_str(self.action.get()) + .map_err(|e| crate::Error::Deserialize(e.to_string())) + } +} + // --------------------------------------------------------------------------- // Convenience / summary types // --------------------------------------------------------------------------- @@ -2366,6 +2453,15 @@ pub struct Capability { /// Further detail, such as how replay behaves. #[serde(default)] pub notes: Option, + /// The WebSocket endpoint that serves this row's channels, when it is not + /// the default endpoint: `wss://stream.0xarchive.io/ws` for `mempool`. + /// `None` means the default endpoint, `wss://api.0xarchive.io/ws`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ws_endpoint: Option, + /// The plans that include it, when that is not every plan: `pro`, `scale` + /// and `enterprise` for `mempool`. `None` means every plan, Free included. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub plans: Option>, } impl Capability { diff --git a/src/ws/client.rs b/src/ws/client.rs index 53c7b47..088e8a8 100644 --- a/src/ws/client.rs +++ b/src/ws/client.rs @@ -6,7 +6,7 @@ //! //! Requires the `websocket` feature: //! ```toml -//! oxarchive = { version = "1.12", features = ["websocket"] } +//! oxarchive = { version = "1.13", features = ["websocket"] } //! ``` use futures_util::{SinkExt, StreamExt}; @@ -18,13 +18,63 @@ use tokio_tungstenite::{connect_async, tungstenite::Message}; use crate::error::{Error, ErrorCode, Result}; use crate::http::API_VERSION; -use crate::types::LighterLiveData; +use crate::types::{LighterLiveData, MempoolItem}; // --------------------------------------------------------------------------- // Configuration // --------------------------------------------------------------------------- +/// The default `ws_url` in [`WsOptions`]. It serves every channel except +/// [`MEMPOOL_CHANNEL`], and every replay. +pub const DEFAULT_WS_URL: &str = "wss://api.0xarchive.io/ws"; + +/// The only endpoint that serves [`MEMPOOL_CHANNEL`], with the same API key +/// and protocol as [`DEFAULT_WS_URL`]. +/// +/// It serves a subset of the live channels; a channel it does not serve, and +/// every replay, is answered with a [`ServerMsg::Error`] that names +/// [`DEFAULT_WS_URL`]. [`MEMPOOL_CHANNEL`] has no replay on any endpoint. +/// Use one connection per endpoint: +/// +/// ```no_run +/// # use oxarchive::ws::{OxArchiveWs, WsOptions, MEMPOOL_CHANNEL, STREAM_WS_URL}; +/// # async fn example() -> oxarchive::Result<()> { +/// let mut ws = OxArchiveWs::new(WsOptions::new("your-api-key").ws_url(STREAM_WS_URL)); +/// ws.connect().await?; +/// ws.subscribe(MEMPOOL_CHANNEL, Some("BTC")).await?; +/// # Ok(()) +/// # } +/// ``` +pub const STREAM_WS_URL: &str = "wss://stream.0xarchive.io/ws"; + +/// The `mempool` channel: signed Hyperliquid transactions as our Hyperliquid +/// node receives them from its peers, before they are included in a block, +/// for every Hyperliquid product (perps, HIP-3, HIP-4 and spot). +/// +/// - Live only: there is no replay, history or REST route. +/// - Served only at [`STREAM_WS_URL`]. Elsewhere a subscribe is answered with +/// [`ErrorCode::EndpointUnsupported`]. +/// - Included with the Pro, Scale and Enterprise plans. On other plans a +/// subscribe is answered with [`ErrorCode::Forbidden`]. Every other channel +/// is on every plan. +/// +/// The symbol is optional on this channel only. `subscribe(MEMPOOL_CHANNEL, +/// None)` streams every pending transaction our Hyperliquid node receives; +/// with a symbol (`BTC`, `xyz:TSLA`, `HYPE-USDC`, `#49720`) it streams every +/// action that references that market, whole. The unfiltered stream is several megabytes a second +/// before compression, and the server limits unfiltered subscriptions: when +/// they are at capacity, a subscribe without a symbol is answered with +/// [`ErrorCode::RateLimited`]. Subscribe with a symbol where you can. +/// +/// [`OxArchiveWs`] delivers each message as a [`ServerMsg::Mempool`], whose +/// `coin` and `symbol` are the subscription's symbol, or `None` on the +/// unfiltered stream, and whose items keep each action's exact bytes. +pub const MEMPOOL_CHANNEL: &str = "mempool"; + /// Options for the WebSocket connection. +/// +/// `ws_url` defaults to [`DEFAULT_WS_URL`]. Set it to [`STREAM_WS_URL`] for +/// [`MEMPOOL_CHANNEL`]. pub struct WsOptions { pub api_key: String, pub ws_url: String, @@ -37,7 +87,7 @@ impl WsOptions { pub fn new(api_key: impl Into) -> Self { Self { api_key: api_key.into(), - ws_url: "wss://api.0xarchive.io/ws".to_string(), + ws_url: DEFAULT_WS_URL.to_string(), auto_reconnect: true, reconnect_delay: Duration::from_secs(1), max_reconnect_attempts: 10, @@ -371,7 +421,9 @@ pub enum ServerMsg { /// replay to resync) and `endpoint_unsupported` (this endpoint does not /// serve the channel or operation; the message names the one that does). /// A channel that does not offer a mode (live, replay, `replay.seek`) - /// answers `unsupported_for_venue`. + /// answers `unsupported_for_venue`, and a channel your plan does not + /// include ([`MEMPOOL_CHANNEL`] on plans other than Pro, Scale and + /// Enterprise) answers `forbidden`. Error { message: String, /// Stable machine-readable reason, `None` when the server sent none. @@ -382,12 +434,31 @@ pub enum ServerMsg { /// /// For the live Lighter channels of both deployments, /// [`ServerMsg::lighter_live_data`] decodes `data` into [`LighterLiveData`]. + /// A [`MEMPOOL_CHANNEL`] message read with [`ServerMsg::from_text`], as + /// [`OxArchiveWs`] reads every message, arrives as [`ServerMsg::Mempool`] + /// instead. Data { channel: String, coin: Option, symbol: Option, data: serde_json::Value, }, + /// A [`MEMPOOL_CHANNEL`] data message: one batch of pending transactions, + /// with one [`MempoolItem`] per signed action in the batch. + /// `coin` and `symbol` are the subscription's symbol, or `None` on the + /// unfiltered stream. + /// + /// [`ServerMsg::from_text`] (which [`OxArchiveWs`] uses) builds it + /// straight from the frame text, so each item's `action` keeps the exact + /// bytes the server sent. Deserializing a `ServerMsg` with serde gives a + /// [`ServerMsg::Data`] instead, and so does a mempool frame whose items do + /// not match [`MempoolItem`]. + #[serde(skip)] + Mempool { + coin: Option, + symbol: Option, + items: Vec, + }, /// Initial order-level L4 state for a live subscription or a replay. /// /// For an L4 replay (every channel in [`L4_REPLAY_CHANNELS`]), this @@ -542,6 +613,52 @@ impl ServerMsg { } } + /// Parse one text frame from the server, as [`OxArchiveWs`] does. + /// + /// The same as `serde_json::from_str::`, except that a + /// [`MEMPOOL_CHANNEL`] data message becomes a [`ServerMsg::Mempool`], + /// read from the text so that each action keeps its exact bytes. + /// + /// ``` + /// use oxarchive::ws::ServerMsg; + /// + /// let frame = r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":[ + /// {"received_at":"2026-10-08T01:57:23.548737209Z","received_at_ms":1791424643548, + /// "symbols":["BTC"],"action":{"type":"cancel","cancels":[{"a":0,"o":1}]}, + /// "nonce":1791424643400,"vault_address":null,"expires_after_ms":null, + /// "signature":{"r":"0x1","s":"0x2","v":27}}]}"#; + /// let msg = ServerMsg::from_text(frame).unwrap(); + /// let items = msg.mempool_items().unwrap(); + /// assert_eq!(items[0].symbols, ["BTC"]); + /// assert_eq!(items[0].action.get(), r#"{"type":"cancel","cancels":[{"a":0,"o":1}]}"#); + /// assert_eq!(items[0].action_value().unwrap()["type"], "cancel"); + /// ``` + pub fn from_text(text: &str) -> Result { + // Only a frame that names the channel can be one; skip the extra + // parse for every other frame. + if text.contains("\"mempool\"") { + if let Ok(frame) = serde_json::from_str::(text) { + if frame.kind == "data" && frame.channel == MEMPOOL_CHANNEL { + return Ok(ServerMsg::Mempool { + coin: frame.coin, + symbol: frame.symbol, + items: frame.data, + }); + } + } + } + serde_json::from_str(text).map_err(|e| Error::Deserialize(e.to_string())) + } + + /// The items of a [`ServerMsg::Mempool`], or `None` for every other + /// message. + pub fn mempool_items(&self) -> Option<&[MempoolItem]> { + match self { + ServerMsg::Mempool { items, .. } => Some(items), + _ => None, + } + } + /// The `error_code` of an [`ServerMsg::Error`], or `None` for other /// messages and for errors the server sent without a code. pub fn error_code(&self) -> Option<&ErrorCode> { @@ -552,6 +669,20 @@ impl ServerMsg { } } +/// A [`MEMPOOL_CHANNEL`] data frame, read directly from its text so that each +/// action keeps its exact bytes (see [`ServerMsg::from_text`]). +#[derive(Deserialize)] +struct MempoolFrame { + #[serde(rename = "type")] + kind: String, + channel: String, + #[serde(default)] + coin: Option, + #[serde(default)] + symbol: Option, + data: Vec, +} + // --------------------------------------------------------------------------- // WebSocket client // --------------------------------------------------------------------------- @@ -622,7 +753,7 @@ impl OxArchiveWs { while let Some(msg) = read.next().await { match msg { Ok(Message::Text(text)) => { - if let Ok(server_msg) = serde_json::from_str::(&text) { + if let Ok(server_msg) = ServerMsg::from_text(&text) { let _ = tx.send(server_msg); } } @@ -671,6 +802,10 @@ impl OxArchiveWs { /// support replay, not live subscriptions, and are rejected before a /// request is sent. Use REST for their current data or a bounded replay /// request for stored history. + /// + /// On [`MEMPOOL_CHANNEL`] the symbol is optional: `None` subscribes to + /// every pending transaction our Hyperliquid node receives. That channel + /// is served only at [`STREAM_WS_URL`]. pub async fn subscribe(&self, channel: &str, symbol: Option<&str>) -> Result<()> { if is_lighter_replay_only_channel(channel) { return Err(Error::InvalidParam(LIGHTER_SUBSCRIPTION_ERROR.to_string())); @@ -723,7 +858,8 @@ impl OxArchiveWs { .await } - /// Unsubscribe from a real-time channel. + /// Unsubscribe from a real-time channel. On [`MEMPOOL_CHANNEL`], `None` + /// ends the unfiltered subscription and a symbol ends that symbol's. pub async fn unsubscribe(&self, channel: &str, symbol: Option<&str>) -> Result<()> { self.send(ClientMsg::Unsubscribe { channel: channel.to_string(), @@ -748,8 +884,9 @@ impl OxArchiveWs { /// Which channels replay, and from when, is listed by /// `client.capabilities()`. A channel without replay (for example /// `ticker` or `spot_trades`) is answered with a - /// [`ServerMsg::Error`] whose `error_code` is `unsupported_for_venue`. A - /// successful replay terminates with a `replay_completed` server message. + /// [`ServerMsg::Error`] whose `error_code` is `unsupported_for_venue`; + /// [`MEMPOOL_CHANNEL`], which is live only, is answered with an error too. + /// A successful replay terminates with a `replay_completed` server message. pub async fn replay( &self, channel: &str, @@ -885,7 +1022,7 @@ impl OxArchiveWs { #[cfg(test)] mod lighter_live_tests { - use super::{ClientMsg, ServerMsg}; + use super::{ClientMsg, ServerMsg, MEMPOOL_CHANNEL}; use crate::types::LighterLiveData; const ORDERBOOK_FRAME: &str = r#"{"type":"data","channel":"lighter_orderbook","coin":"BTC","symbol":"BTC","data":{"coin":"BTC","time":1790294171459,"levels":[[{"px":"84368.7","sz":"0.00020","n":1},{"px":"84368.6","sz":"0.00020","n":1},{"px":"84368.3","sz":"0.00010","n":1}],[{"px":"84368.8","sz":"0.05720","n":1},{"px":"84368.9","sz":"0.14223","n":1},{"px":"84369.1","sz":"0.01198","n":1}]]}}"#; @@ -1070,6 +1207,16 @@ mod lighter_live_tests { serde_json::to_value(&unsubscribe).unwrap(), serde_json::json!({"op": "unsubscribe", "channel": "lighter_trades", "symbol": "BTC"}) ); + + // `mempool` is the one channel whose symbol is optional. + let unfiltered = ClientMsg::Subscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: None, + }; + assert_eq!( + serde_json::to_value(&unfiltered).unwrap(), + serde_json::json!({"op": "subscribe", "channel": "mempool", "symbol": null}) + ); } #[test] diff --git a/src/ws/mod.rs b/src/ws/mod.rs index 276ed1c..72967e8 100644 --- a/src/ws/mod.rs +++ b/src/ws/mod.rs @@ -4,7 +4,7 @@ //! //! Requires the `websocket` feature: //! ```toml -//! oxarchive = { version = "1.12", features = ["websocket"] } +//! oxarchive = { version = "1.13", features = ["websocket"] } //! ``` //! //! For large historical downloads, use the S3 Parquet bulk export at diff --git a/tests/api_contract.rs b/tests/api_contract.rs index 1ff1aa2..d0d1d64 100644 --- a/tests/api_contract.rs +++ b/tests/api_contract.rs @@ -858,6 +858,46 @@ async fn capabilities_are_typed_rows() { assert!(Capability::for_channel(&rows, "nope").is_none()); } +#[test] +fn only_the_mempool_row_names_an_endpoint_and_plans() { + let rows: Vec = serde_json::from_value(json!([ + {"venue": "hyperliquid", "datatype": "mempool", "rest_routes": [], + "ws_channels": ["mempool"], "live": true, "replay": false, + "available_from": null, "cadence": "event", "page_limit": null, "intervals": [], + "notes": "Live only, on wss://stream.0xarchive.io/ws.", + "ws_endpoint": "wss://stream.0xarchive.io/ws", + "plans": ["pro", "scale", "enterprise"]}, + {"venue": "hyperliquid", "datatype": "trades", + "rest_routes": ["/v1/hyperliquid/trades/{symbol}"], "ws_channels": ["trades"], + "live": true, "replay": true, "available_from": "2023-04-15T03:31:00.000Z", + "cadence": "event", "page_limit": 1000, "intervals": [], "notes": null} + ])) + .unwrap(); + + let mempool = Capability::for_channel(&rows, "mempool").unwrap(); + assert!(mempool.live && !mempool.replay); + assert_eq!( + mempool.ws_endpoint.as_deref(), + Some("wss://stream.0xarchive.io/ws") + ); + assert_eq!( + mempool.plans.as_deref(), + Some( + &[ + "pro".to_string(), + "scale".to_string(), + "enterprise".to_string() + ][..] + ) + ); + + let trades = Capability::for_channel(&rows, "trades").unwrap(); + assert_eq!((&trades.ws_endpoint, &trades.plans), (&None, &None)); + // Rows without them serialize as before: the fields are left out. + let value = serde_json::to_value(trades).unwrap(); + assert!(value.get("ws_endpoint").is_none() && value.get("plans").is_none()); +} + // --------------------------------------------------------------------------- // Filters the API honours // --------------------------------------------------------------------------- diff --git a/tests/mempool.rs b/tests/mempool.rs new file mode 100644 index 0000000..4d7ea28 --- /dev/null +++ b/tests/mempool.rs @@ -0,0 +1,261 @@ +#![cfg(feature = "websocket")] +//! The `mempool` WebSocket channel: pending Hyperliquid transactions. + +use oxarchive::error::Error; +use oxarchive::ws::{ + ClientMsg, OxArchiveWs, ServerMsg, WsOptions, DEFAULT_WS_URL, MEMPOOL_CHANNEL, STREAM_WS_URL, +}; +use oxarchive::{ErrorCode, LighterLiveData, MempoolItem, MempoolSignature}; +use serde_json::json; + +/// A frame shaped as the server sends it: an order on BTC and a transfer that +/// references no market, on the unfiltered stream. +const UNFILTERED_FRAME: &str = r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":[ +{"received_at":"2026-10-08T01:57:23.548737209Z","received_at_ms":1791424643548,"symbols":["BTC"],"action":{"type":"order","orders":[{"a":0,"b":true,"p":"83276","s":"0.40011","r":false,"t":{"limit":{"tif":"Alo"}},"c":"0x7849acc2c6c2f6f0fe4bc80ef13d1504"}],"grouping":"na"},"nonce":1791424643400,"vault_address":null,"expires_after_ms":null,"signature":{"r":"0x5afc","s":"0x57e2","v":28}}, +{"received_at":"2026-10-08T01:57:23.548737209Z","received_at_ms":1791424643548,"symbols":[],"action":{"type":"usdSend","signatureChainId":"0xa4b1","hyperliquidChain":"Mainnet","destination":"0x0000000000000000000000000000000000000001","amount":"10","time":1791424643400},"nonce":1791424643400,"vault_address":null,"expires_after_ms":1791424703400,"signature":{"r":"0x01","s":"0x02","v":27}} +]}"#; + +#[test] +fn mempool_is_served_at_the_stream_endpoint_only() { + assert_eq!(MEMPOOL_CHANNEL, "mempool"); + assert_eq!(STREAM_WS_URL, "wss://stream.0xarchive.io/ws"); + assert_eq!(DEFAULT_WS_URL, "wss://api.0xarchive.io/ws"); + assert_eq!(WsOptions::new("k").ws_url, DEFAULT_WS_URL); + assert_eq!( + WsOptions::new("k").ws_url(STREAM_WS_URL).ws_url, + STREAM_WS_URL + ); +} + +#[test] +fn subscribe_without_a_symbol_is_the_unfiltered_stream() { + let unfiltered = ClientMsg::Subscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: None, + }; + assert_eq!( + serde_json::to_value(&unfiltered).unwrap(), + json!({"op": "subscribe", "channel": "mempool", "symbol": null}) + ); + + for symbol in ["BTC", "xyz:TSLA", "HYPE-USDC", "#49720"] { + let filtered = ClientMsg::Subscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: Some(symbol.to_string()), + }; + assert_eq!( + serde_json::to_value(&filtered).unwrap(), + json!({"op": "subscribe", "channel": "mempool", "symbol": symbol}) + ); + } + + let unsubscribe = ClientMsg::Unsubscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: None, + }; + assert_eq!( + serde_json::to_value(&unsubscribe).unwrap(), + json!({"op": "unsubscribe", "channel": "mempool", "symbol": null}) + ); +} + +#[tokio::test] +async fn mempool_subscriptions_are_left_to_the_server() { + // Endpoint and plan are the server's to decide: nothing is refused here. + let ws = OxArchiveWs::new(WsOptions::new("test-key").ws_url(STREAM_WS_URL)); + ws.subscribe(MEMPOOL_CHANNEL, None).await.unwrap(); + ws.subscribe(MEMPOOL_CHANNEL, Some("BTC")).await.unwrap(); + ws.unsubscribe(MEMPOOL_CHANNEL, Some("BTC")).await.unwrap(); + ws.unsubscribe(MEMPOOL_CHANNEL, None).await.unwrap(); +} + +/// The order action exactly as it appears in `UNFILTERED_FRAME`. +const ORDER_ACTION: &str = r#"{"type":"order","orders":[{"a":0,"b":true,"p":"83276","s":"0.40011","r":false,"t":{"limit":{"tif":"Alo"}},"c":"0x7849acc2c6c2f6f0fe4bc80ef13d1504"}],"grouping":"na"}"#; + +fn mempool_items(frame: &str) -> (Option, Option, Vec) { + match ServerMsg::from_text(frame).unwrap() { + ServerMsg::Mempool { + coin, + symbol, + items, + } => (coin, symbol, items), + other => panic!("expected a mempool message, got {other:?}"), + } +} + +#[test] +fn unfiltered_frame_decodes_to_items() { + let msg = ServerMsg::from_text(UNFILTERED_FRAME).unwrap(); + assert!(msg.lighter_live_data().is_none()); + assert_eq!(msg.mempool_items().map(<[MempoolItem]>::len), Some(2)); + + let (coin, symbol, items) = mempool_items(UNFILTERED_FRAME); + assert_eq!((coin, symbol), (None, None)); + assert_eq!(items.len(), 2); + + let order = &items[0]; + assert_eq!( + order.received_at.as_deref(), + Some("2026-10-08T01:57:23.548737209Z") + ); + assert_eq!(order.received_at_ms, Some(1791424643548)); + assert_eq!(order.symbols, ["BTC"]); + let action = order.action_value().unwrap(); + assert_eq!(action["type"], "order"); + assert_eq!(action["orders"][0]["a"], 0); + assert_eq!(action["orders"][0]["p"], "83276"); + assert_eq!(order.nonce, Some(1791424643400)); + assert_eq!(order.vault_address, None); + assert_eq!(order.expires_after_ms, None); + assert_eq!( + order.signature, + Some(MempoolSignature { + r: "0x5afc".to_string(), + s: "0x57e2".to_string(), + v: 28, + }) + ); + + let transfer = &items[1]; + assert!(transfer.symbols.is_empty()); + assert_eq!(transfer.action_value().unwrap()["type"], "usdSend"); + assert_eq!(transfer.expires_after_ms, Some(1791424703400)); +} + +#[test] +fn the_action_keeps_its_exact_bytes_and_key_order() { + let (_, _, items) = mempool_items(UNFILTERED_FRAME); + + // Byte for byte as sent, keys in signing order (not alphabetical). + assert_eq!(items[0].action.get(), ORDER_ACTION); + assert!(items[1] + .action + .get() + .starts_with(r#"{"type":"usdSend","signatureChainId":"0xa4b1","#)); + + // Serializing an item writes the action back unchanged. + let item = serde_json::to_string(&items[0]).unwrap(); + assert!( + item.contains(&format!(r#""action":{ORDER_ACTION}"#)), + "{item}" + ); + + // The parsed form is for inspection and holds the same content. + let parsed: serde_json::Value = serde_json::from_str(ORDER_ACTION).unwrap(); + assert_eq!(items[0].action_value().unwrap(), parsed); +} + +#[test] +fn symbol_frame_carries_the_subscription_symbol() { + let frame = UNFILTERED_FRAME.replace( + r#""coin":null,"symbol":null"#, + r#""coin":"BTC","symbol":"BTC""#, + ); + let (coin, symbol, items) = mempool_items(&frame); + assert_eq!(coin.as_deref(), Some("BTC")); + assert_eq!(symbol.as_deref(), Some("BTC")); + assert_eq!(items.len(), 2); +} + +#[test] +fn other_frames_parse_as_before() { + let trades = r#"{"type":"data","channel":"trades","coin":"BTC","symbol":"BTC","data":[]}"#; + let msg = ServerMsg::from_text(trades).unwrap(); + assert!(matches!(&msg, ServerMsg::Data { channel, .. } if channel == "trades")); + assert!(msg.mempool_items().is_none()); + + let pong = ServerMsg::from_text(r#"{"type":"pong"}"#).unwrap(); + assert!(matches!(pong, ServerMsg::Pong)); + assert!(pong.mempool_items().is_none()); + + assert!(LighterLiveData::decode("mempool", &json!([])).is_none()); + assert!(matches!( + ServerMsg::from_text("not json"), + Err(Error::Deserialize(_)) + )); +} + +#[test] +fn plain_serde_and_unexpected_items_fall_back_to_data() { + // Deserializing with serde alone cannot keep the raw bytes: it gives Data. + let msg: ServerMsg = serde_json::from_str(UNFILTERED_FRAME).unwrap(); + assert!(matches!(&msg, ServerMsg::Data { channel, .. } if channel == "mempool")); + assert!(msg.mempool_items().is_none()); + + // A mempool frame whose items do not match is delivered unchanged. + let odd = + r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":{"not":"a list"}}"#; + let msg = ServerMsg::from_text(odd).unwrap(); + assert!(matches!(&msg, ServerMsg::Data { data, .. } if data["not"] == "a list")); +} + +#[test] +fn items_tolerate_nulls_new_fields_and_new_action_types() { + let frame = r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":[ +{"received_at":null,"received_at_ms":null,"symbols":[],"action":{"x":1,"type":"someFutureAction"},"nonce":null,"vault_address":"0x00000000000000000000000000000000000000aa","expires_after_ms":null,"signature":null,"a_future_field":true}, +{"action":{"type":"noop"}} +]}"#; + let (_, _, items) = mempool_items(frame); + assert_eq!(items[0].received_at, None); + assert_eq!(items[0].nonce, None); + assert_eq!(items[0].signature, None); + assert_eq!( + items[0].action.get(), + r#"{"x":1,"type":"someFutureAction"}"# + ); + assert_eq!( + items[0].vault_address.as_deref(), + Some("0x00000000000000000000000000000000000000aa") + ); + assert!(items[1].symbols.is_empty()); + assert_eq!(items[1].action_value().unwrap(), json!({"type": "noop"})); +} + +#[test] +fn subscribe_acks_without_a_symbol_parse() { + let ack: ServerMsg = serde_json::from_str( + r#"{"type":"subscribed","channel":"mempool","coin":null,"symbol":null}"#, + ) + .unwrap(); + match ack { + ServerMsg::Subscribed { + channel, + coin, + symbol, + } => { + assert_eq!(channel, "mempool"); + assert_eq!((coin, symbol), (None, None)); + } + other => panic!("expected a subscribed message, got {other:?}"), + } +} + +#[test] +fn mempool_refusals_carry_their_error_code() { + for (frame, expected) in [ + ( + r#"{"type":"error","message":"The mempool channel is included with the Pro, Scale and Enterprise plans. Upgrade at https://0xarchive.io/pricing.","error_code":"forbidden"}"#, + ErrorCode::Forbidden, + ), + ( + r#"{"type":"error","message":"The 'mempool' channel is live only and served on wss://stream.0xarchive.io/ws. Subscribe to it there.","error_code":"endpoint_unsupported"}"#, + ErrorCode::EndpointUnsupported, + ), + ( + r#"{"type":"error","message":"The unfiltered mempool stream is at capacity. Subscribe with a symbol, or try again later.","error_code":"rate_limited"}"#, + ErrorCode::RateLimited, + ), + ( + r#"{"type":"error","message":"The mempool channel is temporarily unavailable. Please try again shortly.","error_code":"upstream_unavailable"}"#, + ErrorCode::UpstreamUnavailable, + ), + ( + r#"{"type":"error","message":"Unknown symbol 'NOPE'.","error_code":"invalid_symbol"}"#, + ErrorCode::InvalidSymbol, + ), + ] { + let msg: ServerMsg = serde_json::from_str(frame).unwrap(); + assert_eq!(msg.error_code(), Some(&expected), "{frame}"); + assert!(msg.mempool_items().is_none()); + } +} diff --git a/tests/websocket_capability.rs b/tests/websocket_capability.rs index 4db77e4..20bda11 100644 --- a/tests/websocket_capability.rs +++ b/tests/websocket_capability.rs @@ -540,7 +540,8 @@ fn lighter_replay_rows_decode_with_the_live_shapes() { assert!(msg.lighter_live_data().is_none()); } -/// Channel rows of `GET /v1/capabilities` as served on 2026-10-04. +/// Channel rows of `GET /v1/capabilities` as served on 2026-10-04, plus the +/// `mempool` row. const CAPABILITY_CHANNELS: &str = r#"[ {"venue":"hyperliquid","datatype":"l2_orderbook","ws_channels":["orderbook"],"live":true,"replay":true,"available_from":"2023-04-15T00:00:00.000Z"}, {"venue":"hyperliquid","datatype":"l2_full_depth","ws_channels":["orderbook_full"],"live":true,"replay":true,"available_from":"2026-03-11T01:03:00.000Z"}, @@ -552,6 +553,7 @@ const CAPABILITY_CHANNELS: &str = r#"[ {"venue":"hyperliquid","datatype":"oi","ws_channels":["open_interest"],"live":true,"replay":true,"available_from":"2023-05-20T02:50:00.000Z"}, {"venue":"hyperliquid","datatype":"liquidations","ws_channels":["liquidations"],"live":true,"replay":true,"available_from":"2025-12-22T00:00:00.000Z"}, {"venue":"hyperliquid","datatype":"ticker","ws_channels":["ticker","all_tickers"],"live":true,"replay":false,"available_from":null}, +{"venue":"hyperliquid","datatype":"mempool","ws_channels":["mempool"],"live":true,"replay":false,"available_from":null,"ws_endpoint":"wss://stream.0xarchive.io/ws","plans":["pro","scale","enterprise"]}, {"venue":"hip3","datatype":"l2_orderbook","ws_channels":["hip3_orderbook"],"live":true,"replay":true,"available_from":"2026-02-16T16:57:00.000Z"}, {"venue":"hip3","datatype":"l2_full_depth","ws_channels":["hip3_orderbook_full"],"live":true,"replay":true,"available_from":"2026-03-11T01:03:00.000Z"}, {"venue":"hip3","datatype":"l4_diffs","ws_channels":["hip3_l4_diffs"],"live":true,"replay":true,"available_from":"2026-03-11T01:03:00.000Z"},